call resubmitTask() when re-schedule task

javeme · javeme · commit 9d81eda8952e · 2021-02-20T19:36:23.000+08:00
Change-Id: I2efa9fd4979492c6901e1ae11fbaf2be451f0354
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTask.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/task/HugeTask.java
@@ -382,11 +382,12 @@ protected boolean checkDependenciesSuccess() {
         if (this.dependencies == null || this.dependencies.isEmpty()) {
             return true;
         }
+        TaskScheduler scheduler = this.scheduler();
         for (Id dependency : this.dependencies) {
-            HugeTask<?> task = this.scheduler().task(dependency);
+            HugeTask<?> task = scheduler.task(dependency);
             if (!task.completed()) {
                 // Dependent task not completed, re-schedule self
-                this.scheduler().schedule(this);
+                scheduler.schedule(this);
                 return false;
             } else if (task.status() == TaskStatus.CANCELLED) {
                 this.result(TaskStatus.CANCELLED, String.format(
diff --git a/hugegraph-core/src/main/java/com/baidu/hugegraph/task/StandardTaskScheduler.java b/hugegraph-core/src/main/java/com/baidu/hugegraph/task/StandardTaskScheduler.java
@@ -205,6 +205,15 @@ private <V> Future<?> restore(HugeTask<V> task) {
     public <V> Future<?> schedule(HugeTask<V> task) {
         E.checkArgumentNotNull(task, "Task can't be null");
 
+        if (task.status() == TaskStatus.QUEUED) {
+            /*
+             * Just submit to queue if status=QUEUED (means re-schedule task)
+             * NOTE: schedule() method may be called multi times by
+             * HugeTask.checkDependenciesSuccess() method
+             */
+            return this.resubmitTask(task);
+        }
+
         if (task.callable() instanceof EphemeralJob) {
             /*
              * Due to EphemeralJob won't be serialized and deserialized through
@@ -252,6 +261,16 @@ private <V> Future<?> submitTask(HugeTask<V> task) {
         return this.taskExecutor.submit(task);
     }
 
+    private <V> Future<?> resubmitTask(HugeTask<V> task) {
+        E.checkArgument(task.status() == TaskStatus.QUEUED,
+                        "Can't resubmit task '%s' with status %s",
+                        task.id(), TaskStatus.QUEUED);
+        E.checkArgument(this.tasks.containsKey(task.id()),
+                        "Can't resubmit task '%s' not been submitted before",
+                        task.id());
+        return this.taskExecutor.submit(task);
+    }
+
     public <V> void initTaskCallable(HugeTask<V> task) {
         task.scheduler(this);
 
