hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/AzureFileSystemThreadPoolExecutor.java - hadoop - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.hadoop.fs.azure;

 import java.io.IOException;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;

 import org.apache.hadoop.util.Time;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.google.common.annotations.VisibleForTesting;

 class AzureFileSystemThreadPoolExecutor {

   public static final Logger LOG = LoggerFactory.getLogger(AzureFileSystemThreadPoolExecutor.class);

   /*
    * Number of threads to keep in the pool.
    */
   private int threadCount;

   /*
    * Prefix to be used for naming threads.
    */
   private String threadNamePrefix;

   /*
    * File system operation like delete/rename. Used for logging purpose.
    */
   private String operation;

   /*
    * Source blob key used for file operation. Used for logging purpose.
    */
   private String key;

   /*
    * Configuration name for recommendations. Used for logging purpose.
    */
   private String config;

   /**
    * Creates a new AzureFileSystemThreadPoolExecutor object.
    *
    * @param threadCount
    *        Number of threads to be used after reading user configuration.
    * @param threadNamePrefix
    *        Prefix to be used to name threads for the file operation.
    * @param operation
    *        File system operation like delete/rename. Used for logging purpose.
    * @param key
    *        Source blob key used for file operation. Used for logging purpose.
    * @param config
    *        Configuration name for recommendations. Used for logging purpose.
    */
   public AzureFileSystemThreadPoolExecutor(int threadCount, String threadNamePrefix,
       String operation, String key, String config) {
     this.threadCount = threadCount;
     this.threadNamePrefix = threadNamePrefix;
     this.operation = operation;
     this.key = key;
     this.config = config;
   }

   /**
    * Gets a new thread pool
    * @param threadCount
    *        Number of threads to keep in the pool.
    * @param threadNamePrefix
    *        Prefix to be used for naming threads.
    *
    * @return
    *        Returns a new thread pool.
    */
   @VisibleForTesting
   ThreadPoolExecutor getThreadPool(int threadCount) throws Exception {
     return new ThreadPoolExecutor(threadCount, threadCount, 2, TimeUnit.SECONDS,
         new LinkedBlockingQueue<Runnable>(), new AzureFileSystemThreadFactory(this.threadNamePrefix));
   }

   /**
    * Execute the file operation parallel using threads. All threads works on a
    * single working set of files stored in input 'contents'. The synchronization
    * between multiple threads is achieved through retrieving atomic index value
    * from the array. Once thread gets the index, it retrieves the file and initiates
    * the file operation. The advantage with this method is that file operations
    * doesn't get serialized due to any thread. Also, the input copy is not changed
    * such that caller can reuse the list for other purposes.
    *
    * This implementation also considers that failure of operation on single file
    * is considered as overall operation failure. All threads bail out their execution
    * as soon as they detect any single thread either got exception or operation is failed.
    *
    * @param contents
    *        List of blobs on which operation to be done.
    * @param threadOperation
    *        The actual operation to be executed by each thread on a file.
    *
    * @param operationStatus
    *        Returns true if the operation is success, false if operation is failed.
    * @throws IOException
    *
    */
   boolean executeParallel(FileMetadata[] contents, AzureFileSystemThreadTask threadOperation) throws IOException {

     boolean operationStatus = false;
     boolean threadsEnabled = false;
     int threadCount = this.threadCount;
     ThreadPoolExecutor ioThreadPool = null;

     // Start time for file operation
     long start = Time.monotonicNow();

     // If number of files  are less then reduce threads to file count.
     threadCount = Math.min(contents.length, threadCount);

     if (threadCount > 1) {
       try {
         ioThreadPool = getThreadPool(threadCount);
         threadsEnabled = true;
       } catch(Exception e) {
         // The possibility of this scenario is very remote. Added this code as safety net.
         LOG.warn("Failed to create thread pool with threads {} for operation {} on blob {}."
             + " Use config {} to set less number of threads. Setting config value to <= 1 will disable threads.",
             threadCount, operation, key, config);
       }
     } else {
       LOG.warn("Disabling threads for {} operation as thread count {} is <= 1", operation, threadCount);
     }

     if (threadsEnabled) {
       LOG.debug("Using thread pool for {} operation with threads {}", operation, threadCount);
       boolean started = false;
       AzureFileSystemThreadRunnable runnable = new AzureFileSystemThreadRunnable(contents, threadOperation, operation);

       // Don't start any new requests if there is an exception from any one thread.
       for (int i = 0; i < threadCount && runnable.lastException == null && runnable.operationStatus; i++)
       {
         try {
           ioThreadPool.execute(runnable);
           started = true;
         } catch (RejectedExecutionException ex) {
           // If threads can't be scheduled then report error and move ahead with next thread.
           // Don't fail operation due to this issue.
           LOG.error("Rejected execution of thread for {} operation on blob {}."
               + " Continuing with existing threads. Use config {} to set less number of threads"
               + " to avoid this error", operation, key, config);
         }
       }

       // Stop accepting any new execute requests.
       ioThreadPool.shutdown();

       try {
         // Wait for threads to terminate. Keep time out as large value
         ioThreadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
       } catch(InterruptedException intrEx) {
         // If current thread got interrupted then shutdown all threads now.
         ioThreadPool.shutdownNow();

         // Restore the interrupted status
         Thread.currentThread().interrupt();
         LOG.error("Threads got interrupted {} blob operation for {} "
             , operation, key);
       }

       int threadsNotUsed = threadCount - runnable.threadsUsed.get();
       if (threadsNotUsed > 0) {
         LOG.warn("{} threads not used for {} operation on blob {}", threadsNotUsed, operation, key);
       }

       if (!started) {
         // No threads started. Fall back to serial mode.
         threadsEnabled = false;
         LOG.info("Not able to schedule threads to {} blob {}. Fall back to {} blob serially."
             , operation, key, operation);
       } else {
         IOException lastException = runnable.lastException;

         // There are no exceptions from threads and no operation failures. Consider this scenario
         // as failure only if file operations are not done on all files.
         if (lastException == null && runnable.operationStatus && runnable.filesProcessed.get() < contents.length) {
           LOG.error("{} failed as operation on subfolders and files failed.", operation);
           lastException = new IOException(operation + " failed as operation on subfolders and files failed.");
         }

         if (lastException != null) {
           // Threads started and executed. One or more threads seems to have hit exception.
           // Raise the same exception.
           throw lastException;
         }

         operationStatus = runnable.operationStatus;
       }
     }

     if (!threadsEnabled) {
       // No threads. Serialize the operation. Clear any last exceptions.
       LOG.debug("Serializing the {} operation", operation);
       for (int i = 0; i < contents.length; i++) {
         if (!threadOperation.execute(contents[i])) {
           LOG.warn("Failed to {} file {}", operation, contents[i]);
           return false;
         }
       }

       // Operation is success
       operationStatus = true;
     }

     // Find the duration of time taken for file operation
     long end = Time.monotonicNow();
     LOG.info("Time taken for {} operation is: {} ms with threads: {}", operation, (end - start), threadCount);

     return operationStatus;
   }

   /**
    * A ThreadFactory for Azure File operation threads with meaningful names helpful
    * for debugging purposes.
    */
   static class AzureFileSystemThreadFactory implements ThreadFactory {

     private String threadIdPrefix = "AzureFileSystemThread";

     /**
      * Atomic integer to provide thread id for thread names.
      */
     private AtomicInteger threadSequenceNumber = new AtomicInteger(0);

     public AzureFileSystemThreadFactory(String prefix) {
       threadIdPrefix = prefix;
     }

     @Override
     public Thread newThread(Runnable r) {
       Thread t = new Thread(r);

       // Use current thread name as part in naming thread such that use of
       // same file system object will have unique names.
       t.setName(String.format("%s-%s-%d", threadIdPrefix, Thread.currentThread().getName(),
           threadSequenceNumber.getAndIncrement()));
       return t;
     }

   }

   static class AzureFileSystemThreadRunnable implements Runnable {

     // Tracks if any thread has raised exception.
     private volatile IOException lastException = null;

     // Tracks if any thread has failed execution.
     private volatile boolean operationStatus = true;

     // Atomic tracker to retrieve index of next file to be processed
     private AtomicInteger fileIndex = new AtomicInteger(0);

     // Atomic tracker to count number of files successfully processed
     private AtomicInteger filesProcessed = new AtomicInteger(0);

     // Atomic tracker to retrieve number of threads used to do at least one file operation.
     private AtomicInteger threadsUsed = new AtomicInteger(0);

     // Type of file system operation
     private String operation = "Unknown";

     // List of files to be processed.
     private final FileMetadata[] files;

     // Thread task which encapsulates the file system operation work on a file.
     private AzureFileSystemThreadTask task;

     public AzureFileSystemThreadRunnable(final FileMetadata[] files,
         AzureFileSystemThreadTask task, String operation) {
       this.operation = operation;
       this.files = files;
       this.task = task;
     }

     @Override
     public void run() {
       long start = Time.monotonicNow();
       int currentIndex;
       int processedFilesCount = 0;

       while ((currentIndex = fileIndex.getAndIncrement()) < files.length) {
         processedFilesCount++;
         FileMetadata file = files[currentIndex];

         try {
           // Execute the file operation.
           if (!task.execute(file)) {
             LOG.error("{} operation failed for file {}",
                 this.operation, file.getKey());
             operationStatus = false;
           } else {
             filesProcessed.getAndIncrement();
           }
         } catch (Exception e) {
           LOG.error("Encountered Exception for {} operation for file {}",
               this.operation, file.getKey());
           lastException = new IOException("Encountered Exception for "
               + this.operation + " operation for file " + file.getKey(), e);
         }

         // If any thread has seen exception or operation failed then we
         // don't have to process further.
         if (lastException != null || !operationStatus) {
           LOG.warn("Terminating execution of {} operation now as some other thread"
               + " already got exception or operation failed", this.operation, file.getKey());
           break;
         }
       }

       long end = Time.monotonicNow();
       LOG.debug("Time taken to process {} files count for {} operation: {} ms",
           processedFilesCount, this.operation, (end - start));
       if (processedFilesCount > 0) {
         threadsUsed.getAndIncrement();
       }
     }
   }
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.hadoop.fs.azure;

	import java.io.IOException;
	import java.util.concurrent.LinkedBlockingQueue;
	import java.util.concurrent.RejectedExecutionException;
	import java.util.concurrent.ThreadFactory;
	import java.util.concurrent.ThreadPoolExecutor;
	import java.util.concurrent.TimeUnit;
	import java.util.concurrent.atomic.AtomicInteger;

	import org.apache.hadoop.util.Time;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.google.common.annotations.VisibleForTesting;

	class AzureFileSystemThreadPoolExecutor {

	public static final Logger LOG = LoggerFactory.getLogger(AzureFileSystemThreadPoolExecutor.class);

	/*
	* Number of threads to keep in the pool.
	*/
	private int threadCount;

	/*
	* Prefix to be used for naming threads.
	*/
	private String threadNamePrefix;

	/*
	* File system operation like delete/rename. Used for logging purpose.
	*/
	private String operation;

	/*
	* Source blob key used for file operation. Used for logging purpose.
	*/
	private String key;

	/*
	* Configuration name for recommendations. Used for logging purpose.
	*/
	private String config;

	/**
	* Creates a new AzureFileSystemThreadPoolExecutor object.
	*
	* @param threadCount
	* Number of threads to be used after reading user configuration.
	* @param threadNamePrefix
	* Prefix to be used to name threads for the file operation.
	* @param operation
	* File system operation like delete/rename. Used for logging purpose.
	* @param key
	* Source blob key used for file operation. Used for logging purpose.
	* @param config
	* Configuration name for recommendations. Used for logging purpose.
	*/
	public AzureFileSystemThreadPoolExecutor(int threadCount, String threadNamePrefix,
	String operation, String key, String config) {
	this.threadCount = threadCount;
	this.threadNamePrefix = threadNamePrefix;
	this.operation = operation;
	this.key = key;
	this.config = config;
	}

	/**
	* Gets a new thread pool
	* @param threadCount
	* Number of threads to keep in the pool.
	* @param threadNamePrefix
	* Prefix to be used for naming threads.
	*
	* @return
	* Returns a new thread pool.
	*/
	@VisibleForTesting
	ThreadPoolExecutor getThreadPool(int threadCount) throws Exception {
	return new ThreadPoolExecutor(threadCount, threadCount, 2, TimeUnit.SECONDS,
	new LinkedBlockingQueue<Runnable>(), new AzureFileSystemThreadFactory(this.threadNamePrefix));
	}

	/**
	* Execute the file operation parallel using threads. All threads works on a
	* single working set of files stored in input 'contents'. The synchronization
	* between multiple threads is achieved through retrieving atomic index value
	* from the array. Once thread gets the index, it retrieves the file and initiates
	* the file operation. The advantage with this method is that file operations
	* doesn't get serialized due to any thread. Also, the input copy is not changed
	* such that caller can reuse the list for other purposes.
	*
	* This implementation also considers that failure of operation on single file
	* is considered as overall operation failure. All threads bail out their execution
	* as soon as they detect any single thread either got exception or operation is failed.
	*
	* @param contents
	* List of blobs on which operation to be done.
	* @param threadOperation
	* The actual operation to be executed by each thread on a file.
	*
	* @param operationStatus
	* Returns true if the operation is success, false if operation is failed.
	* @throws IOException
	*
	*/
	boolean executeParallel(FileMetadata[] contents, AzureFileSystemThreadTask threadOperation) throws IOException {

	boolean operationStatus = false;
	boolean threadsEnabled = false;
	int threadCount = this.threadCount;
	ThreadPoolExecutor ioThreadPool = null;

	// Start time for file operation
	long start = Time.monotonicNow();

	// If number of files are less then reduce threads to file count.
	threadCount = Math.min(contents.length, threadCount);

	if (threadCount > 1) {
	try {
	ioThreadPool = getThreadPool(threadCount);
	threadsEnabled = true;
	} catch(Exception e) {
	// The possibility of this scenario is very remote. Added this code as safety net.
	LOG.warn("Failed to create thread pool with threads {} for operation {} on blob {}."
	+ " Use config {} to set less number of threads. Setting config value to <= 1 will disable threads.",
	threadCount, operation, key, config);
	}
	} else {
	LOG.warn("Disabling threads for {} operation as thread count {} is <= 1", operation, threadCount);
	}

	if (threadsEnabled) {
	LOG.debug("Using thread pool for {} operation with threads {}", operation, threadCount);
	boolean started = false;
	AzureFileSystemThreadRunnable runnable = new AzureFileSystemThreadRunnable(contents, threadOperation, operation);

	// Don't start any new requests if there is an exception from any one thread.
	for (int i = 0; i < threadCount && runnable.lastException == null && runnable.operationStatus; i++)
	{
	try {
	ioThreadPool.execute(runnable);
	started = true;
	} catch (RejectedExecutionException ex) {
	// If threads can't be scheduled then report error and move ahead with next thread.
	// Don't fail operation due to this issue.
	LOG.error("Rejected execution of thread for {} operation on blob {}."
	+ " Continuing with existing threads. Use config {} to set less number of threads"
	+ " to avoid this error", operation, key, config);
	}
	}

	// Stop accepting any new execute requests.
	ioThreadPool.shutdown();

	try {
	// Wait for threads to terminate. Keep time out as large value
	ioThreadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
	} catch(InterruptedException intrEx) {
	// If current thread got interrupted then shutdown all threads now.
	ioThreadPool.shutdownNow();

	// Restore the interrupted status
	Thread.currentThread().interrupt();
	LOG.error("Threads got interrupted {} blob operation for {} "
	, operation, key);
	}

	int threadsNotUsed = threadCount - runnable.threadsUsed.get();
	if (threadsNotUsed > 0) {
	LOG.warn("{} threads not used for {} operation on blob {}", threadsNotUsed, operation, key);
	}

	if (!started) {
	// No threads started. Fall back to serial mode.
	threadsEnabled = false;
	LOG.info("Not able to schedule threads to {} blob {}. Fall back to {} blob serially."
	, operation, key, operation);
	} else {
	IOException lastException = runnable.lastException;

	// There are no exceptions from threads and no operation failures. Consider this scenario
	// as failure only if file operations are not done on all files.
	if (lastException == null && runnable.operationStatus && runnable.filesProcessed.get() < contents.length) {
	LOG.error("{} failed as operation on subfolders and files failed.", operation);
	lastException = new IOException(operation + " failed as operation on subfolders and files failed.");
	}

	if (lastException != null) {
	// Threads started and executed. One or more threads seems to have hit exception.
	// Raise the same exception.
	throw lastException;
	}

	operationStatus = runnable.operationStatus;
	}
	}

	if (!threadsEnabled) {
	// No threads. Serialize the operation. Clear any last exceptions.
	LOG.debug("Serializing the {} operation", operation);
	for (int i = 0; i < contents.length; i++) {
	if (!threadOperation.execute(contents[i])) {
	LOG.warn("Failed to {} file {}", operation, contents[i]);
	return false;
	}
	}

	// Operation is success
	operationStatus = true;
	}

	// Find the duration of time taken for file operation
	long end = Time.monotonicNow();
	LOG.info("Time taken for {} operation is: {} ms with threads: {}", operation, (end - start), threadCount);

	return operationStatus;
	}

	/**
	* A ThreadFactory for Azure File operation threads with meaningful names helpful
	* for debugging purposes.
	*/
	static class AzureFileSystemThreadFactory implements ThreadFactory {

	private String threadIdPrefix = "AzureFileSystemThread";

	/**
	* Atomic integer to provide thread id for thread names.
	*/
	private AtomicInteger threadSequenceNumber = new AtomicInteger(0);

	public AzureFileSystemThreadFactory(String prefix) {
	threadIdPrefix = prefix;
	}

	@Override
	public Thread newThread(Runnable r) {
	Thread t = new Thread(r);

	// Use current thread name as part in naming thread such that use of
	// same file system object will have unique names.
	t.setName(String.format("%s-%s-%d", threadIdPrefix, Thread.currentThread().getName(),
	threadSequenceNumber.getAndIncrement()));
	return t;
	}

	}

	static class AzureFileSystemThreadRunnable implements Runnable {

	// Tracks if any thread has raised exception.
	private volatile IOException lastException = null;

	// Tracks if any thread has failed execution.
	private volatile boolean operationStatus = true;

	// Atomic tracker to retrieve index of next file to be processed
	private AtomicInteger fileIndex = new AtomicInteger(0);

	// Atomic tracker to count number of files successfully processed
	private AtomicInteger filesProcessed = new AtomicInteger(0);

	// Atomic tracker to retrieve number of threads used to do at least one file operation.
	private AtomicInteger threadsUsed = new AtomicInteger(0);

	// Type of file system operation
	private String operation = "Unknown";

	// List of files to be processed.
	private final FileMetadata[] files;

	// Thread task which encapsulates the file system operation work on a file.
	private AzureFileSystemThreadTask task;

	public AzureFileSystemThreadRunnable(final FileMetadata[] files,
	AzureFileSystemThreadTask task, String operation) {
	this.operation = operation;
	this.files = files;
	this.task = task;
	}

	@Override
	public void run() {
	long start = Time.monotonicNow();
	int currentIndex;
	int processedFilesCount = 0;

	while ((currentIndex = fileIndex.getAndIncrement()) < files.length) {
	processedFilesCount++;
	FileMetadata file = files[currentIndex];

	try {
	// Execute the file operation.
	if (!task.execute(file)) {
	LOG.error("{} operation failed for file {}",
	this.operation, file.getKey());
	operationStatus = false;
	} else {
	filesProcessed.getAndIncrement();
	}
	} catch (Exception e) {
	LOG.error("Encountered Exception for {} operation for file {}",
	this.operation, file.getKey());
	lastException = new IOException("Encountered Exception for "
	+ this.operation + " operation for file " + file.getKey(), e);
	}

	// If any thread has seen exception or operation failed then we
	// don't have to process further.
	if (lastException != null \|\| !operationStatus) {
	LOG.warn("Terminating execution of {} operation now as some other thread"
	+ " already got exception or operation failed", this.operation, file.getKey());
	break;
	}
	}

	long end = Time.monotonicNow();
	LOG.debug("Time taken to process {} files count for {} operation: {} ms",
	processedFilesCount, this.operation, (end - start));
	if (processedFilesCount > 0) {
	threadsUsed.getAndIncrement();
	}
	}
	}
	}