jBPM Multithreading: Is there a specification?

[archer77]

Is jBPM working in a real multithreaded environment?

Is there a specification how the thread interaction should work?

I looked for information on both topic but did not find any.

It seems that the current design/implementation allows to have two different representations of the same task, both containing different data.


My problem:

A process definition will make the jBPM generate an event on task-assign. When executed following happens:

* Thread A ends the previous task, jBPM continues execution.
* jBPM creates, assigns next task, triggers the event handler (also Thread A)
* The eventhandler dispatches a message to the actor of the new task

* The actor receives the message (other host, Thread B)
* Connects to the jBPM (Thread B) and tries to fetch the task information
* Depending on the transport speed, thread B will be so fast, that he asks for the task when it is
** not yet assigned to him (causes not authorised exception)
** already there (everything ok)

The same game can be played using the task-create event, result:
* Task does not exist (error)
* Task already there (ok)

I added a loop around the getRemoteTaskInstanceInfo call because I know the Id of the task from the message, so I can wait for it to appear.

Disadvantage: The code gets really ugly because every multithreaded jBPM interaction has to retry an operation until it succeeds. Apart from that the workaround causes high server load (polling) or inefficient timing (wait).

[tom.baeyens]

in a persistent environment, jBPM synchronizes on the database. and you'll get your concurrency from your environment (ejbs, mdbs, web app and the likes) note that when moving transactional from one wait state to the next, jBPM synchronizes on the DB, so transactions should be handled concurrently, but inside one transaction, no multi threading is needed.

regards, tom.

[estaub]

Tom,

I think this is another case, like the one a month or so ago with ProcessInstances not existing when needed that I tried to help with, where additional database commits are necessary.

In this case, the commit must be somewhere in the flow of execution between:

In TaskInstance:

public void setActorId(String actorId, boolean overwriteSwimlane){
 // do the actual assignment
 this.previousActorId = this.actorId;
 this.actorId = actorId;
 if ( (swimlaneInstance!=null)
 && (overwriteSwimlane) ) {
 log.debug("assigning task '"+name+"' to '"+actorId+"'");
 swimlaneInstance.setActorId(actorId);
 }
COMMIT AFTER HERE...

and in GraphElement:

void executeActions(List actions, ExecutionContext executionContext, boolean isPropagated) {
 if (actions!=null) {
 Iterator iter = actions.iterator();
 while (iter.hasNext()) {
 Action action = (Action) iter.next();
 if ( action.acceptsPropagatedEvents()
 || (!isPropagated)
 ) {
 if (action.isAsync()) {
 Message continuationMsg = new ExecuteActionCommand(action, executionContext.getToken());
 MessageService messageService
 = (MessageService) Services.getCurrentService(Services.SERVICENAME_MESSAGE);
COMMIT BEFORE HERE:
 messageService.send(continuationMsg);

Because the next instruction executed after the send may be in the other thread - you haven't committed yet, I don't think.

An efficient implementation would be to keep an internal queue of messages to send after you've committed, before you enter a wait state. This would avoid any additional commits, and I think it would preserve the correct semantics.

I easily may be overlooking something - I don't know this code anywhere near as well as you, of course!

-Ed Staub

[archer77]

in a persistent environment, jBPM synchronizes on the database.

What behaviour should I expect in this case? I would expect that a select on a table where a write-transaction from another thread is open should block until the transaction is ended (It might also throw something like a ConcurrentAccessException to indicate the access conflict).

Looking at my problem, this should cause the second thread to block until the first transaction is commited, but instead second thread is not block (or receives an "concurrent access exception"), which I would expect, instead it returns data (directly from the DB?) which conflicts with data held in another thread (out of sync data).

you'll get your concurrency from your environment (ejbs, mdbs, web app and the likes) note that when moving transactional from one wait state to the next, jBPM synchronizes on the DB, so transactions should be handled concurrently, but inside one transaction, no multi threading is needed.

Does that mean that the jBPM keeps one transaction open from ending one task until wait() in the next task? Will an error in task-assign also rollback the task-end (node-leave) of the previous node?

And when is the task-assign event handling triggered?

I assumed that the first transaction (Thread A) should be ended before sending the events (because the event handlers already receive a fixed task assignment). The event actions then would have to create more transactions if they need them, but the task instance assignment


BTW: My setup is a standalone jBPM (with in-memory sql database), no ejbs, jboss container, ..

[tom.baeyens]

"estaub" wrote:
Tom,

I think this is another case, like the one a month or so ago with ProcessInstances not existing when needed that I tried to help with, where additional database commits are necessary.

In this case, the commit must be somewhere in the flow of execution between:

In TaskInstance:

public void setActorId(String actorId, boolean overwriteSwimlane){
 // do the actual assignment
 this.previousActorId = this.actorId;
 this.actorId = actorId;
 if ( (swimlaneInstance!=null)
 && (overwriteSwimlane) ) {
 log.debug("assigning task '"+name+"' to '"+actorId+"'");
 swimlaneInstance.setActorId(actorId);
 }
COMMIT AFTER HERE...

and in GraphElement:

void executeActions(List actions, ExecutionContext executionContext, boolean isPropagated) {
 if (actions!=null) {
 Iterator iter = actions.iterator();
 while (iter.hasNext()) {
 Action action = (Action) iter.next();
 if ( action.acceptsPropagatedEvents()
 || (!isPropagated)
 ) {
 if (action.isAsync()) {
 Message continuationMsg = new ExecuteActionCommand(action, executionContext.getToken());
 MessageService messageService
 = (MessageService) Services.getCurrentService(Services.SERVICENAME_MESSAGE);
COMMIT BEFORE HERE:
 messageService.send(continuationMsg);

Because the next instruction executed after the send may be in the other thread - you haven't committed yet, I don't think.

An efficient implementation would be to keep an internal queue of messages to send after you've committed, before you enter a wait state. This would avoid any additional commits, and I think it would preserve the correct semantics.

I easily may be overlooking something - I don't know this code anywhere near as well as you, of course!

-Ed Staub

the idea is that there should never be a persistence operation during process execution.

always, when you are in a wait state, that is when you can save your process.

regarding the async message, you are right if you use the default JMS configurations. in which case, the message sending is committed before the send method returns. but you should use/configure your async message service so that it takes part in the same transaction as the jbpm updates. in the POJO configuration of jbpm, this is done by making sure that the message service operates on the same JDBC connection as the jbpm persistence sessions.

i don't get the task instance example you give.

[tom.baeyens]

"archer77" wrote:

BTW: My setup is a standalone jBPM (with in-memory sql database), no ejbs, jboss container, ..

a transient usage of jBPM is a whole other story.

jBPM is usable in transient mode. you don't even have to use an in memory db in that case. you can just parse your process definitions and create new instances of ProcessInstance for it, and then execute those.

but for that scenario, we don't have the right forks and joins yet. in such situations, you should use a different kind of fork and join behaviour, based on multithreading. we don't have that yet.

the simplest way to get that non-persistent runtime concurrency behaviour is by just using the current forks and joins, but writing your own MessageService. Then you can leverage async="true" and your async message executor will generate multiple concurrent threads.

Later, we might actually create a new process language for non persisted multi threaded processes.

[archer77]

It seems that I've caused some misunderstanding (or perhaps I don't understand correctly).

I do not have process forks and joins in the process defninition, my jBPM just runs one linear process definition. To simplify testing even more, there is just one process instance available a time (process id = 1).

The problem is that two threads can interact with the process instance by calling some method on the jBPM core. Since these calls come from completely different directions, they cannot share the execution context (would be a security issue also), they have to rely only on the deterministic behaviour of the jBPM, including defined synchronisation behaviour.

As I understand it this problem should also exist in other setups e.g. jBoss SEAM framework with jBPM in the background, where multithreaded jBPM access occurs from different tomcat-webserver threads, but since these frameworks are comparatively slow, time-race conditions with vulnerable interval of 1-10ms or even less will not occur frequently.

Perhaps the central question is:

Is it allowed to access jBPM methods or objects from two different threads or do I have to build a "synchronized" cage around the whole jBPM to make sure that only one request is executed a time. (Still I'm not sure if this would really solve it since the jBPM includes timers which I guess are implemented as separate threads that could also cause the time-raze conditions).


BTW: If useful, I could try to setup a system with full persistence, but I guess that the problem won't be solved that way (unless the desynchronisation is due to broken lock handling in the underlying hibernate code or the in-memory database implementation).