How to execute wait state and tasks parallelly in Temporal

Community Support
1

Team, Below image is my sample workflow

sample-workflow
sample-workflow339×1280 31.8 KB

Green - Tasks (it might have an API call, function call, or webhook)
Yellow - Wait state

In this diagram, the red color marked portion needs to be executed parallelly but among that we need path1, Path2, and Path3 to be executed in linear fashion.

T2 has to start its execution after 9 states are completed. So it has to wait for all 3 paths to complete.

Question 1) Is it possible to put wait ( Workflow.sleep(some duration) ) for 3 states at a time?

I can achieve parallel execution using Async, but using .get on the promise object will block my thread.

Question 2) Is it possible to achieve without blocking?

I tried using thenApply,handle on the promise objects but my workflow code has completed without executing this thenAppy and handle callbacks.

Awaiting your response, Thanks

2

Question 1) Is it possible to put wait ( Workflow.sleep(some duration) ) for 3 states at a time?
I can achieve parallel execution using Async, but using .get on the promise object will block my thread.

Yes, you can use Async to start three threads and use Workflow.sleep in each of them. Then you use Promise.allOf on the Promises returned by the Async.<…> function to join the threads.

Question 2) Is it possible to achieve without blocking?

Yes, you can implement these branches asynchronously. You still have to block the main workflow thread as fully asynchronous workflow implementation is not supported yet.

3

you are referring to this way, right?

for(Task task :  parallelFlow.getTaskList() ) {
   if (task instanceof FunctionalTask) {
         pilotIO.getUserData().setTask((FunctionalTask) task).setPrevResult(prevResult);
         Promise<PilotIO> promise = Async.function(runnableTask::run, pilotIO);
         promiseList.add(promise);
    }
}

Promise.allOf(promiseList).get();

for(Promise promise : promiseList){
   System.out.println(((PilotIO) promise.get()).getUserData().getPrevResult());
}

I hope using Promise.allOf(promiseList).get(); and Promise.allOf(promiseList); makes no difference. Correct me if i am wrong

So as you said, in this way main workflow thread will be blocked until it completes all its promises as full asynchronous workflow implementation is not supported yet.

4

I hope using Promise.allOf(promiseList).get(); and Promise.allOf(promiseList); makes no difference. Correct me if i am wrong

Correcting :). Promise.allOf(promiseList) returns a promise. So

Promise.allOf(promiseList);

creates a promise and ignores it. So it is practically NOOP.

You want to wait on this promise to join:

Promise.allOf(promiseList).get();
5

Oops, I tried both, both works fine, that’s why I came for that conclusion. Thanks for addressing this.

All other my understanding are correct?

6

The reason that it worked was that you didn’t need Promise.allOf in your case at all, as the following loop called get on every Promise anyway. So NOOP was fine :).

7

Yes, I just noticed the promise.get() called in the loop.

Promise.allOf(promiseList).get();

After the above line of code, how to get my list of values?

8

By calling them in the loop :slight_smile: and populating the result collection.

9

Sorry, seriously I don’t know what you’re referring to.
Like below, you want me to iterate ?

for(Promise promise : promiseList){
System.out.println(((PilotIO) promise.get()).getUserData().getPrevResult());
}

I don’t know any other options. Kindly tell me how to loop it :anguished:

10

Yes, your loop code is fine. I think the cast to (PilotIO) type is unnecessary as Promise is already generic with this type.

1 Like
11

No, my use case not solved.

It is either waits for all threads to complete or it waits for any of the threads to complete. (allOf, anyOf)

If you see the diagram,
I will be calling all the three tasks parallelly using Async which are,

  1. Path1-T1
  2. Path2-T1
  3. Path3-T1

Say, if Path1-T1 takes 2 mins and Path2-T1 and Path3-T1 take just 3 sec to complete.
My case is after completing Path2-T1 it should call Path2Wait (5 min) and after completing Path3-T1 it should call Path3-T2.

if I use allOf then all my threads will wait for each other to complete.
If I use anyOf then my first completed thread returning its result and workflow cmpleted
If I use normal loop .get() on the promise object then the first promise.get() will block before others to continue.

@maxim kindly give your feedback!!

12

Don’t call individual tasks in async mode. Call each branch in async mode.
Something like:


  void branch1() {
    activities.task11();
    Workflow.sleep(Duration.ofMinutes(1));
    activities.task12();
  }
  void branch2() {
    activities.task21();
    Workflow.sleep(Duration.ofMinutes(5));
    activities.task22();
  }
  void branch3() {
    activities.task31();
    activities.task32();
    Workflow.sleep(Duration.ofHours(2));
  }

...
    Promise<Void> b1 = Async.procedure(this::branch1);
    Promise<Void> b2 = Async.procedure(this::branch2);
    Promise<Void> b3 = Async.procedure(this::branch3);
    Promise.anyOf(b1, b2, b3);
1 Like
13

Thank you so much @maxim for your response,

I hope Workflow.sleep(Duration.ofHours(2)) will not hold a thread,

will it be handled similar to the scheduler model where at a constant interval a single thread will fetch DB and process the timer which is completed?

14

Yes, it is not going to hold the thread. You absolutely can have more workflows waiting than your worker has threads.

1 Like
15

Thanks for the response

When to use .thenAppply, .handle callbacks in promise objects.

because if we use these callbacks then the workflow has completed before executing these callbacks

16

As soon as the main workflow method (annotated with @WorkflowMethod) returns workflow completes. So make sure that you block the main workflow thread until all the work of the workflow is done.

17

In my context, I have to block my main workflow thread until it execute the call back of my promise object

18

@maxim Is there any plan in future to implement workflow code fully asynchronous?

19

Yes, but why do you care? It is a purely caching optimization.

20

I care as it blocks a thread in my app server. It will slow down the execution right?