Creating loops between nodes

[im_ytk]

I am using JBPM 3.1 + MySQL. I need to create a process definition which process continous events from external systems. I tried to create the process definition as following:

Start -> Wait -> Dispatcher -> Process1
.................................+-----> Process2

(Process2 starts from Dispatcher. The dots are used to align the graph)

Initially the process instance will stop at "Wait" state. When the event arrvies, it will continue its journey to Dispatcher, then Process1, return to Dispatcher then visit Process2. After the event has be processed, the token will be returned to Wait, and wait for next event by suspending the token. After suspending, it will return to caller, processInstance.save() and sleep for a while. Next time when the event arrives, I wake up the thread, I will resume() and signal() the token to move again.

I tried to create two Child Token to execute the processes, after that I will move the token from Dispatcher to Wait, and suspend the process.
I need to separate Process1 and Process2 because of user requirement.

Currently I have implemented a small test program to see if this method works, somehow it works as we are expected. But when the no. of event arrives increases, the processing speed increases. My experience is that after a thousand events have been triggered, the processing time required will be inceased to one second. Howeve, initially it takes only 1/3 seconds or so to process it (Just a feeling, I haven't use any tools to measure it)

Be honest I am not sure if what I am doing is correct. We needs to stuck with the graph of the Process Definition stated above, and the concept of looping. Please help.



Philip

[enazareno]

Hi Philip,

I suggest you take a look at this site www.workflowpatterns.com, select a pattern there that meets your requirement and reference it. I think most people here are familiar with the site and makes it easy to understand what you want. It is also quite informative.


Regards,

Elmo

[im_ytk]

Hello Elmo,

Thanks for your pointer. I have checked out the site and look like Pattern 10 (Arbitrary Cycle) and Pattern 11 (Implicit Termination) suitable my needs. I am trying to work on Pattern 10 right now. When I mimic the pattern in Process Definition, it run. But when it has run for a while, it threw a stack overflow exception. It looks to me if the no. of nested function calls reaches a certain limit, it will overflow. I have tried to bypass it by suspending the Process Instance, save it into database, clean up existing Process Instance, load, resume and continue. I do this every round. At the beginning, the performance is OK. However, when it has processed some events, say 200 rounds, its performance steadily drops. No exception has been thrown however. I am not very sure how to improves it. Please help.

Thanks.



Philip

[enazareno]

Hi Philip,

My guess would be a database issue. How are you closing your transactions? You should not hold on to sessions for long periods. Saving should only take a few seconds, then the database connection should be released. For long transactions, it is preferrable to signal your process through asynchronous transactions, like JMS perhaps.

Can you post your definition and code? I remember also having a stack overflow once but I cant quite remember. :(

Regards,

Elmo

[im_ytk]

Elmo,

I am using MySQL 5.0.18. I am not sure if it is the cause of the problem.

Please find attached the processdefinition I am using. I draw it using jBPM-Designer and then modify myself.

<?xml version="1.0" encoding="UTF-8"?>

<process-definition
 xmlns="urn:jbpm.org:jpdl-3.1" name="MultipleEvent">
 <start-state name="start">
 <event type="node-enter">
 <action name="startAction" class="com.multipleevent.action.StartHandler"></action>
 </event>
 <transition name="" to="Location"></transition>
 </start-state>
 <state name="Location">
 <event type="node-enter">
 <action name="locationAction" class="com.multipleevent.action.LocationHandler"></action>
 </event>
 <transition name="Location-Wait" to="Wait"></transition>
 <transition name="Location-Wait2" to="Wait2"></transition>
 </state>
 <node name="Wait">
 <event type="node-enter">
 <action name="waitAction" class="com.multipleevent.action.WaitHandler"></action>
 </event>
 <transition name="" to="Process"></transition>
 </node>
 <state name="Process">
 <event type="node-enter">
 <action name="processAction" class="com.multipleevent.action.ProcessHandler"></action>
 </event>
 </state>
 <node name="Wait2">
 <event type="node-enter">
 <action name="waitAction2" class="com.multipleevent.action.WaitHandler"></action>
 </event>
 <transition name="Wait2-Process2" to="Process2"></transition>
 </node>
 <state name="Process2">
 <event type="node-enter">
 <action name="processAction2" class="com.multipleevent.action.ProcessHandler"></action>
 </event>
 </state>
</process-definition>

The main driver is the Location handler, I ask the handler to visit each branch by creating child tokens, then set the location back to Start, and suspend the process.

public void execute(ExecutionContext executionContext) throws Exception {
 Token token = executionContext.getProcessInstance().getRootToken();
 Token childToken = null;
 ExecutionContext childExecutionContext = null;

 childToken = new Token(token, "childToken");
 childExecutionContext = new ExecutionContext(childToken);
 childExecutionContext.getNode().leave(childExecutionContext,"Location-Wait");

 childToken = new Token(token, "childToken2");
 childExecutionContext = new ExecutionContext(childToken);
 childExecutionContext.getNode().leave(childExecutionContext,"Location-Wait2");

 executionContext.getToken().setNode(executionContext.getProcessDefinition().getNode("start"));
 executionContext.getProcessInstance().suspend();
 }

In main, I write like this (The codes are very messy. I am sorry).

public void run() {
 JbpmConfiguration jbpmConfiguration = null;
 ProcessDefinition processDefinition = null;

 JbpmContext jbpmContext = null;
 ProcessInstance processInstance = null;

 try {
 jbpmConfiguration = JbpmConfiguration.getInstance();
 processDefinition = ProcessDefinition.parseXmlResource("MultipleEvent.par/processdefinition.xml");
 jbpmContext = jbpmConfiguration.createJbpmContext();
 jbpmContext.deployProcessDefinition(processDefinition);
 processInstance = jbpmContext.newProcessInstance("MultipleEvent");
 processInstance.signal();
 jbpmContext.save(processInstance);
 } finally {
 jbpmContext.close();
 }

 while (true) {
 jbpmConfiguration = null;
 processDefinition = null;
 jbpmContext = null;
 processInstance = null;
 try {
 jbpmConfiguration = JbpmConfiguration.getInstance();
 processDefinition = ProcessDefinition.parseXmlResource("MultipleEvent.par/processdefinition.xml");
 jbpmContext = jbpmConfiguration.createJbpmContext();
 jbpmContext.deployProcessDefinition(processDefinition);
 processInstance = jbpmContext.loadProcessInstanceForUpdate(process_id);
 processInstance.resume();
 processInstance.signal();
 }
 finally {
 jbpmContext.close();
 }
 }
 }

Other Action handlers are just print outs.

When it was running, the JVM was growing slowly. I am not sure if somewhere in the middle I have done anything wrong causing memory leak.

Thanks.



Philip

[enazareno]

Hi Philip,

Sorry for the delayed response. We've been busy catching up on deadlines. Anyway may I ask why this code, looping then redeploying?

while (true) {
 jbpmConfiguration = null;
 processDefinition = null;
 jbpmContext = null;
 processInstance = null;
 try {
 jbpmConfiguration = JbpmConfiguration.getInstance();
 processDefinition = ProcessDefinition.parseXmlResource("MultipleEvent.par/processdefinition.xml");
 jbpmContext = jbpmConfiguration.createJbpmContext();
 jbpmContext.deployProcessDefinition(processDefinition);
 processInstance = jbpmContext.loadProcessInstanceForUpdate(process_id);
 processInstance.resume();
 processInstance.signal();
 }
 finally {
 jbpmContext.close();
 }
 }

Just a quick thing. As I remeber, one JbpmSession allocates one DB connection. Like I said, its not good to hold on to connections very long bec. normally connections have timeouts, and connections are expensive resources. As I've seen the code, it seems you are continually looping using the connection without releasing it. I would suggest recoding esp. the action handler part.

Your action handler is denying your main connection to get closed, its like its hogging it. What I suggest is use asynhronous transactions instead that signals your state when finished. However, I am not quite familiar with the suspend part and what it does to the connection, so I still have to check this one out.

Regards,

Elmo

[im_ytk]

Hi Elmo,

Sorry I keeps on bothering you. Hopefully I didn't interfere your normal work.

Ummm... It is a long story. In our project, we wish to use jBPM engine to process incoming events. We treat a ProcessInstance as the big boss. When the program starts, a ProcessInstance will be created, process something and reached a wait state. The most simpliest ProcessDefinition we have created is something like this.

<Start> ---> <Event Listener> ---> <Process it> ---> <End1>
 ^-----------------------+

The ProcessInstance will terminate when some conditions are fulfilled. Say it has received an END event. We do not exactly knows how many loops it will loop through. In above scenario, it will loop forever until the user issued the END event.

The problem arise when using loop. If I draw the graph this way, allowing jBPM to handle the loop, it will break with StackOverFlow after a few hundred rounds. I tried to learn from jBPM source code, and I guess it is because the nodes are called recursively, causing function stack overflow.

I tried to bypass this problem by saving the ProcessInstance into database (anyway, my project needs to save everything into the database to guard against system crashes). When it has finished one cycle, it suspends and flushes everything. Then when another event arrives, it loads back the ProcessInstance, puts the event into Context variable, and starts again. This method somehow works (of course the performance is horrible, we are not supposed to code like this), but the memory consumed will grow slowly, and the performance will gradually decline.

About the connection, I am not sure. I have opened MySQL Administrator to have a quick check. It has 10 DB connections on average. Looks like Hibernate's connection pool has handled most of the problem. I don't know how to check the no. of connections in Hibernate. I will see if I can find it out. If the no. of Hibernate connections keep on increasing, then it can explain why the program gradually stall.

Let me rework my code to see if it improves. Keep in touch.

Thanks for your help.



Philip

[enazareno]

Hi Philip,

I'd just like to illustrate what I meant with an example. Why don't you try it this way and tell me if it works for you:

<process-definition name="sample-event-listener">
 <start-state>
 <transition to="event-listener"/>
 </start-state>

 <state name="event-listener">
 <event type="node-enter">
 <action class="..." />
 </event>
 <transition name="end-process" to="end-state"/>
 </state>

 <end-state name="end-state"/>

</process-definition>

action handler class:

public void execute(ExecutionContext ctx) throws Exception {
 long processid = ctx.getProcessInstance().getId();

 //since this is asynchronous, it will pass this immediately
 //and return handle to the main context
 AsyncEventListenerThread th = new AsyncEventListenerThread( id );
 th.start();
}

the sample listener thread

public class AsyncEventListenerThread extends Thread {
 long processid = processid;

 public AsyncEventListenerThread( long processid ) {
 this.processid = processid;
 }

 public void run() {
 //place your event listenter here. A JMS client would be good here
 //after event arrives, call a new JbpmSession, locate the process,
 //signal it and save and don't forget to close the jbpm context.
 }
}

While you're at it, try also reading on asynchronous continuations. Maybe they have something there that's already done for this job. I have not yet read on that, let me know. Good luck.

Regards,

Elmo

[im_ytk]

Hi Elmo,

Ummm.. I am not very understand. You mean when ProcessInstance starts, in one of the nodes, it will create a separate thread, then the ProcessInstance will suspend, or return to main and save.

When an event happens, this thread will load that ProcessInstance, and then continue its execution. Umm... I cannot figure out the looping mechanism. I guess above code implement the looping by not saving the ProcessInstance when it is invoked by the forked thread. Then next time when the forked thread loads the ProcessInstance again, it will load the ProcessInstance sits in the correct position.

I don't know, I usually save the ProcessInstance immediately after I have used it. If I load the ProcessInstance from DB and let it run till the end, next time when I load and execute the same ProcessInstance again, it will said the ProcessInstance has already reached its end state. (Of course I can always move the token back to the start state by "manipulating" the token myself. But I guess it is always not preferred).

I am reading async='true' now in chapter asynchronous continuations. I've tried adding the attribute into process definition, it doesn't work. I must have missed out something.

Thanks.


Philip

[enazareno]

This sample is an excerpt from arbitrary cycle of the workflow patterns:

<process-definition name="arbitrary-cycle">

 <start-state name="A">
 <transition to="B"/>
 </start-state>

 <state name="B">
 <timer duedate="20 seconds" transition="C">
 <script>print( "Your 20 seconds is up. Just letting you know I will go to C now" );</script>
 </timer>
 <transition name="C" to="C">
 </state>

 <state name="C">
 <event type="node-enter">
 <action class="DoAsyncStuff"/>
 </event>
 <transition name="fail" to="B"/>
 <transition name="pass" to="E"/>
 </state>

 <end-state name="E"/>

</process-definition>

This is the main execution part. Executed only once.

 JbpmContext ctx = config.createJbpmContext();
 try {
 ProcessInstance proc = ctx.newProcessInstance( "arbitrary-cycle" );
 proc.signal();
 ctx.save(proc);
 }
 catch(Exception ex)
 {
 ctx.setRollbackOnly();
 throw ex;
 }
 finally
 {
 ctx.close();
 }

In the EventListenerThread example from my previous code

 public void run() {
 //this class executes a long running process. It returns a status
 boolean pass = ALongRunningProcess.execute();

 JbpmContext ctx = config.createJbpmContext();
 try {
 ProcessInstance proc = ctx.newProcessInstance( "arbitrary-cycle" );
 if( !pass )
 proc.signal("fail");
 else
 proc.signal("pass");
 ctx.save(proc);
 }
 catch(Exception ex)
 {
 ctx.setRollbackOnly();
 throw ex;
 }
 finally
 {
 ctx.close();
 }
 }

Take note: Running threads has some issues in a J2ee environment, also this will not run if you restart the server. But thats another story. So this example is just used to illustrate a point. In fact you can accomplish something similar using timers instead. Hope this is clear.

Regards,

Elmo

[enazareno]

Oops sorry, must have been sleepy. In the thread you must retrieve the old process instance not create a new one. Here is the correction

public void run() {
 //this class executes a long running process. It returns a status
 boolean pass = ALongRunningProcess.execute();

 JbpmContext ctx = config.createJbpmContext();
 try {
 ProcessInstance proc = ctx.loadProcessInstance( processid );
 if( !pass )
 proc.signal("fail");
 else
 proc.signal("pass");
 ctx.save(proc);
 }
 catch(Exception ex)
 {
 ctx.setRollbackOnly();
 }
 finally
 {
 ctx.close();
 }
 }

Regards,

Elmo

[im_ytk]

Dear Elmo,

I am trying the modify my code and run it, but I got a "No row with the given identifier exists". Not sure why.

BTW, I have tried the newProcessInstance() also. It runs but it suffers from performance and stack overflow issue. I guess we are not supposed to loop a ProcessInstance forever....

Thanks.


Philip