Cancel plan execution if its serial executor loop was rejected

If the engine's task executor is configured to reject execution upon
saturation (e.g. with AbortPolicy) we will now abort the plan that is
being executed. We do not recommend rejecting work at this level since
it will cause in-flight plans to be aborted. Instead, we recommend using
standard techniques for managing at the engine level, e.g.
back-pressure, load shedding, or degraded responses. The time spent on
plan execution can also be bounded using a timeout task.

Author: cpettitt-linkedin

CHANGELOG.md

@@ -1,3 +1,9 @@
+v1.3.7
+------
+
+* We now cancel a plan if its execution fails due to a
+  RejectedExecutionException being raised from the engine's task executor.
+
 v1.3.6
 ------
 

src/com/linkedin/parseq/Engine.java

@@ -18,18 +18,19 @@
 
 import com.linkedin.parseq.internal.ContextImpl;
 import com.linkedin.parseq.internal.InternalUtil;
+import com.linkedin.parseq.internal.RejectedSerialExecutionHandler;
+import com.linkedin.parseq.internal.SerialExecutionException;
 import com.linkedin.parseq.internal.SerialExecutor;
 import com.linkedin.parseq.internal.TaskLogImpl;
 import com.linkedin.parseq.promise.Promise;
 import com.linkedin.parseq.promise.PromiseListener;
 import org.slf4j.ILoggerFactory;
 import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -42,6 +43,7 @@
 public class Engine
 {
   private static final String LOGGER_BASE = Engine.class.getName();
+  private static final Logger LOG = LoggerFactory.getLogger(LOGGER_BASE);
 
   private static final State INIT = new State(StateName.RUN, 0);
   private static final State TERMINATED = new State(StateName.TERMINATED, 0);
@@ -126,7 +128,8 @@ public void run(final Task<?> task)
 
     final Logger planLogger = _loggerFactory.getLogger(LOGGER_BASE + ":planClass=" + task.getClass().getName());
     final TaskLog taskLog = new TaskLogImpl(task, _allLogger, _rootLogger, planLogger);
-    new ContextImpl(new SerialExecutor(_taskExecutor), _timerExecutor, task, taskLog, this).runTask();
+    new ContextImpl(new SerialExecutor(_taskExecutor, new CancelPlanRejectionHandler(task)),
+                    _timerExecutor, task, taskLog, this).runTask();
 
     InternalUtil.unwildcardTask(task).addListener(_taskDoneListener);
   }
@@ -234,4 +237,23 @@ private State(final StateName stateName, final long pendingCount)
       _stateName = stateName;
     }
   }
+
+  private static class CancelPlanRejectionHandler implements RejectedSerialExecutionHandler
+  {
+    private final Task<?> _task;
+
+    private CancelPlanRejectionHandler(Task<?> task)
+    {
+      _task = task;
+    }
+
+    @Override
+    public void rejectedExecution(Throwable error)
+    {
+      final String msg = "Serial executor loop failed for plan: " + _task.getName();
+      final SerialExecutionException ex = new SerialExecutionException(msg, error);
+      final boolean wasCancelled = _task.cancel(ex);
+      LOG.error(msg + ". The plan was " + (wasCancelled ? "" : "not ") + "cancelled.", ex);
+    }
+  }
 }

src/com/linkedin/parseq/internal/RejectedSerialExecutionHandler.java

@@ -0,0 +1,16 @@
+package com.linkedin.parseq.internal;
+
+/**
+ * A handler that is invoked if the {@link SerialExecutor}'s execution loop
+ * fails during resubmission to the underlying executor.
+ */
+public interface RejectedSerialExecutionHandler
+{
+  /**
+   * This method is invoked if a {@link SerialExecutor}'s execution loop cannot
+   * be resubmitted to the underlying executor.
+   *
+   * @param error the error that was raised by the underlying executor.
+   */
+  void rejectedExecution(Throwable error);
+}

src/com/linkedin/parseq/internal/SerialExecutionException.java

@@ -0,0 +1,11 @@
+package com.linkedin.parseq.internal;
+
+public class SerialExecutionException extends Exception
+{
+  private static final long serialVersionUID = 0L;
+
+  public SerialExecutionException(String msg, Throwable error)
+  {
+    super(msg, error);
+  }
+}

src/com/linkedin/parseq/internal/SerialExecutor.java

@@ -27,28 +27,52 @@
  * <p/>
  * For more on the happens-before constraint see the {@code java.util.concurrent}
  * package documentation.
+ * <p/>
+ * It is possible for the underlying executor to throw an exception signaling
+ * that it is not able to accept new work. For example, this can occur with an
+ * executor that has a bounded queue size and an
+ * {@link java.util.concurrent.ThreadPoolExecutor.AbortPolicy}. If this occurs
+ * the executor will run the {@code rejectionHandler} to signal this failure
+ * to a layer that can more appropriate handle this event.
  *
  * @author Chris Pettitt ([email protected])
  */
 public class SerialExecutor implements Executor
 {
   private final Executor _executor;
+  private final RejectedSerialExecutionHandler _rejectionHandler;
   private final ExecutorLoop _executorLoop = new ExecutorLoop();
   private final FIFOPriorityQueue<Runnable> _queue = new FIFOPriorityQueue<Runnable>();
   private final AtomicInteger _pendingCount = new AtomicInteger();
 
-  public SerialExecutor(final Executor executor)
+  public SerialExecutor(final Executor executor, final RejectedSerialExecutionHandler rejectionHandler)
   {
+    assert executor != null;
+    assert rejectionHandler != null;
+
     _executor = executor;
+    _rejectionHandler = rejectionHandler;
   }
 
   public void execute(final Runnable runnable)
   {
     _queue.add(runnable);
     if (_pendingCount.getAndIncrement() == 0)
+    {
+      tryExecuteLoop();
+    }
+  }
+
+  private void tryExecuteLoop()
+  {
+    try
     {
       _executor.execute(_executorLoop);
     }
+    catch (Throwable t)
+    {
+      _rejectionHandler.rejectedExecution(t);
+    }
   }
 
   private class ExecutorLoop implements Runnable
@@ -78,7 +102,7 @@ public void run()
         // above for more details.
         if (_pendingCount.decrementAndGet() > 0)
         {
-          _executor.execute(this);
+          tryExecuteLoop();
         }
       }
     }

test/com/linkedin/parseq/TestEngine.java

@@ -22,9 +22,13 @@
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
 
+import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
 import static org.testng.AssertJUnit.assertEquals;
@@ -155,4 +159,66 @@ protected Promise<? extends String> run(final Context context) throws Exception
     assertTrue(sucTask.await(50, TimeUnit.MILLISECONDS));
     assertEquals(sucValue, sucTask.get());
   }
+
+  @Test
+  public void testFailPlanExecution() throws InterruptedException
+  {
+    // This test ensures that if execution of a plan's serial executor loop
+    // fails, e.g. in the case that the underlying executor is saturated, that
+    // we fail the plan. To simplify this test, we constructor our own executor
+    // instead of using the default executor set up for test.
+    final ExecutorService executorService = new ThreadPoolExecutor(1, 1,
+                                                                   0, TimeUnit.SECONDS,
+                                                                   new ArrayBlockingQueue<Runnable>(1),
+                                                                   new ThreadPoolExecutor.AbortPolicy());
+    final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();
+
+    try
+    {
+      final Engine engine = new EngineBuilder()
+          .setTaskExecutor(executorService)
+          .setTimerScheduler(scheduledExecutorService)
+          .build();
+
+      // First we submit two tasks that will never finish. This saturates the
+      // underlying executor by using its only thread and saturating its
+      // single slot queue.
+      engine.run(neverEndingBlockingTask());
+      engine.run(neverEndingBlockingTask());
+
+      // Now we submit another task. The execution loop for this task will fail
+      // during submit to the underlying executor. We expect that it will be
+      // cancelled.
+      final Task<?> task = neverEndingBlockingTask();
+      engine.run(task);
+      assertTrue(task.await(5, TimeUnit.SECONDS));
+      assertTrue(task.isFailed());
+      assertTrue("Expected underlying exception to be instance of RejectedExecutionException, but was: " + task.getError().getCause(),
+                 task.getError().getCause() instanceof RejectedExecutionException);
+
+      engine.shutdown();
+    }
+    finally
+    {
+      scheduledExecutorService.shutdownNow();
+      executorService.shutdownNow();
+    }
+  }
+
+  /**
+   * A task that blocks forever when it is executed, tying up whatever thread
+   * executes it.
+   */
+  private Task<?> neverEndingBlockingTask()
+  {
+    return new BaseTask<Object>()
+    {
+      @Override
+      protected Promise<?> run(Context context) throws Throwable
+      {
+        new CountDownLatch(1).await();
+        return Promises.value("A value that should never be seen!");
+      }
+    };
+  }
 }