| By Parameswaran Seshan |
Article Rating: |
|
| October 16, 2008 08:00 PM EDT |
Reads: |
5,045 |
The second argument (fairness) is given a "true" value to indicate that the Semaphore class would honor the order in which the threads arrived to acquire the lock. In other words, after the first thread that acquired the lock releases it, Semaphore follows the fairness principle and allows the thread that is next in order of invocation arrival to acquire the lock (FIFO - first in first out). If this argument is given "false" as value, then Java's default thread order comes into play. i.e., there is no guarantee made on the order of the threads acquiring locks and it is possible that, among the threads that are waiting to acquire the lock, one that arrived later is given a chance to acquire it ahead of a thread that arrived first.
As is shown in the process object code above, to acquire a permit to use the critical section, the thread needs to call the Semaphore's acquire() method and after completing the execution of the critical section code, it must release it by calling the release() method on the Semaphore object. Notice that the release() is done in "finally" block. This is to ensure that irrespective of a successful execution of the critical section code or not (an exception happening in it), the Semaphore is released. Otherwise, other threads waiting to access it would end up in indefinite wait.
Receive Activity and Correlation
There could be system activities in the process that are expected to receive invocations into them from outside; they are called "receive activities." In the case of receive activities (and user activities too), the incoming message would contain input parameters for the activity from the caller and these parameters are supplied to the activity directly. Any output parameters from the receive activity is passed back in return to the caller directly also during this time of exchange. Once this is done, the execution of the process then proceeds separately in its own thread and the caller's execution proceeds separately in the caller thread.
In BPML, an activity that is defined with a child element <input> followed by an optional <output> is considered a receive activity. Input and output elements define the input and output parameters for the activity. This means that this type of an activity receives invocation into it, i.e., the activity responds to an input message. Hence when process execution control reaches this activity, the process has to wait for an external party to invoke this action. Once the invocation (or message) is received from the outside, the activity execution proceeds and reaches the next activity in the process flow. The equivalent for this in WS-BPEL is "receive". In the case of the other type of activity, i.e., where we have <output> followed by <input>, the BPMS server invokes the action corresponding to the activity and so no waiting is involved. Once the process instance pauses for the message to arrive, it gets hibernated/passivated after a configured waiting time similar to user activity.
An important concept in receive activity is correlation. This refers to the method of mapping the incoming message to the receive activity of the right process instance among the set of running process instances. For example, let's take an order process that has a confirmation activity (a receive activity) in which the customer has to send a confirmation message to this activity to indicate order confirmation so the order process can proceed. Here, there would be as many order process instances running in the BPMS server as there are orders being processed at a time. The incoming message would have to specify a business attribute and its value to identify which order the message is intended for. In this case, the order ID, which is a unique identifier for each order, could be used as the correlation parameter. The incoming message to the confirmation activity would come with the order ID value for the order it wants to confirm.
In cases where the receive action does not have any output parameters to return to the caller (i.e., asynchronous invocation into the receive activity) or it has output parameters to return to the caller but does not have any activities to perform within it before returning the output parameters, we can implement the receive activity by using the class Exchanger from the concurrent library.
Exchanger is a synchronization point at which two threads can pair and swap elements within pairs. However, unlike a CyclicBarrier, no common task (to be executed before each of these threads continues executing) can be performed. In the ReceiveActivity class, we create an instance of the Exchanger class. Then we call the exchange() method on it, supplying the prepared output parameters in the name-value form as a HashMap object to make this thread wait for the input message to arrive. If there are no output parameters for this activity, then null is supplied. The incoming message thread (the caller thread) would join this Exchanger by calling the exchange() method and passing the input parameters (as a name-value HashMap object) to it as the argument. The exchanger then passes this input parameter HashMap to the ReceiveActivity thread and in turn passes the output parameter HashMap supplied by the ReceiveActivity to the incoming message thread. If there are no output parameters to be returned, null is returned to the incoming message thread. After this, both the threads depart and proceed independently with their executions. The ReceiveActivity's thread (the process execution thread) now completes the ReceiveActivity execution (post-execute) and then moves on to the next activity in the process's sequence.
In Listing 4, before putting the entry in the correlation table, we have to make it thread-safe since parallel threads could be trying to update its entries at the same time. We use a Semaphore class here (with argument 1 and true). The acquire() method on this Semaphore is called before the correlation table entry is made and the release() method on it is called after it but before the activity threads start waiting (via the exchange() method call) for the incoming message. The invoker (i.e., the incoming message to this activity) would do the following (see Listing 5).
If the receive activity has activities within it to be performed before returning the output parameters, we can implement the receive activity by using the class CyclicBarrier (of size two parties) instead of the Exchanger, since we would need to execute a common task (to be executed before each of these threads departs) at the time when the incoming message thread joins with the receive activity's thread. This common task (which extends the Java.lang.Runnable) will execute all the activities defined under (i.e., within) this receive activity in the order in which they are defined. Both the threads, the receive activity thread and the incoming message thread, are held by the CyclicBarrier until the common task's execution is completed. At the end of common task's execution, the two threads depart and move on with their respective executions independently. The code to use the CyclicBarrier here would be similar to what we saw in above for the stateBarrier.
Conclusion
This article showed how we could use the concurrent utility that is now part of Java 5.0 to build multithreaded execution capabilities in a Java application. We saw how a BPMS runtime platform uses this utility library as its core to handle multithreaded execution and synchronization of business processes and realize a high degree of parallelization with accompanying performance benefits.
Resources
Subscribe to JAVA Developer's Journal Email News Alerts
Subscribe via RSS
