You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@uima.apache.org by cw...@apache.org on 2014/06/02 21:19:01 UTC
svn commit: r1599315 - in /uima/uima-as/trunk:
uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/
uimaj-as-core/src/main/java/org/apache/uima/aae/controller/
uimaj-as-core/src/main/java/org/apache/uima/aae/handler/input/
Author: cwiklik
Date: Mon Jun 2 19:19:01 2014
New Revision: 1599315
URL: http://svn.apache.org/r1599315
Log:
UIMA-3643 modified code to prevent deserializing a CAS if it has not been dispatched to all parallel delegates.
Modified:
uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsEndpointConnection_impl.java
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/LocalCache.java
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/input/ProcessResponseHandler.java
Modified: uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsEndpointConnection_impl.java
URL: http://svn.apache.org/viewvc/uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsEndpointConnection_impl.java?rev=1599315&r1=1599314&r2=1599315&view=diff
==============================================================================
--- uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsEndpointConnection_impl.java (original)
+++ uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/JmsEndpointConnection_impl.java Mon Jun 2 19:19:01 2014
@@ -688,6 +688,19 @@ public class JmsEndpointConnection_impl
// to find inactive sessions.
lastDispatchTimestamp.set(System.currentTimeMillis());
+ // If in ParallelStep its possible to receive a reply from one of the delegates in parallel
+ // step *before* a CAS is dispatched to all of the delegates. This can cause a problem
+ // as replies are merged which causes the CAS to be in an inconsistent state.
+ // The following code calls dispatchCasToParallelDelegate() which count down
+ // a java latch. The same latch is used when receiving replies. If the latch is non zero
+ // the code blocks a thread from performing deserialization.
+ if ( msgType == AsynchAEMessage.Request && command == AsynchAEMessage.Process ) {
+ String casReferenceId = aMessage.getStringProperty(AsynchAEMessage.CasReference);
+ CasStateEntry casStateEntry = controller.getLocalCache().lookupEntry(casReferenceId);
+ if ( casStateEntry.getNumberOfParallelDelegates() > 0) {
+ casStateEntry.dispatchedCasToParallelDelegate();
+ }
+ }
// Succeeded sending the CAS
return true;
} catch (Exception e) {
Modified: uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/LocalCache.java
URL: http://svn.apache.org/viewvc/uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/LocalCache.java?rev=1599315&r1=1599314&r2=1599315&view=diff
==============================================================================
--- uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/LocalCache.java (original)
+++ uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/LocalCache.java Mon Jun 2 19:19:01 2014
@@ -26,6 +26,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
@@ -240,6 +241,12 @@ public class LocalCache extends Concurre
private Object monitor = new Object();
+ // used only for parallel step. Counts down when a CAS is dispatched to a each delegate.
+ // Prevents a race when reply from 1st parallel delegate arrives before a dispatch to a remaining
+ // delegates has not yet completed. In such scenario, the CAS may be corrupted by merging
+ // reply from the 1st delegate.
+ private CountDownLatch latch=null;
+
public boolean waitingForChildrenToFinish() {
return waitingForChildren;
}
@@ -386,12 +393,35 @@ public class LocalCache extends Concurre
public void setNumberOfParallelDelegates(int aNumberOfParallelDelegates) {
numberOfParallelDelegates = aNumberOfParallelDelegates;
+ latch = new CountDownLatch(aNumberOfParallelDelegates);
}
public int getNumberOfParallelDelegates() {
return numberOfParallelDelegates;
}
-
+
+ public void dispatchedCasToParallelDelegate() {
+ if (latch != null ) {
+ if ( latch.getCount() > 0 ) {
+ latch.countDown();
+ }
+ }
+ }
+ public void blockIfParallelDispatchNotComplete() {
+ try {
+ if (latch != null ) {
+ latch.await();
+ }
+ } catch( InterruptedException e) {
+ }
+ }
+ public void resetDispatchLatch() {
+ if (latch != null ) {
+ while ( latch.getCount() > 0 ) {
+ latch.countDown();
+ }
+ }
+ }
public boolean isFailed() {
return failed;
}
Modified: uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/input/ProcessResponseHandler.java
URL: http://svn.apache.org/viewvc/uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/input/ProcessResponseHandler.java?rev=1599315&r1=1599314&r2=1599315&view=diff
==============================================================================
--- uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/input/ProcessResponseHandler.java (original)
+++ uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/input/ProcessResponseHandler.java Mon Jun 2 19:19:01 2014
@@ -246,6 +246,13 @@ public class ProcessResponseHandler exte
"UIMAEE_number_parallel_delegates_FINE",
new Object[] { totalNumberOfParallelDelegatesProcessingCas, Thread.currentThread().getId(), Thread.currentThread().getName() });
}
+ if (totalNumberOfParallelDelegatesProcessingCas > 1) {
+ // Block this thread until CAS is dispatched to all delegates in parallel step. Fixes race condition where
+ // a reply comes from one of delegates in parallel step before dispatch sequence completes. Without
+ // this blocking the result of analysis are merged into a CAS.
+ casStateEntry.blockIfParallelDispatchNotComplete();
+ }
+
if (cas == null) {
throw new AsynchAEException(Thread.currentThread().getName()
+ "-Cache Does not contain a CAS. Cas Reference Id::" + casReferenceId);