From 6dcdbdfea13e52425f94f56137cc34634adb43a4 Mon Sep 17 00:00:00 2001 From: anuchan Date: Wed, 20 Jul 2016 22:46:29 -0700 Subject: [PATCH 01/10] 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 02/10] 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 03/10] 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 04/10] 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 05/10] 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(); } } } From fb0172ef44cef499f4584a84dc909adf7102af79 Mon Sep 17 00:00:00 2001 From: anuchan Date: Fri, 22 Jul 2016 14:47:46 -0700 Subject: [PATCH 06/10] Updated runtime sync script, Fixing the bug of not calling callback on stg create success, fixing the bug of not unlocking node --- .../src/main/java/com/microsoft/azure/DAGraph.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 2a3e41d768..01917a77cf 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 @@ -139,7 +139,7 @@ public void reportedCompleted(U completed) { queue.add(dependent.key()); } } finally { - dependent.lock().lock(); + dependent.lock().unlock(); } } } From 39672551f664ab4e19913de7a0322f883776c062 Mon Sep 17 00:00:00 2001 From: Jianghao Lu Date: Fri, 22 Jul 2016 16:51:18 -0700 Subject: [PATCH 07/10] Remove subtree checks in travis --- .travis.yml | 3 --- 1 file changed, 3 deletions(-) diff --git a/.travis.yml b/.travis.yml index bc2a8e6b83..00de321762 100644 --- a/.travis.yml +++ b/.travis.yml @@ -12,6 +12,3 @@ script: - mvn clean install - mvn checkstyle:check - cd ./azure-android-client-authentication && ./gradlew check -env: - global: - secure: RnU4qnuCJRzSvAAXApk9yVhIH+gtl5RNmoVva/hzuSF0WcSaRh2CBe37KUNnbBDJaHd53L5AvHBpbcmaylbOYaRQ/vxUG1gAEHuyrX3ANvGLKYKjtg1F8i853h1Y/y/TZ9MNfzLOqlo/DtD/jAl6pOAMyxNxSEwwFjHY+zbzTOA4kXjTDPGtNwB253v46j5vzSUcEKHpAXER2RjTeurVFeMFDT78Ou+4DCFBqpenObwr1CH+YPTwIoRBvzzPFDKy8+3rdWXOCB2QloaeFntxH1NDbkhReBe4KL3Ue03ksxrAqcmoJR6qFCQebvHYQYSXTk7kmoOxRFTowGqJ0SErUQwT85MA+3JYDmWDKvI3Zq1lQvhmv7dhsbgJASTxpwW+cqStCzKhAYADB6nB8Nv5AUE/5wQDWXEANUQ2t/eWooip0IcIHeJfBj0qDk+WvjAKGSGu0tIAD71Z1h36WHxf0U/fTi673tguwrhWpnUCbQaQCiDqMm7sSWOlyIdAoxy3JwLGaJq3bRGAjgwbfY32VQ/GN00ttSYu63SFDkxYATC4/FtYheNB0Yi4bDU//me9WcB/mt4bi98nC1CBK1tVS1Rn6ATkXu5K9oMvfCX7UlKe6rGmW7tVdXqtwMEFBpH3Z94tXxQEkZ+AITociIBV85+Oy/BOYvMCwysNv/PEqx0= From 686ac75aae6fa7cb095788bdca8da7005e34a9ca Mon Sep 17 00:00:00 2001 From: anuchan Date: Fri, 22 Jul 2016 17:23:57 -0700 Subject: [PATCH 08/10] Wiring up the final call-back correctly --- .../java/com/microsoft/azure/TaskGroup.java | 2 +- .../com/microsoft/azure/TaskGroupBase.java | 24 ++++--------------- .../java/com/microsoft/azure/TaskItem.java | 3 ++- 3 files changed, 7 insertions(+), 22 deletions(-) diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroup.java b/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroup.java index 26bbbece26..3ad514d938 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroup.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/TaskGroup.java @@ -63,7 +63,7 @@ public interface TaskGroup> { * @param callback the callback to call on failure or success * @return the handle to the REST call */ - ServiceCall executeAsync(ServiceCallback callback); + ServiceCall executeAsync(ServiceCallback callback); /** * Gets the result of execution of a task in the group. 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 5ce63987de..1cd0050e5d 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,7 +9,6 @@ import com.microsoft.rest.ServiceCall; import com.microsoft.rest.ServiceCallback; -import com.microsoft.rest.ServiceResponse; import java.util.Arrays; import java.util.Collections; @@ -125,29 +124,14 @@ public void execute() throws Exception { } @Override - public ServiceCall executeAsync(final ServiceCallback callback) { + public ServiceCall executeAsync(final ServiceCallback callback) { 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); - } - + serviceCall = nextNode.data().executeAsync(this, nextNode, dag.isRootNode(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 + // Filter out the null value returned by executeAsync. that happen + // when TaskItem::executeAsync invokes TaskGroupBase::executeAsync // but there is no task available in the queue at the moment. this.serviceCalls.add(serviceCall); } diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/TaskItem.java b/azure-client-runtime/src/main/java/com/microsoft/azure/TaskItem.java index fb74c23845..1612ff4f83 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/TaskItem.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/TaskItem.java @@ -39,8 +39,9 @@ public interface TaskItem { * @param taskGroup the task group dispatching tasks * @param node the node the task item is associated with + * @param isRootNode true if the node is root node * @param callback callback to call on success or failure * @return the handle of the REST call */ - ServiceCall executeAsync(TaskGroup> taskGroup, DAGNode> node, ServiceCallback callback); + ServiceCall executeAsync(TaskGroup> taskGroup, DAGNode> node, final boolean isRootNode, ServiceCallback callback); } From 0493b5e76d854b7408971fea1f72b73100042f57 Mon Sep 17 00:00:00 2001 From: anuchan Date: Sun, 24 Jul 2016 00:29:36 -0700 Subject: [PATCH 09/10] Simplify the parallel async logic and adding circular depedency detection logic --- .../java/com/microsoft/azure/DAGraph.java | 11 +- .../main/java/com/microsoft/azure/Graph.java | 89 ++++++++- .../com/microsoft/azure/TaskGroupBase.java | 172 ++++++++++-------- .../java/com/microsoft/azure/TaskItem.java | 11 +- 4 files changed, 190 insertions(+), 93 deletions(-) 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 01917a77cf..df3ba483f6 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 @@ -153,9 +153,8 @@ public void reportedCompleted(U completed) { */ private void initializeDependentKeys() { visit(new Visitor() { - // This 'visit' will be called only once per each node. @Override - public void visit(U node) { + public void visitNode(U node) { if (node.dependencyKeys().isEmpty()) { return; } @@ -166,6 +165,14 @@ public void visit(U node) { .addDependent(dependentKey); } } + + @Override + public void visitEdge(String fromKey, String toKey, GraphEdgeType edgeType) { + System.out.println("{" + fromKey + ", " + toKey + "} " + edgeType); + if (edgeType == GraphEdgeType.BACK) { + throw new IllegalStateException("Detected circular dependency: " + findPath(fromKey, toKey)); + } + } }); } diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/Graph.java b/azure-client-runtime/src/main/java/com/microsoft/azure/Graph.java index 40ceebaa50..17440ddbeb 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/Graph.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/Graph.java @@ -12,6 +12,16 @@ import java.util.Map; import java.util.Set; +/** + * The edge types in a graph. + */ +enum GraphEdgeType { + TREE, + FORWARD, + BACK, + CROSS +} + /** * Type representing a directed graph data structure. *

@@ -23,6 +33,11 @@ public class Graph> { protected Map graph; private Set visited; + private Integer time; + private Map entryTime; + private Map exitTime; + private Map parent; + private Set processed; /** * Creates a directed graph. @@ -30,6 +45,11 @@ public class Graph> { public Graph() { this.graph = new HashMap<>(); this.visited = new HashSet<>(); + this.time = 0; + this.entryTime = new HashMap<>(); + this.exitTime = new HashMap<>(); + this.parent = new HashMap<>(); + this.processed = new HashSet<>(); } /** @@ -53,14 +73,23 @@ interface Visitor { * * @param node the node to visited */ - void visit(U node); + void visitNode(U node); + + /** + * visit an edge. + * + * @param fromKey key of the from node + * @param toKey key of the to node + * @param graphEdgeType the edge type + */ + void visitEdge(String fromKey, String toKey, GraphEdgeType graphEdgeType); } /** * Perform DFS visit in this graph. *

* The directed graph will be traversed in DFS order and the visitor will be notified as - * search explores each node + * search explores each node and edge. * * @param visitor the graph visitor */ @@ -71,15 +100,61 @@ public void visit(Visitor visitor) { } } visited.clear(); + time = 0; + entryTime.clear(); + exitTime.clear(); + parent.clear(); + processed.clear(); } private void dfs(Visitor visitor, Node node) { - visitor.visit(node); - visited.add(node.key()); - for (String childKey : node.children()) { - if (!visited.contains(childKey)) { - this.dfs(visitor, this.graph.get(childKey)); + visitor.visitNode(node); + + String fromKey = node.key(); + visited.add(fromKey); + time++; + entryTime.put(fromKey, time); + for (String toKey : node.children()) { + if (!visited.contains(toKey)) { + parent.put(toKey, fromKey); + visitor.visitEdge(fromKey, toKey, edgeType(fromKey, toKey)); + this.dfs(visitor, this.graph.get(toKey)); + } else { + visitor.visitEdge(fromKey, toKey, edgeType(fromKey, toKey)); } } + time++; + exitTime.put(fromKey, time); + processed.add(fromKey); + } + + private GraphEdgeType edgeType(String fromKey, String toKey) { + if (parent.containsKey(toKey) && parent.get(toKey).equals(fromKey)) { + return GraphEdgeType.TREE; + } + + if (visited.contains(toKey) && !processed.contains(toKey)) { + return GraphEdgeType.BACK; + } + + if (processed.contains(toKey) && entryTime.containsKey(toKey) && entryTime.containsKey(fromKey)) { + if (entryTime.get(toKey) > entryTime.get(fromKey)) { + return GraphEdgeType.FORWARD; + } + + if (entryTime.get(toKey) < entryTime.get(fromKey)) { + return GraphEdgeType.CROSS; + } + } + + throw new IllegalStateException("Internal Error: Unable to locate the edge type {" + fromKey + ", " + toKey + "}"); + } + + protected String findPath(String start, String end) { + if (start.equals(end)) { + return start; + } else { + return findPath(start, parent.get(end)) + " -> " + end; + } } } 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 1cd0050e5d..b607a41f54 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,54 +9,10 @@ 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 - */ - 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. * @@ -65,8 +21,7 @@ public boolean isCancelled() { public abstract class TaskGroupBase implements TaskGroup> { private DAGraph, DAGNode>> dag; - private ConcurrentLinkedQueue serviceCalls = new ConcurrentLinkedQueue<>(); - private ParallelServiceCall parallelServiceCall; + private ParallelServiceCall parallelServiceCall; /** * Creates TaskGroupBase. @@ -76,11 +31,7 @@ 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]))); + this.parallelServiceCall = new ParallelServiceCall(); } @Override @@ -93,14 +44,6 @@ 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()); @@ -116,32 +59,109 @@ public void prepare() { @Override public void execute() throws Exception { DAGNode> nextNode = dag.getNext(); - if (nextNode == null) { - return; + while (nextNode != null) { + nextNode.data().execute(); + this.dag().reportedCompleted(nextNode); + nextNode = dag.getNext(); } - - nextNode.data().execute(this, nextNode); } @Override public ServiceCall executeAsync(final ServiceCallback callback) { - ServiceCall serviceCall = null; + executeReadyTasksAsync(callback); + return parallelServiceCall; + } + + @Override + public T taskResult(String taskId) { + return dag.getNodeData(taskId).result(); + } + + /** + * Executes all runnable tasks, a task is runnable when all the tasks its depends + * on are finished running. + * + * @param callback the callback + */ + private void executeReadyTasksAsync(final ServiceCallback callback) { DAGNode> nextNode = dag.getNext(); while (nextNode != null) { - serviceCall = nextNode.data().executeAsync(this, nextNode, dag.isRootNode(nextNode), callback); - if (serviceCall != null) { - // Filter out the null value returned by executeAsync. that happen - // when TaskItem::executeAsync invokes TaskGroupBase::executeAsync - // but there is no task available in the queue at the moment. - this.serviceCalls.add(serviceCall); - } + ServiceCall serviceCall = nextNode.data().executeAsync(taskCallback(nextNode, callback)); + this.parallelServiceCall.addCall(serviceCall); nextNode = dag.getNext(); } - return serviceCall; } - @Override - public T taskResult(String taskId) { - return dag.getNodeData(taskId).result(); + /** + * This method create and return a callback for the runnable task stored in the given node. + * This callback wraps the given callback. + * + * @param taskNode the node containing runnable task + * @param callback the callback to wrap + * @return the task callback + */ + private ServiceCallback taskCallback(final DAGNode> taskNode, final ServiceCallback callback) { + final TaskGroupBase self = this; + return new ServiceCallback() { + @Override + public void failure(Throwable t) { + callback.failure(t); + } + + @Override + public void success(ServiceResponse result) { + self.dag().reportedCompleted(taskNode); + if (self.dag().isRootNode(taskNode)) { + callback.success(result); + } else { + self.executeReadyTasksAsync(callback); + } + } + }; + } + + /** + * Type represents a set of REST calls running possibly in parallel. + */ + private class ParallelServiceCall extends ServiceCall { + private ConcurrentLinkedQueue serviceCalls; + + /** + * Creates a ParallelServiceCall. + */ + ParallelServiceCall() { + super(null); + this.serviceCalls = new ConcurrentLinkedQueue<>(); + } + + /** + * Cancels all the service calls currently executing. + */ + public void cancel() { + for (ServiceCall call : this.serviceCalls) { + call.cancel(); + } + } + + /** + * @return true if the call has been canceled; false otherwise. + */ + public boolean isCancelled() { + for (ServiceCall call : this.serviceCalls) { + if (!call.isCanceled()) { + return false; + } + } + return true; + } + + /** + * Add a call to the list of parallel calls. + * + * @param call the call + */ + private void addCall(ServiceCall call) { + this.serviceCalls.add(call); + } } } diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/TaskItem.java b/azure-client-runtime/src/main/java/com/microsoft/azure/TaskItem.java index 1612ff4f83..8f0a3459a2 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/TaskItem.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/TaskItem.java @@ -26,22 +26,17 @@ public interface TaskItem { *

* once executed the result will be available through result getter * - * @param taskGroup the task group dispatching tasks - * @param node the node the task item is associated with * @throws Exception exception */ - void execute(TaskGroup> taskGroup, DAGNode> node) throws Exception; + void execute() throws Exception; /** * Executes the task asynchronously. *

* once executed the result will be available through result getter - - * @param taskGroup the task group dispatching tasks - * @param node the node the task item is associated with - * @param isRootNode true if the node is root node + * * @param callback callback to call on success or failure * @return the handle of the REST call */ - ServiceCall executeAsync(TaskGroup> taskGroup, DAGNode> node, final boolean isRootNode, ServiceCallback callback); + ServiceCall executeAsync(ServiceCallback callback); } From afc4173ff06b84e2d1e4367c81481376f53aeec7 Mon Sep 17 00:00:00 2001 From: anuchan Date: Mon, 25 Jul 2016 11:22:00 -0700 Subject: [PATCH 10/10] Adding comments for EdgeType num --- .../java/com/microsoft/azure/DAGraph.java | 5 +- .../main/java/com/microsoft/azure/Graph.java | 90 +++++++++++-------- 2 files changed, 53 insertions(+), 42 deletions(-) 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 df3ba483f6..153c38f259 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 @@ -167,9 +167,8 @@ public void visitNode(U node) { } @Override - public void visitEdge(String fromKey, String toKey, GraphEdgeType edgeType) { - System.out.println("{" + fromKey + ", " + toKey + "} " + edgeType); - if (edgeType == GraphEdgeType.BACK) { + public void visitEdge(String fromKey, String toKey, EdgeType edgeType) { + if (edgeType == EdgeType.BACK) { throw new IllegalStateException("Detected circular dependency: " + findPath(fromKey, toKey)); } } diff --git a/azure-client-runtime/src/main/java/com/microsoft/azure/Graph.java b/azure-client-runtime/src/main/java/com/microsoft/azure/Graph.java index 17440ddbeb..89087dd6d3 100644 --- a/azure-client-runtime/src/main/java/com/microsoft/azure/Graph.java +++ b/azure-client-runtime/src/main/java/com/microsoft/azure/Graph.java @@ -12,16 +12,6 @@ import java.util.Map; import java.util.Set; -/** - * The edge types in a graph. - */ -enum GraphEdgeType { - TREE, - FORWARD, - BACK, - CROSS -} - /** * Type representing a directed graph data structure. *

@@ -61,30 +51,6 @@ public void addNode(U node) { graph.put(node.key(), node); } - /** - * Represents a visitor to be implemented by the consumer who want to visit the - * graph's nodes in DFS order. - * - * @param the type of the node - */ - interface Visitor { - /** - * visit a node. - * - * @param node the node to visited - */ - void visitNode(U node); - - /** - * visit an edge. - * - * @param fromKey key of the from node - * @param toKey key of the to node - * @param graphEdgeType the edge type - */ - void visitEdge(String fromKey, String toKey, GraphEdgeType graphEdgeType); - } - /** * Perform DFS visit in this graph. *

@@ -128,22 +94,22 @@ private void dfs(Visitor visitor, Node node) { processed.add(fromKey); } - private GraphEdgeType edgeType(String fromKey, String toKey) { + private EdgeType edgeType(String fromKey, String toKey) { if (parent.containsKey(toKey) && parent.get(toKey).equals(fromKey)) { - return GraphEdgeType.TREE; + return EdgeType.TREE; } if (visited.contains(toKey) && !processed.contains(toKey)) { - return GraphEdgeType.BACK; + return EdgeType.BACK; } if (processed.contains(toKey) && entryTime.containsKey(toKey) && entryTime.containsKey(fromKey)) { if (entryTime.get(toKey) > entryTime.get(fromKey)) { - return GraphEdgeType.FORWARD; + return EdgeType.FORWARD; } if (entryTime.get(toKey) < entryTime.get(fromKey)) { - return GraphEdgeType.CROSS; + return EdgeType.CROSS; } } @@ -157,4 +123,50 @@ protected String findPath(String start, String end) { return findPath(start, parent.get(end)) + " -> " + end; } } + + /** + * The edge types in a graph. + */ + enum EdgeType { + /** + * An edge (u, v) is a tree edge if v is visited the first time. + */ + TREE, + /** + * An edge (u, v) is a forward edge if v is descendant of u. + */ + FORWARD, + /** + * An edge (u, v) is a back edge if v is ancestor of u. + */ + BACK, + /** + * An edge (u, v) is a cross edge if v is neither ancestor or descendant of u. + */ + CROSS + } + + /** + * Represents a visitor to be implemented by the consumer who want to visit the + * graph's nodes in DFS order by calling visit method. + * + * @param the type of the node + */ + interface Visitor { + /** + * visit a node. + * + * @param node the node to visited + */ + void visitNode(U node); + + /** + * visit an edge. + * + * @param fromKey key of the from node + * @param toKey key of the to node + * @param edgeType the edge type + */ + void visitEdge(String fromKey, String toKey, EdgeType edgeType); + } }