Introduce PlanContext for plan-level configuration and facilities

Author: cpettitt-linkedin

src/com/linkedin/parseq/Engine.java

@@ -21,6 +21,7 @@
 import com.linkedin.parseq.internal.RejectedSerialExecutionHandler;
 import com.linkedin.parseq.internal.SerialExecutionException;
 import com.linkedin.parseq.internal.SerialExecutor;
+import com.linkedin.parseq.internal.PlanContext;
 import com.linkedin.parseq.internal.TaskLogger;
 import com.linkedin.parseq.promise.Promise;
 import com.linkedin.parseq.promise.PromiseListener;
@@ -32,6 +33,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
 /**
@@ -50,6 +52,8 @@ public class Engine
 
   private static enum StateName { RUN, SHUTDOWN, TERMINATED }
 
+  private final AtomicLong NEXT_PLAN_ID = new AtomicLong();
+
   private final Executor _taskExecutor;
   private final DelayedExecutor _timerExecutor;
   private final ILoggerFactory _loggerFactory;
@@ -126,10 +130,11 @@ public void run(final Task<?> task)
       newState = new State(StateName.RUN, currState._pendingCount + 1);
     } while (!_stateRef.compareAndSet(currState, newState));
 
+    final long planId = NEXT_PLAN_ID.getAndIncrement();
     final Logger planLogger = _loggerFactory.getLogger(LOGGER_BASE + ":planClass=" + task.getClass().getName());
     final TaskLogger taskLogger = new TaskLogger(task, _allLogger, _rootLogger, planLogger);
-    new ContextImpl(new SerialExecutor(_taskExecutor, new CancelPlanRejectionHandler(task)),
-                    _timerExecutor, task, taskLogger, this).runTask();
+    final Executor taskExecutor = new SerialExecutor(_taskExecutor, new CancelPlanRejectionHandler(task));
+    new ContextImpl(new PlanContext(planId, this, taskExecutor, _timerExecutor, taskLogger), task).runTask();
 
     InternalUtil.unwildcardTask(task).addListener(_taskDoneListener);
   }

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

@@ -19,9 +19,7 @@
 import com.linkedin.parseq.After;
 import com.linkedin.parseq.Cancellable;
 import com.linkedin.parseq.Context;
-import com.linkedin.parseq.DelayedExecutor;
 import com.linkedin.parseq.EarlyFinishException;
-import com.linkedin.parseq.Engine;
 import com.linkedin.parseq.Task;
 import com.linkedin.parseq.promise.Promise;
 import com.linkedin.parseq.promise.PromiseListener;
@@ -32,7 +30,6 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.Executor;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -57,17 +54,10 @@ public class ContextImpl implements Context, Cancellable
     }
   }
 
-  private final DelayedExecutor _timerScheduler;
-
-  /* An executor that provides two guarantees:
-   *
-   * 1. Only one task is executed at a time
-   * 2. The completion of a task happens-before the execution of the next task
-   *
-   * For more on the happens-before constraint see the java.util.concurrent
-   * package documentation.
+  /**
+   * Plan level configuration and facilities.
    */
-  private final Executor _taskExecutor;
+  private final PlanContext _planContext;
 
   private final Task<Object> _task;
 
@@ -78,36 +68,24 @@ public class ContextImpl implements Context, Cancellable
 
   private final Task<?> _parent;
   private final List<Task<?>> _predecessorTasks;
-  private final TaskLogger _taskLogger;
-
-  private final Engine _engine;
 
   private final ConcurrentLinkedQueue<Cancellable> _cancellables = new ConcurrentLinkedQueue<Cancellable>();
 
-  public ContextImpl(final Executor taskExecutor,
-                     final DelayedExecutor timerScheduler,
-                     final Task<?> task,
-                     final TaskLogger taskLogger,
-                     final Engine engine)
+  public ContextImpl(final PlanContext planContext,
+                     final Task<?> task)
   {
-    this(taskExecutor, timerScheduler, task, NO_PARENT, NO_PREDECESSORS, taskLogger, engine);
+    this(planContext, task, NO_PARENT, NO_PREDECESSORS);
   }
 
-  private ContextImpl(final Executor taskExecutor,
-                      final DelayedExecutor timerScheduler,
+  private ContextImpl(final PlanContext planContext,
                       final Task<?> task,
                       final Task<?> parent,
-                      final List<Task<?>> predecessorTasks,
-                      final TaskLogger taskLogger,
-                      final Engine engine)
+                      final List<Task<?>> predecessorTasks)
   {
-    _timerScheduler = timerScheduler;
-    _taskExecutor = taskExecutor;
+    _planContext = planContext;
     _task = InternalUtil.unwildcardTask(task);
     _parent = parent;
     _predecessorTasks = predecessorTasks;
-    _taskLogger = taskLogger;
-    _engine = engine;
   }
 
   public void runTask()
@@ -127,15 +105,15 @@ public void onResolved(Promise<Object> resolvedPromise)
       }
     });
 
-    _taskExecutor.execute(new PrioritizableRunnable()
+    _planContext.execute(new PrioritizableRunnable()
     {
       @Override
       public void run()
       {
         _inTask.set(_task);
         try
         {
-          _task.contextRun(ContextImpl.this, _taskLogger, _parent, _predecessorTasks);
+          _task.contextRun(ContextImpl.this, _planContext.getTaskLogger(), _parent, _predecessorTasks);
         }
         finally
         {
@@ -156,7 +134,7 @@ public Cancellable createTimer(final long time, final TimeUnit unit,
                                  final Task<?> task)
   {
     checkInTask();
-    final Cancellable cancellable = _timerScheduler.schedule(time, unit, new Runnable()
+    final Cancellable cancellable = _planContext.schedule(time, unit, new Runnable()
     {
       @Override
       public void run()
@@ -214,19 +192,19 @@ public boolean cancel(Exception reason)
   {
     boolean result = _task.cancel(reason);
     //run the task to capture the trace data
-    _task.contextRun(this, _taskLogger, _parent, _predecessorTasks);
+    _task.contextRun(this, _planContext.getTaskLogger(), _parent, _predecessorTasks);
     return result;
   }
 
   @Override
   public Object getEngineProperty(String key)
   {
-    return _engine.getProperty(key);
+    return _planContext.getEngineProperty(key);
   }
 
   private ContextImpl createSubContext(final Task<?> task, final List<Task<?>> predecessors)
   {
-    return new ContextImpl(_taskExecutor, _timerScheduler, task, _task, predecessors, _taskLogger, _engine);
+    return new ContextImpl(_planContext, task, _task, predecessors);
   }
 
   private void runSubTask(final Task<?> task, final List<Task<?>> predecessors)

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

@@ -0,0 +1,71 @@
+package com.linkedin.parseq.internal;
+
+import com.linkedin.parseq.Cancellable;
+import com.linkedin.parseq.DelayedExecutor;
+import com.linkedin.parseq.Engine;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
+
+public class PlanContext
+{
+  /** Unique identifier for this plan. */
+  private final long _id;
+
+  /** The engine used to execute this plan. */
+  private final Engine _engine;
+
+  /**
+   *  An executor that provides two guarantees:
+   *
+   * 1. Only one task is executed at a time
+   * 2. The completion of a task happens-before the execution of the next task
+   *
+   * For more on the happens-before constraint see the java.util.concurrent
+   * package documentation.
+   */
+  private final Executor _taskExecutor;
+
+  /** Scheduler for running time delayed tasks. */
+  private final DelayedExecutor _timerScheduler;
+
+  private final TaskLogger _taskLogger;
+
+  public PlanContext(final long id,
+                     final Engine engine,
+                     final Executor taskExecutor,
+                     final DelayedExecutor timerScheduler,
+                     final TaskLogger taskLogger)
+  {
+    _id = id;
+    _engine = engine;
+    _taskExecutor = taskExecutor;
+    _timerScheduler = timerScheduler;
+    _taskLogger = taskLogger;
+  }
+
+  public long getId()
+  {
+    return _id;
+  }
+
+  public void execute(Runnable runnable)
+  {
+    _taskExecutor.execute(runnable);
+  }
+
+  public Cancellable schedule(long time, TimeUnit unit, Runnable runnable)
+  {
+    return _timerScheduler.schedule(time, unit, runnable);
+  }
+
+  public Object getEngineProperty(String key)
+  {
+    return _engine.getProperty(key);
+  }
+
+  public TaskLogger getTaskLogger()
+  {
+    return _taskLogger;
+  }
+}