drill-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jacq...@apache.org
Subject [1/2] drill git commit: SerializedExecutor sketch for consideration.
Date Mon, 02 Nov 2015 19:37:13 GMT
Repository: drill
Updated Branches:
  refs/heads/master ce593eb4c -> e7db9dcac


SerializedExecutor sketch for consideration.

Based on the EventProcessor, but recast as an Executor that accepts Runnables.
The idea would be to use this on RPC reception points to immediately hand off
work. Each Fragment would have a SerializedExecutor associated with it, serving
as a queue to deliver events to when the RPC thread receives them.


Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/839f8dac
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/839f8dac
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/839f8dac

Branch: refs/heads/master
Commit: 839f8dac2e2d0479a1552701a5274ebe8416fea6
Parents: ce593eb
Author: Chris Westin <cwestin@yahoo.com>
Authored: Fri May 22 14:17:41 2015 -0700
Committer: Jacques Nadeau <jacques@apache.org>
Committed: Mon Nov 2 07:54:24 2015 -0800

----------------------------------------------------------------------
 .../apache/drill/common/SerializedExecutor.java | 110 +++++++++++++++++++
 1 file changed, 110 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/drill/blob/839f8dac/common/src/main/java/org/apache/drill/common/SerializedExecutor.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/drill/common/SerializedExecutor.java b/common/src/main/java/org/apache/drill/common/SerializedExecutor.java
new file mode 100644
index 0000000..203be49
--- /dev/null
+++ b/common/src/main/java/org/apache/drill/common/SerializedExecutor.java
@@ -0,0 +1,110 @@
+/**
+ * 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.drill.common;
+
+import java.util.LinkedList;
+import java.util.concurrent.Executor;
+
+/**
+ * Serializes execution of multiple submissions to a single target, while
+ * still using a thread pool to execute those submissions. Provides an
+ * implicit queueing capability for a single target that requires any commands
+ * that execute against it to be serialized.
+ */
+public class SerializedExecutor implements Executor {
+  private boolean isProcessing = false;
+  private final LinkedList<Runnable> queuedRunnables = new LinkedList<>();
+  private final Executor underlyingExecutor;
+
+  /**
+   * Constructor.
+   *
+   * @param underlyingExecutor underlying executor to use to execute commands
+   *   submitted to this SerializedExecutor
+   */
+  public SerializedExecutor(Executor underlyingExecutor) {
+    this.underlyingExecutor = underlyingExecutor;
+  }
+
+  /**
+   * An exception occurred in the last command executed; this reports that
+   * to the subclass of SerializedExecutor.
+   *
+   * <p>The default implementation of this method throws an exception, which
+   * is considered an error (see below). Implementors have two alternatives:
+   * Arrange not to throw from your commands' run(), or if they do,
+   * provide an override of this method that handles any exception that
+   * is thrown.</p>
+   *
+   * <p>It is an error for this to throw an exception, and doing so will
+   * terminate the thread with an IllegalStateException. Derivers must
+   * handle any reported exceptions in other ways.</p>
+   *
+   * @param command the command that caused the exception
+   * @param t the exception
+   */
+  protected void runException(Runnable command, Throwable t) {
+    throw new IllegalStateException("unhandled exception thrown by command");
+  }
+
+  private class RunnableProcessor implements Runnable {
+    private Runnable command;
+
+    public RunnableProcessor(Runnable command) {
+      this.command = command;
+    }
+
+    @Override
+    public void run() {
+      while (true) {
+        try {
+          command.run();
+        } catch(Exception | AssertionError e) {
+          try {
+            runException(command, e);
+          } catch(Exception | AssertionError ee) {
+            throw new IllegalStateException("Exception handler threw an exception", ee);
+          }
+        }
+
+        synchronized (queuedRunnables) {
+          if (queuedRunnables.isEmpty()) {
+            isProcessing = false;
+            break;
+          }
+
+          command = queuedRunnables.removeFirst();
+        }
+      }
+    }
+  }
+
+  @Override
+  public void execute(Runnable command) {
+    synchronized (queuedRunnables) {
+      if (isProcessing) {
+        queuedRunnables.addLast(command);
+        return;
+      }
+
+      isProcessing = true;
+    }
+
+    underlyingExecutor.execute(new RunnableProcessor(command));
+  }
+}


Mime
View raw message