Changed priorities of tasks scheduled by TimeoutWithErrorTask to execute

related tasks close together in order to avoid impact of
SerialExecutor's queue on timeouts.

Author: jodzga

src/com/linkedin/parseq/TimeoutWithErrorTask.java

@@ -48,6 +48,7 @@ public TimeoutWithErrorTask(final String name, final long time,
     _time = time;
     _unit = unit;
     _task = task;
+    setPriority(task.getPriority());
   }
 
   @Override
@@ -68,7 +69,14 @@ public void run()
       }
     });
 
+    //timeout tasks should run as early as possible
+    timeoutTask.setPriority(Priority.MAX_PRIORITY);
     context.createTimer(_time, _unit, timeoutTask);
+
+    //set priority of the task for which we just scheduled timeout
+    //to be executed next, unless there exist other tasks with higher priority
+    //e.g. other timeouts
+    _task.setPriority(higherThan(getPriority()));
     context.run(_task);
 
     _task.addListener(new PromiseListener<T>()
@@ -85,4 +93,16 @@ public void onResolved(Promise<T> resolvedPromise)
 
     return result;
   }
+
+  /**
+   * Returns priority higher than given priority. If given
+   * priority has max value, then max is returned.
+   */
+  private static int higherThan(int priority) {
+    if (priority == Priority.MAX_PRIORITY) {
+      return Priority.MAX_PRIORITY;
+    } else {
+      return priority + 1;
+    }
+  }
 }

test/com/linkedin/parseq/TestTasks.java

@@ -16,21 +16,28 @@
 
 package com.linkedin.parseq;
 
-import com.linkedin.parseq.promise.Promise;
-import com.linkedin.parseq.promise.PromiseListener;
-import com.linkedin.parseq.promise.Promises;
-import org.testng.annotations.Test;
+import static com.linkedin.parseq.TestUtil.value;
+import static org.testng.AssertJUnit.assertEquals;
+import static org.testng.AssertJUnit.assertFalse;
+import static org.testng.AssertJUnit.assertTrue;
+import static org.testng.AssertJUnit.fail;
 
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.Callable;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 
-import static com.linkedin.parseq.TestUtil.value;
-import static org.testng.AssertJUnit.assertEquals;
-import static org.testng.AssertJUnit.assertFalse;
-import static org.testng.AssertJUnit.assertTrue;
-import static org.testng.AssertJUnit.fail;
+import org.testng.annotations.Test;
+
+import com.linkedin.parseq.promise.Promise;
+import com.linkedin.parseq.promise.PromiseListener;
+import com.linkedin.parseq.promise.Promises;
+import com.linkedin.parseq.promise.SettablePromise;
 
 /**
  * @author Chris Pettitt ([email protected])
@@ -171,6 +178,52 @@ public String call() throws Exception
     assertEquals(error, timeoutTask.getError());
   }
 
+  /**
+   * Test scenario in which there are many TimeoutWithErrorTasks scheduled for execution
+   * e.g. by using Tasks.par().
+   */
+  @Test
+  public void testManyTimeoutTaskWithoutTimeoutOnAQueeu() throws InterruptedException, IOException
+  {
+    final String value = "value";
+    final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
+
+    List<Task<String>> tasks = new ArrayList<Task<String>>();
+    for (int i = 0; i < 50; i++) {
+
+      // Task which simulates doing something for 0.5ms and setting response
+      // asynchronously after 5ms.
+      Task<String> t = new BaseTask<String>("test") {
+        @Override
+        protected Promise<? extends String> run(Context context) throws Throwable {
+          final SettablePromise<String> result = Promises.settable();
+          Thread.sleep(0, 500000);
+          scheduler.schedule(new Runnable() {
+            @Override
+            public void run() {
+              result.done(value);
+            }
+          }, 5, TimeUnit.MILLISECONDS);
+          return result;
+        }
+      };
+      // add 50ms timeout for the task
+      tasks.add(Tasks.timeoutWithError(50, TimeUnit.MILLISECONDS, t));
+    }
+
+    // final task runs all the tasks in parallel
+    final Task<?> timeoutTask = Tasks.par(tasks);
+
+    getEngine().run(timeoutTask);
+
+    assertTrue(timeoutTask.await(5, TimeUnit.SECONDS));
+
+    scheduler.shutdown();
+
+    //tasks should not time out
+    assertEquals(false, timeoutTask.isFailed());
+  }
+
   @Test
   public void testSetPriorityBelowMinValue()
   {