nav[aria-label="Primary Navigation"] { padding: 0; & ul { list-style: none; width: 100%; display: flex; flex-direction: row; justify-content: start; align-items: start; gap: 30px; padding: 0; & li { margin: 0; } & ul li { list-style: none; } } }

As we transition to the new Hyland Community, some post links may be temporarily disrupted as we finalize redirects. We apologize for the inconvenience and are working to resolve this as quickly as possible. Details.

Running parallel tasks with acitiviti

hellowrakesh123

Hi,
I am new with Activiti and trying to get the parallel gateways work but i couldn't get it working the way I was expecting. Am I doing something wrong here?

I have 2 tasks Build-1 and Build-2, I want to fork off the 2 tasks at the same time as soon as the workflow starts.

When i am running the workflow, its first starting the first process (Build-1) and after it completes, it starts the other (Build-2). I am not sure what is wrong here. I tried using activiti:async but it didnt work for me.

<process id="ccaProcess" name="ccaProcess">
      <startEvent id="startevent1" name="Start"></startEvent>
       <serviceTask id="servicetask1" name="Build-1" activiti:class="com.mycomp.step.cca.bpm.CcaDelegate"></serviceTask>
       <serviceTask id="servicetask2" name="Build-2" activiti:class="com.mycomp.step.cca.bpm.CcaDelegate"></serviceTask>
       <parallelGateway id="parallelgateway2" name="Parallel Gateway"></parallelGateway>
       <endEvent id="endevent1" name="End"></endEvent>
       <sequenceFlow id="flow4" name="" sourceRef="servicetask1" targetRef="parallelgateway2"></sequenceFlow>
       <sequenceFlow id="flow5" name="" sourceRef="servicetask2" targetRef="parallelgateway2"></sequenceFlow>
       <sequenceFlow id="flow6" name="" sourceRef="parallelgateway2" targetRef="endevent1"></sequenceFlow>
       <parallelGateway id="parallelgateway3" name="Parallel Gateway"></parallelGateway>
       <sequenceFlow id="flow8" name="" sourceRef="parallelgateway3" targetRef="servicetask1"></sequenceFlow>
       <sequenceFlow id="flow9" name="" sourceRef="parallelgateway3" targetRef="servicetask2"></sequenceFlow>
       <sequenceFlow id="flow10" name="" sourceRef="startevent1" targetRef="parallelgateway3"></sequenceFlow>
   </process>‍‍‍‍‍‍‍‍‍‍‍‍‍‍‍

Here is the java code i used to start the process:

runtimeService.startProcessInstanceByKey("ccaProcess");‍

– Assume runtimeService is a spring managed bean.
– The Delegates implement JavaDelegate.

Find more posts tagged with

Comments

frederikherema1

It seems like you can't to have the 2 service-tasks being executed in parallel?

The parallel gateway in BPMN indicates the path of execution splits into several paths. Dis doesn't necessarily mean they should be executed by different threads. When you call the startProcessInstance(), the calling thread will execute the workflow until a wait state is reached. When reaching a parallel gateway, there are two "concurrent execution paths" created and each of them is executed until it reaches a wait state/end or parallel join. This is done by THE SAME THREAD, so the paths are executed sequentially internally. However, when the startProcessInstance() returns, all parallel executions are executed and waiting/finished. So they are actually executed in parallel in the same API-call but not parallel as in multi-threaded.

If you want to execute the 2 service-tasks in parallel, they should both have "async=true".

hellowrakesh123

Thank you for the post. This works but I have another problem. I am running activiti as a standalone application which is forked off using a command line. Now the problem with async=true is that, my program exits as soon as the activiti process is started. Is there a way to block until the whole process is done.
I can work around this problem by starting it in a thread and waiting till the process moves to the history but it would be great if i can get an api or more robust solution/way of doing it.?
Thank you.!
Rakesh

frederikherema1

Most of the time, your application will run in some kind of environment (web server, osgi, ..) which is not just a simple main-thread and creates at least one non-deamon thread that keeps the app alive. You should google for solutions around this, it's a common java question.

hellowrakesh123

Hi,
I tried running the same application in tomcat but still can't get the acitiviti:async="true" working. Here is the bpm definition I have, when i set acitiviti:async="true", nothing gets executed, when i remove acitiviti:async="true", it gets executed but all in sequential order. Am i missing something here.

<definitions xmlns="<a href="http://www.omg.org/spec/BPMN/20100524/MODEL" rel="nofollow noopener noreferrer">http://www.omg.org/spec/BPMN/20100524/MODEL</a>"<br> xmlns:activiti="<a href="http://activiti.org/bpmn" rel="nofollow noopener noreferrer">http://activiti.org/bpmn</a>" targetNamespace="Examples"><br> <process id="cca-workflow-process" name="cca-workflow-process"><br>&nbsp; <documentation>Place documentation for the 'mtrose-bpm-workflow'<br>&nbsp;&nbsp; process here.</documentation><br>&nbsp; <startEvent id="startevent1" name="Start"></startEvent><br>&nbsp; <parallelGateway id="parallelgateway1" name="Parallel Gateway"></parallelGateway><br>&nbsp; <serviceTask id="build" name="build"<br>&nbsp;&nbsp; activiti:class="com.cisco.step.cca.engine.delegate.BuildTaskDelegate" activiti:async="true"></serviceTask><br>&nbsp; <parallelGateway id="parallelgateway2" name="Parallel Gateway"></parallelGateway><br>&nbsp; <serviceTask id="hss_sanity_c4" name="hss_sanity_c4"<br>&nbsp;&nbsp; activiti:class="com.cisco.step.cca.engine.delegate.SanityTaskDelegate"></serviceTask><br>&nbsp; <serviceTask id="sa" name="sa"<br>&nbsp;&nbsp; activiti:class="com.cisco.step.cca.engine.delegate.StaticAnalysisTaskDelegate"></serviceTask><br>&nbsp; <serviceTask id="cr" name="cr"<br>&nbsp;&nbsp; activiti:class="com.cisco.step.cca.engine.delegate.CodeReviewTaskDelegate" activiti:async="true"></serviceTask><br>&nbsp; <serviceTask id="ut" name="ut"<br>&nbsp;&nbsp; activiti:class="com.cisco.step.cca.engine.delegate.UnitTestTaskDelegate" activiti:async="true"></serviceTask><br>&nbsp; <sequenceFlow id="flow1" name="" sourceRef="startevent1"<br>&nbsp;&nbsp; targetRef="parallelgateway1"></sequenceFlow><br>&nbsp; <parallelGateway id="parallelgateway3" name="Parallel Gateway"></parallelGateway><br>&nbsp; <endEvent id="endevent1" name="End"></endEvent><br>&nbsp; <sequenceFlow id="flow5" name="" sourceRef="build"<br>&nbsp;&nbsp; targetRef="parallelgateway2"></sequenceFlow><br>&nbsp; <sequenceFlow id="flow6" name="" sourceRef="parallelgateway2"<br>&nbsp;&nbsp; targetRef="hss_sanity_c4"></sequenceFlow><br>&nbsp; <sequenceFlow id="flow7" name="" sourceRef="parallelgateway2"<br>&nbsp;&nbsp; targetRef="sa"></sequenceFlow><br>&nbsp; <parallelGateway id="parallelgateway4" name="Parallel Gateway"></parallelGateway><br>&nbsp; <sequenceFlow id="flow10" name="" sourceRef="parallelgateway3"<br>&nbsp;&nbsp; targetRef="parallelgateway4"></sequenceFlow><br>&nbsp; <sequenceFlow id="flow11" name="" sourceRef="cr"<br>&nbsp;&nbsp; targetRef="parallelgateway4"></sequenceFlow><br>&nbsp; <sequenceFlow id="flow12" name="" sourceRef="ut"<br>&nbsp;&nbsp; targetRef="parallelgateway4"></sequenceFlow><br>&nbsp; <sequenceFlow id="flow13" name="" sourceRef="parallelgateway4"<br>&nbsp;&nbsp; targetRef="endevent1"></sequenceFlow><br>&nbsp; <sequenceFlow id="flow14" name="" sourceRef="parallelgateway1"<br>&nbsp;&nbsp; targetRef="build"></sequenceFlow><br>&nbsp; <sequenceFlow id="flow15" name="" sourceRef="parallelgateway1"<br>&nbsp;&nbsp; targetRef="cr"></sequenceFlow><br>&nbsp; <sequenceFlow id="flow16" name="" sourceRef="parallelgateway1"<br>&nbsp;&nbsp; targetRef="ut"></sequenceFlow><br>&nbsp; <receiveTask id="sanity_response" name="sanity_response" activiti:class="com.cisco.step.cca.engine.delegate.SanityStatusTaskDelegate"></receiveTask><br>&nbsp; <sequenceFlow id="flow17" name="" sourceRef="hss_sanity_c4"<br>&nbsp;&nbsp; targetRef="sanity_response"></sequenceFlow><br>&nbsp; <sequenceFlow id="flow18" name="" sourceRef="sanity_response"<br>&nbsp;&nbsp; targetRef="parallelgateway3"></sequenceFlow><br>&nbsp; <sequenceFlow id="flow19" name="" sourceRef="sa"<br>&nbsp;&nbsp; targetRef="parallelgateway3"></sequenceFlow><br> </process><br></definitions>

My delegates are implemented using JavaDelegate:

public class BuildTaskDelegate implements JavaDelegate {<br> <br> public void execute(DelegateExecution execution) throws Exception {<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; System.out.println("before&hellip;");<br>&nbsp; Process process=Runtime.getRuntime().exec("sleep 10");<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; System.out.println("after&hellip;");<br> }<br>}<br>

frederikherema1

Make sure your job-executor is active. You can verify this by checking your engine-settings. The jobExecutorActivate should be true. Alternatively, you can double-check in code by getting the processEngineConfiguration-object from the ProcessEngine (cast to IMPL may be needed) en call getJobExecutor().isActive().

hellowrakesh123

ohh so thats the issue, it was set as false, it worked as I changed the jobExecutorActivate=true.
Thank you so much for the help.
May be this is something you might want to add to the documentation where you explained async.

frederikherema1

Glad it helped

The section on the async states the job executor is used for this:

<br>This background thread is the activiti job executor (actually a thread pool) which periodically polls the database for jobs. So behind the scenes, when we reach the "generate invoice" task, we are creating a job "message" for activiti to continue the process later and persisting it into the database. This job is then picked up by the job executor and executed. We are also giving the local job executor a little hint that there is a new job, to improve performance.<br>

hellowrakesh123

Hi,
Another issue, although i read about this but how do we handle this (i mean how to make sure it doesn't happen). With async=true and exclusiveJob=false, it is working as expected but now i am seeing exceptions:
org.activiti.engine.ActivitiOptimisticLockingException: ExecutionEntity[4] was updated by another transaction concurrently

This happens when more than one task tries to update the parallel join nodes. If it fails, it tries again and run the task again. This is undesirable and how should i make sure that it doesn't get executed again.
I have a dirty solution, create one more task in front of each parallel task which doesn't do anything but just passes by to make sure even if it is executed, it doesn't affect anything.
Thanks in advance.!
Rakesh

frederikherema1

When both tasks are "async", they both get acquired by the job-executor at the same time (since it has multiple threads available) and actually executed concurrently. When the two paths reach the join, one of them will get an exception, trying to update the shared execution that forked off the 2 parallel. The job-executor will retry a job for 3 times until it stops.

I'm afraid this is just how the engine works. Concurrent updates of the same entity in different transactions can't be handled differently, smart merging or updating can cause to stale state or even data-loss so that's not desirable.

What you can do to have minimal effect of this is trying to make sure the service YOU call in the service-task use the same transaction, so the rollback by the failing execution cleans up any work performed. If this is not possible (other DB, non-transactional operations,…) you should consider something like this:

(read full post at http://forums.activiti.org/en/viewtopic.php?f=6&t=4139&p=16108&hilit=exeternal%20thread%20pool)
The service-task prepares (eg. read variables from process) an creates the "job" or runnable to be executed by the external thread poolService-task queues the job (saving reference to process-instance id) and finishesRight after the service-taks, there's a receive-task that waits, releasing the calling thead and not holding open the transaction.Long-running operation completes and signals the process-instance and perhaps set some variables on the process based on the results before signaling.
Also interesting info about signaling/jobexecutor:
http://forums.activiti.org/en/viewtopic.php?f=6&t=4138
http://forums.activiti.org/en/viewtopic.php?f=6&t=4165&hilit=signal

hellowrakesh123

Thank you for the response. I read the threads and have managed to make the workflow work with receive tasks. However, it has invoked a doubt regarding the way workflow will serve, please correct me if my understanding is wrong.

BPMN specification defines parallel workflows which is implemented in Activiti as well. Going over the documentation and suggestions by you, the *real* parallel execution which runs tasks concurrently really isn't there by default due to the reasons mentioned in documentation and by you as well. This means even we express the workflow via parallel gateways it is really sequential. Receive tasks do provide us a way to achieve this but this limits us to use the normal service tasks. So even if the tasks are not really long running but also non db, we have to choose receive task which doesn't look right. Also with receive task, there is a problem if two tasks which run in parallel joins nodes at the same time, it still has locking issue but now we can manually synchronize it to make sure it doesn't hit the issue during signal. So the problem is same but it works using receive tasks because we know when to synchronize which we can't do with normal service task. The way I solve is to lock a common monitor in the thread spawned by the service tasks which signals the receive tasks. I acquire the class lock and makes sure all threads wait for the lock to be acquired before signal.

So here are my few questions:
1) Is there is way in service tasks to synchronize using our code (override any method) similar to common monitor.? This will solve the problem with async=true and exclusiveJob=false and allow real parallel paths concurrently.
2) Isn't it something which activiti/engine should do by default or is there is something i missing to understand how it is implemented.? This is a common problem and going for a workaround using receive tasks doesn't sound very convincing although its a simple solution. The problem with the approach you suggested is that, the first service task is just a dummy tasks which queue the request. The other problem with that approach is the service tasks is required to know the proceeding receive task which means i can't have dynamically created workflows else i will have to put some real code to make service task aware of the proceeding receive task. Infact we can work without service tasks and have the receive task do it extending ReceiveTaskActivityBehavior and starting a thread from it. Is there a specific reason for using service task before receive task or its only for queuing or i am missing anything here.

I am looking at all the aspects and trying to understand if i am making the right choice.
Thanks in advance.!
Rakesh

frederikherema1

I suggest reading this thread, as it's a similar discussion…

http://forums.activiti.org/en/viewtopic.php?f=6&t=4138&start=20

hellowrakesh123

[img]Hi,
I read the discussion and understand the suggestions. My question is more on how to make sure *pure* service task synchronize on a monitor before completing. You mentioned following example to have mutex logic. How can this be implemented in the delegate analogous to signal event in receive tasks. The code you gave requires me to write logic outside my delegate. This will cause me to have flow control logic outside of delegate which will not be scalable in our case. We are dynamically generating the workflow and running it. The few tasks are long running ones where we are using receive tasks and works as we discussed. For the tasks which are not long running and implement using JavaDelegate (another web service call but are non-db transactional i.e. can't be rolled back), having parallel gateway causes tasks to be re-executed with OptimisticLockException. I want to make sure when they join, i implicitly synchronize complete() on a common monitor either in delegate or somewhere which is generic for all JavaDelegates and can use async=true and exclusiveJob=false for concurrent tasks.

So my requirement is simple, i have set of parallel tasks and dynamic workflow (i don't want to put too much code outside delegate as it will restrict us from generating workflows. Each time the workflow could have different number of tasks for each user). So the logic for locking needs to be generic for service tasks. All this is achieved using Receive Tasks and locking before signal and releasing in finally block and its works quite wel. How do we achieve same for the service tasks else we would be restricted to use Receive Tasks for all our requirements. Also attached the workflow we are using.

public&nbsp; class YourCompanyService {<br><br><br>&nbsp; private Set<String> processLocks = new HashSet<&hellip;>;<br>&hellip;<br><br>public boolean printTheCard(String cardIdentifier) {<br>&nbsp;&nbsp; // Correlate between card-id and task/process to use<br>&nbsp;&nbsp; String processId = runtimeService.createtaskQuery()&hellip;&hellip;singleResult();<br><br>&nbsp;&nbsp; // JVM-wide lock for this specific use-case.<br>&nbsp;&nbsp; boolean isSafe = false;<br>&nbsp;&nbsp; synchronized(this) {<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; isSafe = processLocks.contains(processId);<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; if(isSafe) {<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; processLocks.add(processId);<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }<br>&nbsp;&nbsp; }<br><br>&nbsp;&nbsp; if(!isSafe) {<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; throw new MyCompanyException("looks like you're too late&hellip;");<br>&nbsp;&nbsp; }<br><br>&nbsp;&nbsp; // important, release the lock in "finally"<br>&nbsp;&nbsp; try {<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; // Finish task<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; taskService&hellip;&hellip;<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; &hellip;.<br>&nbsp;&nbsp;&nbsp; } <br>&nbsp;&nbsp;&nbsp; finally<br>&nbsp;&nbsp;&nbsp; {<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; synchronized(this) {<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; processLocks.remove(processId);<br>&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; }<br><br>&nbsp;&nbsp;&nbsp; }<br>}<br><br>}

[/img]

async.png