From 6dcdbdfea13e52425f94f56137cc34634adb43a4 Mon Sep 17 00:00:00 2001 From: anuchan Date: Wed, 20 Jul 2016 22:46:29 -0700 Subject: [PATCH 1/5] Simplifying the task group logic --- .../java/com/microsoft/azure/DAGNode.java | 4 +- .../java/com/microsoft/azure/DAGraph.java | 26 ++++++++----- .../com/microsoft/azure/TaskGroupBase.java | 37 +------------------ 3 files changed, 22 insertions(+), 45 deletions(-) diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java b/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java index 112130413f..c340314f1f 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java @@ -10,6 +10,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.concurrent.locks.ReentrantLock; /** * The type representing node in a {@link DAGraph}. @@ -20,7 +21,7 @@ public class DAGNode extends Node { private List dependentKeys; private int toBeResolved; private boolean isPreparer; - + public ReentrantLock lock; /** * Creates a DAG node. * @@ -30,6 +31,7 @@ public class DAGNode extends Node { public DAGNode(String key, T data) { super(key, data); dependentKeys = new ArrayList<>(); + lock = new ReentrantLock(); } /** diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/DAGraph.java b/azure-client-runtime/src/main/java/com/microsoft/azure/DAGraph.java index 58179e0152..b9c9ec291d 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/DAGraph.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/DAGraph.java @@ -7,9 +7,8 @@ package com.microsoft.azure; -import java.util.ArrayDeque; import java.util.Map; -import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; /** * Type representing a DAG (directed acyclic graph). @@ -20,7 +19,7 @@ * @param the type of the nodes in the graph */ public class DAGraph> extends Graph { - private Queue queue; + private ConcurrentLinkedQueue queue; private boolean hasParent; private U rootNode; @@ -31,7 +30,7 @@ public class DAGraph> extends Graph { */ public DAGraph(U rootNode) { this.rootNode = rootNode; - this.queue = new ArrayDeque<>(); + this.queue = new ConcurrentLinkedQueue<>(); this.rootNode.setPreparer(true); this.addNode(rootNode); } @@ -103,10 +102,14 @@ public void prepare() { * Gets next node in the DAG which has no dependency or all of it's dependencies are resolved and * ready to be consumed. * - * @return next node or null if all the nodes have been explored + * @return next node or null if all the nodes have been explored or no node is available at this moment. */ public U getNext() { - return graph.get(queue.poll()); + String nextItemKey = queue.poll(); + if (nextItemKey == null) { + return null; + } + return graph.get(nextItemKey); } /** @@ -129,9 +132,14 @@ public void reportedCompleted(U completed) { String dependency = completed.key(); for (String dependentKey : graph.get(dependency).dependentKeys()) { DAGNode dependent = graph.get(dependentKey); - dependent.reportResolved(dependency); - if (dependent.hasAllResolved()) { - queue.add(dependent.key()); + dependent.lock.lock(); + try { + dependent.reportResolved(dependency); + if (dependent.hasAllResolved()) { + queue.add(dependent.key()); + } + } finally { + dependent.lock.lock(); } } } diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java b/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java index efc8d30e49..faedbb7d1c 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java @@ -58,11 +58,7 @@ public void execute() throws Exception { return; } - if (dag.isRootNode(nextNode)) { - executeRootTask(nextNode.data()); - } else { - nextNode.data().execute(this, nextNode); - } + nextNode.data().execute(this, nextNode); } @Override @@ -72,40 +68,11 @@ public ServiceCall executeAsync(final ServiceCallback callback) { return null; } - if (dag.isRootNode(nextNode)) { - return executeRootTaskAsync(nextNode.data(), callback); - } else { - return nextNode.data().executeAsync(this, nextNode, callback); - } + return nextNode.data().executeAsync(this, nextNode, callback); } @Override public T taskResult(String taskId) { return dag.getNodeData(taskId).result(); } - - /** - * Executes the root task in this group. - *

- * This method will be invoked when all the task dependencies of the root task are finished - * executing, at this point root task can be executed by consuming the result of tasks it - * depends on. - * - * @param task the root task in this group - * @throws Exception the exception - */ - public abstract void executeRootTask(TaskItem task) throws Exception; - - /** - * Executes the root task in this group asynchronously. - *

- * This method will be invoked when all the task dependencies of the root task are finished - * executing, at this point root task can be executed by consuming the result of tasks it - * depends on. - * - * @param task the root task in this group - * @param callback the callback when the task fails or succeeds - * @return the handle to the REST call - */ - public abstract ServiceCall executeRootTaskAsync(TaskItem task, ServiceCallback callback); } From bdb7479fbbb2685fb1d3276a15cf55b90eaeb036 Mon Sep 17 00:00:00 2001 From: anuchan Date: Thu, 21 Jul 2016 15:51:04 -0700 Subject: [PATCH 2/5] Restructuring and inital parallization work --- .../com/microsoft/azure/TaskGroupBase.java | 88 +++++++++++++++++-- 1 file changed, 83 insertions(+), 5 deletions(-) diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java b/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java index faedbb7d1c..a589c7149c 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java @@ -9,6 +9,54 @@ import com.microsoft.rest.ServiceCall; import com.microsoft.rest.ServiceCallback; +import com.microsoft.rest.ServiceResponse; + +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.ConcurrentLinkedQueue; + +/** + /** + * An instance of this class provides access to the underlying REST service call running + * in parallel. + * + * @param + */ +class ParallelServiceCall extends ServiceCall { + private TaskGroupBase taskGroup; + + /** + * Creates a ParallelServiceCall + * + * @param taskGroup the task group + */ + public ParallelServiceCall(TaskGroupBase taskGroup) { + super(null); + this.taskGroup = taskGroup; + } + + /** + * Cancels all the service calls currently executing. + */ + public void cancel() { + for (ServiceCall call : this.taskGroup.calls()) { + call.cancel(); + } + } + + /** + * @return true if the call has been canceled; false otherwise. + */ + public boolean isCancelled() { + for (ServiceCall call : this.taskGroup.calls()) { + if (!call.isCanceled()) { + return false; + } + } + return true; + } +} /** * The base implementation of TaskGroup interface. @@ -18,6 +66,9 @@ public abstract class TaskGroupBase implements TaskGroup> { private DAGraph, DAGNode>> dag; + private ConcurrentLinkedQueue serviceCalls = new ConcurrentLinkedQueue<>(); + + public ParallelServiceCall parallelServiceCall; /** * Creates TaskGroupBase. @@ -27,6 +78,11 @@ public abstract class TaskGroupBase */ public TaskGroupBase(String rootTaskItemId, TaskItem rootTaskItem) { this.dag = new DAGraph<>(new DAGNode<>(rootTaskItemId, rootTaskItem)); + this.parallelServiceCall = new ParallelServiceCall<>(this); + } + + List calls() { + return Collections.unmodifiableList(Arrays.asList(serviceCalls.toArray(new ServiceCall[0]))); } @Override @@ -63,12 +119,34 @@ public void execute() throws Exception { @Override public ServiceCall executeAsync(final ServiceCallback callback) { - final DAGNode> nextNode = dag.getNext(); - if (nextNode == null) { - return null; - } + ServiceCall serviceCall = null; + DAGNode> nextNode = dag.getNext(); + while (nextNode != null) { + if (dag.isRootNode(nextNode)) { + serviceCall = nextNode.data().executeAsync(this, nextNode, new ServiceCallback() { + @Override + public void failure(Throwable t) { + callback.failure(t); + } + + @Override + public void success(ServiceResponse result) { + callback.success(result); + } + }); + } else { + serviceCall = nextNode.data().executeAsync(this, nextNode, callback); + } - return nextNode.data().executeAsync(this, nextNode, callback); + if (serviceCall != null) { + // We need to filter out the null value returned by executeAsync. This can + // happen when TaskItem::executeAsync invokes TaskGroupBase::executeAsync + // but there is no task available in the queue at the moment. + this.serviceCalls.add(serviceCall); + } + nextNode = dag.getNext(); + } + return serviceCall; } @Override From f05ff945a8c3c74e64caf29dbd8cbcb3ce670c0a Mon Sep 17 00:00:00 2001 From: anuchan Date: Thu, 21 Jul 2016 17:14:17 -0700 Subject: [PATCH 3/5] Fixing checkstyle errors --- .../src/main/java/com/microsoft/azure/DAGNode.java | 5 +++++ .../src/main/java/com/microsoft/azure/TaskGroupBase.java | 7 +++++-- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java b/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java index c340314f1f..ba6676232a 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java @@ -21,7 +21,12 @@ public class DAGNode extends Node { private List dependentKeys; private int toBeResolved; private boolean isPreparer; + + /** + * The lock to be used while performing thread safe operation on this node. + */ public ReentrantLock lock; + /** * Creates a DAG node. * diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java b/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java index a589c7149c..9b5fcf16cd 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java @@ -27,11 +27,11 @@ class ParallelServiceCall extends ServiceCall { private TaskGroupBase taskGroup; /** - * Creates a ParallelServiceCall + * Creates a ParallelServiceCall. * * @param taskGroup the task group */ - public ParallelServiceCall(TaskGroupBase taskGroup) { + ParallelServiceCall(TaskGroupBase taskGroup) { super(null); this.taskGroup = taskGroup; } @@ -68,6 +68,9 @@ public abstract class TaskGroupBase private DAGraph, DAGNode>> dag; private ConcurrentLinkedQueue serviceCalls = new ConcurrentLinkedQueue<>(); + /** + * Wraps the services calls running in parallel. + */ public ParallelServiceCall parallelServiceCall; /** From d0abca40cebb6d9b0d67b1d4323262e336dcc5f4 Mon Sep 17 00:00:00 2001 From: anuchan Date: Thu, 21 Jul 2016 17:32:06 -0700 Subject: [PATCH 4/5] More check-style fixes --- .../src/main/java/com/microsoft/azure/DAGNode.java | 13 ++++++++----- .../src/main/java/com/microsoft/azure/DAGraph.java | 4 ++-- .../java/com/microsoft/azure/TaskGroupBase.java | 14 +++++++++----- 3 files changed, 19 insertions(+), 12 deletions(-) diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java b/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java index ba6676232a..2a340b31db 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java @@ -21,11 +21,7 @@ public class DAGNode extends Node { private List dependentKeys; private int toBeResolved; private boolean isPreparer; - - /** - * The lock to be used while performing thread safe operation on this node. - */ - public ReentrantLock lock; + private ReentrantLock lock; /** * Creates a DAG node. @@ -39,6 +35,13 @@ public DAGNode(String key, T data) { lock = new ReentrantLock(); } + /** + * @return the lock to be used while performing thread safe operation on this node. + */ + public ReentrantLock Lock() { + return this.lock; + } + /** * @return a list of keys of nodes in {@link DAGraph} those are dependents on this node */ diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/DAGraph.java b/azure-client-runtime/src/main/java/com/microsoft/azure/DAGraph.java index b9c9ec291d..f0f7b66298 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/DAGraph.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/DAGraph.java @@ -132,14 +132,14 @@ public void reportedCompleted(U completed) { String dependency = completed.key(); for (String dependentKey : graph.get(dependency).dependentKeys()) { DAGNode dependent = graph.get(dependentKey); - dependent.lock.lock(); + dependent.Lock().lock(); try { dependent.reportResolved(dependency); if (dependent.hasAllResolved()) { queue.add(dependent.key()); } } finally { - dependent.lock.lock(); + dependent.Lock().lock(); } } } diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java b/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java index 9b5fcf16cd..5ce63987de 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroupBase.java @@ -67,11 +67,7 @@ public abstract class TaskGroupBase implements TaskGroup> { private DAGraph, DAGNode>> dag; private ConcurrentLinkedQueue serviceCalls = new ConcurrentLinkedQueue<>(); - - /** - * Wraps the services calls running in parallel. - */ - public ParallelServiceCall parallelServiceCall; + private ParallelServiceCall parallelServiceCall; /** * Creates TaskGroupBase. @@ -98,6 +94,14 @@ public boolean isPreparer() { return dag.isPreparer(); } + /** + * @return Gets the ParallelServiceCall instance that wraps the service calls running + * in parallel. + */ + public ParallelServiceCall parallelServiceCall() { + return this.parallelServiceCall; + } + @Override public void merge(TaskGroup> parentTaskGroup) { dag.merge(parentTaskGroup.dag()); From da1f9a469e4a03188765f678eb9f6c70762a41f0 Mon Sep 17 00:00:00 2001 From: anuchan Date: Thu, 21 Jul 2016 17:37:29 -0700 Subject: [PATCH 5/5] last checkstyle fix --- .../src/main/java/com/microsoft/azure/DAGNode.java | 2 +- .../src/main/java/com/microsoft/azure/DAGraph.java | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java b/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java index 2a340b31db..4e1848fd8c 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/DAGNode.java @@ -38,7 +38,7 @@ public DAGNode(String key, T data) { /** * @return the lock to be used while performing thread safe operation on this node. */ - public ReentrantLock Lock() { + public ReentrantLock lock() { return this.lock; } diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/DAGraph.java b/azure-client-runtime/src/main/java/com/microsoft/azure/DAGraph.java index f0f7b66298..2a3e41d768 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/DAGraph.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/DAGraph.java @@ -132,14 +132,14 @@ public void reportedCompleted(U completed) { String dependency = completed.key(); for (String dependentKey : graph.get(dependency).dependentKeys()) { DAGNode dependent = graph.get(dependentKey); - dependent.Lock().lock(); + dependent.lock().lock(); try { dependent.reportResolved(dependency); if (dependent.hasAllResolved()) { queue.add(dependent.key()); } } finally { - dependent.Lock().lock(); + dependent.lock().lock(); } } }