Workflow for running a DAG in DSL

Community Support
1

We have a DSL for DAG workflow which is defined from UI. Users can create workflows similar to the one in below image.
Sample_DAG (1)
Sample JSON definition of Task 5 from above DAG. It has references to both parent and children tasks.

TaskDefinition
{
  "task": "Task 5",
  "parentTasks": [
    "Task 1"
    "Task 2"
  ],
  "childrenTasks": [
    "Task 7"
  ]
}

Workflow implementation to run the DSL workflow.

public class DslWorkflowImpl implements DslWorkflow {

  private TaskActivity taskActivity = getActivityStub();
  private Map<String, Object> resultMap; // maintains the return object of an activity
  private Map<String, TaskDefinition> taskMap;  // DSL of the workflow

  @Override
  public String startDslWorkflow(String startTask, Map<String, TaskDefinition> taskMap) {
    resultMap = new HashMap<>();
    this.taskMap = taskMap;
    Promise<Object> promise =
            Async.function(taskActivity::executeTask, taskMap.get(startTask));
    promise.handle(getHandleFn(startTask));
    promise.get();
    return "SUCCESS";
  }

  private Functions.Func2<Object, RuntimeException, Object> getHandleFn(String task) {
    return (taskResult, e) -> {
      resultMap.put(task, taskResult);
      List<Promise<Object>> promiseList = triggerChildTasks(taskResult, task);
      Promise.allOf(promiseList).get();
      return taskResult;
    };
  }

  private List<Promise<Object>> triggerChildTasks(String task) {
    List<String> childrenTasks = taskMap.get(task).getChildrenTasks();
    List<Promise<Object>> promiseList = childrenTasks.stream().map(childTask -> {
      TaskDefinition childTaskDefinition = taskMap.get(childTask);
      // Check if all the parent tasks are completed. Only then start this task.
      if (childTaskDefinition.getParentTasks().stream().allMatch(parent -> resultMap.containsKey(parent))) {
        Promise<Object> promise =
                Async.function(taskActivity::executeTask, childTaskDefinition);
        // Calling getHandleFn recursively here.
        promise = promise.handle(getHandleFn(childTask));
        return promise;
      } else {
        log.info("Parent tasks are not yet done. task:" + childTask);
        return null;
      }
    }).filter(Objects::nonNull).collect(Collectors.toList());
    return promiseList;
  }
}

I am mainly concerned about the case where one task has multiple parent tasks and it needs to be started only when all parents are completed. What if two parent tasks are completed at the same time, then temporal workflow threads will try to invoke the getHandleFn at the same time and what if both the threads fails to run the child task. Is there any way to synchronise the workflow code execution using locks?
So far above code seems to be working fine. I have never run into this issue after testing the scenario hundreds of times. But I am worried about it as it is like a race condition.
Also please advice if this is the correct approach or needs any improvement.

Thanks,
Swaroop

2

promise.handle(getHandleFn(startTask));

handle is called after promise is finished so getHandleFn can do something with its result. Note that it itself returns a Promise so one thing I would look into is if you need to call .get() on that promise to wait until getHandleFn is fully completed.

In getHandleFn you do wait for all promises to complete with Promise.allOf(promiseList).get(); so I don’t think you have to worry about the race condition you are talking about,
but pinging @spikhalskiy in case I am missing something here.

3

All workflow threads of a single workflow execution are serialized. Workflow threads of the same workflow NEVER run at the same time.
JavaSDK utilizes an idea of cooperative multithreading for that (a thread needs to block on one of the Temporal blocking operations (like a blocking activity call or Workflow.sleep) to let other threads an opportunity to advance)
So, the first getHandleFn(childTask) will get executed until it gets blocked (or workflow thread will finish), after that the second getHandleFn(childTask) one will get executed. Two handlers will never be run truly in parallel.

Is there any way to synchronize the workflow code execution using locks?

Using parallel threads and synchronization primitives are prohibited in workflow code for a reason, workflow threads are always synchronized and ordered according to the workflow history.
Using threads inside workflow code will lead to non-deterministic problems.
Using locks is either harmless, but pointless or will lead to a deadlock (depends on a specific scenario).

4

@tihomir I do have a .get() call on the promise in startDslWorkflow method. I missed it while posting here. Fixed the code now.

5

promise.handle(getHandleFn(startTask)); returns a promise that contains the result of your getHandleFn function, so if you need to wait for getHandleFn to complete before you return from your workflow method, then yes.