add multi-threaded executor

Author: functioner

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSEditLogAsync.java

@@ -24,7 +24,8 @@
 import java.util.List;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -48,7 +49,9 @@ class FSEditLogAsync extends FSEditLog implements Runnable {
 
   // requires concurrent access from caller threads and syncing thread.
   private final BlockingQueue<Edit> editPendingQ;
-  private final BlockingQueue<EditSyncEx> logSyncNotifyQ;
+
+  // Thread pool for executing logSyncNotify
+  private final ExecutorService logSyncNotifyExecutor;
 
   // only accessed by syncing thread so no synchronization required.
   // queue is unbounded because it's effectively limited by the size
@@ -65,7 +68,9 @@ class FSEditLogAsync extends FSEditLog implements Runnable {
             DFS_NAMENODE_EDITS_ASYNC_LOGGING_PENDING_QUEUE_SIZE_DEFAULT);
 
     editPendingQ = new ArrayBlockingQueue<>(editPendingQSize);
-    logSyncNotifyQ = new LinkedBlockingQueue<>();
+
+    // the thread pool size should be configurable later, and justified with a rationale
+    logSyncNotifyExecutor = Executors.newFixedThreadPool(10);
   }
 
   private boolean isSyncThreadAlive() {
@@ -120,6 +125,7 @@ void openForWrite(int layoutVersion) throws IOException {
   public void close() {
     super.close();
     stopSyncThread();
+    logSyncNotifyExecutor.shutdown();
   }
 
   @Override
@@ -230,33 +236,8 @@ private Edit dequeueEdit() throws InterruptedException {
     return syncWaitQ.isEmpty() ? editPendingQ.take() : editPendingQ.poll();
   }
 
-  private static class EditSyncEx {
-    final Edit edit;
-    final RuntimeException ex;
-    EditSyncEx(Edit edit, RuntimeException ex) {
-      this.edit = edit;
-      this.ex = ex;
-    }
-  }
-
-  private class LogSyncNotifyThread extends Thread {
-    volatile boolean stopped = false;
-
-    @Override
-    public void run() {
-      try {
-        while (!stopped) {
-          EditSyncEx editSyncEx = logSyncNotifyQ.take();
-          editSyncEx.edit.logSyncNotify(editSyncEx.ex);
-        }
-      } catch(InterruptedException ie) {} // just swallow it
-    }
-  }
-
   @Override
   public void run() {
-    final LogSyncNotifyThread logSyncNotifyThread = new LogSyncNotifyThread();
-    logSyncNotifyThread.start();
     try {
       while (true) {
         boolean doSync;
@@ -279,14 +260,14 @@ public void run() {
             syncEx = ex;
           }
           while ((edit = syncWaitQ.poll()) != null) {
-            logSyncNotifyQ.put(new EditSyncEx(edit, syncEx));
+            final Edit notifyEdit = edit;
+            final RuntimeException ex = syncEx;
+            logSyncNotifyExecutor.submit(() -> notifyEdit.logSyncNotify(ex));
           }
         }
       }
     } catch (InterruptedException ie) {
       LOG.info(Thread.currentThread().getName() + " was interrupted, exiting");
-      logSyncNotifyThread.stopped = true;
-      logSyncNotifyThread.interrupt();
     } catch (Throwable t) {
       terminate(t);
     }