Workflow.await(), does it release thread?

Community Support
1

Is there any way in temporal to resequence the messages to process them as a group identified by a functional key in the message?

Here is the proposed approach.

Have a single workflow per unique functional key which receives a message (possibly with SignalWithStart) and queues, workflow will complete processing everything in the queue and go to wait state and wait for more messages to process.

The actual processing is implemented as a child workflow or directly as part of the parent workflow.

Sudo code

SampleWorkflow {
    WorkflowQueue<String> queue = Workflow.newQueue(10);
    ChildWorkflowOptions childWorkflowOptions = ChildWorkflowOptions.newBuilder().build();
    @Override
    public void runProcess(String workflowRequest) {
        queue.offer(workflowRequest);
        while (true) {
            workflowRequest = queue.poll();
            SampleChildWorkflow child = Workflow.newChildWorkflowStub(SampleChildWorkflow.class, childWorkflowOptions);
            child.runChildProcess(workflowRequest);
            if (null == queue.peek()) {
                Workflow.await(() -> (null != queue.peek()));
            }
        }
    }
    @Override
    public void addEntry(String workflowRequest) {
        this.queue.offer(workflowRequest);
    }
}

Once after I finish processing all the events in the queue,

I am moving the workflow to await state until it receive a new request using getEntry-Signal.

Any new entry will be processed by coming out from the wait state. Once after processing all the events in the WorkflowQueue, the workflow will await for new requests to be added to the queue.

When I checked the logs, I see that the workflow thread ID is same, every time a new request come and the workflow start running.
Does it mean that this thread is not release when it go to await state?

We are planning to run a million instances of such workflows (not always active) and would like to understand the performance and memory impact.

5 Likes
2

The proposed approach is fine. Just some nits:

  • WorkflowQueue is a blocking queue.
  • WorkflowQueue.take blocks until a message is available.
  • To support graceful cancellation consider using WorkflowQueue.cancellableTake.
  • Don’t use the offer without checking the result as this is going to lose a message if the queue is full.
  • I would use an unbounded queue and deal with the message shedding in the consumer code if necessary.
  • Running such workflow infinitely is going to grow its history to an unbounded size. To keep the history size bounded the workflow has to call continue as new periodically.
  • To not lose messages when calling continue as new the buffered messages have to be passed from one run to another as a workflow argument.

So I would rewrite your code to:

SampleWorkflow {

    WorkflowQueue<String> queue = Workflow.newQueue();
    ChildWorkflowOptions childWorkflowOptions = ChildWorkflowOptions.newBuilder().build();

    @Override
    public void runProcess(WorkflowQueue<String> buffered) {
        while(true) {
             String request =  buffered.peek();
             if (request == null) {
                 break;
             }
            queue.put(request); 
         }
         for(int i=0; i<CONTINUE_AS_NEW_FREQUENCY; i++) {
             workflowRequest = queue.take();
             SampleChildWorkflow child = Workflow.newChildWorkflowStub(SampleChildWorkflow.class, childWorkflowOptions);
             child.runChildProcess(workflowRequest);
          }
          // Continue as new
          SampleWorkflow next = Workflow.newContinueAsNewStub(SampleWorkflow.class);
          next.runProcess(queue);
     }

     @Override
     public void addEntry(String workflowRequest) {
          this.queue.put(workflowRequest);
     }
}

This is a pretty frequently used pattern. I’ll try to write a full sample that demonstrates it at some point.

5 Likes
3

Great inputs @maxim. Thank You.

4

Hi regarding the original question does the await() blocks the thread ? Or it’s suspends the flow and later recover it from the the event source ?

We plan to use it as a batch aggregator, a signal will add to the batch and as soon the batch will meet size or time window requirements the messages will be processed by a child flow.

The question, is it better to use await or the timer sleep call, in order not to block the executing thread and cause performance degradation. Please advise ?

Thank you.

5

Hi regarding the original question does the await() blocks the thread ? Or it’s suspends the flow and later recover it from the the event source ?

It suspends the workflow execution and later recovers it. So a workflow that is blocked for a long time doesn’t consume resources on the worker.

We plan to use it as a batch aggregator, a signal will add to the batch and as soon the batch will meet size or time window requirements the messages will be processed by a child flow.

Keep in mind that a single workflow instance cannot grow very large or process a large update rate. What is the batch size and number of signals per second processed by a single workflow execution (instance)?