You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@uima.apache.org by ea...@apache.org on 2008/05/29 21:41:59 UTC
svn commit: r661446 [3/4] - in
/incubator/uima/sandbox/branches/uima-as-post1st:
uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/activemq/
uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/client/
uimaj-as-activemq/src/test/java/o...
Modified: incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java?rev=661446&r1=661445&r2=661446&view=diff
==============================================================================
--- incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java (original)
+++ incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java Thu May 29 12:41:58 2008
@@ -148,6 +148,14 @@
protected ConcurrentHashMap perCasStatistics = new ConcurrentHashMap();
+ private boolean casMultiplier = false;
+
+ protected Object syncObject = new Object();
+
+ // Map holding outstanding CASes produced by Cas Multiplier that have to be acked
+ protected ConcurrentHashMap cmOutstandingCASes = new ConcurrentHashMap();
+
+
public BaseAnalysisEngineController(AnalysisEngineController aParentController, int aComponentCasPoolSize, String anEndpointName, String aDescriptor, AsynchAECasManager aCasManager, InProcessCache anInProcessCache) throws Exception
{
this(aParentController, aComponentCasPoolSize, 0, anEndpointName, aDescriptor, aCasManager, anInProcessCache, null, null);
@@ -239,10 +247,13 @@
getUimaContextAdmin().getManagementInterface();
// Override uima core jmx domain setting
mbean.setName(getComponentName(), getUimaContextAdmin(),jmxManagement.getJmxDomain());
- if ( this instanceof PrimitiveAnalysisEngineController && resourceSpecifier instanceof AnalysisEngineDescription )
+// if ( this instanceof PrimitiveAnalysisEngineController && resourceSpecifier instanceof AnalysisEngineDescription )
+ if ( resourceSpecifier instanceof AnalysisEngineDescription )
{
+ // Is this service a CAS Multiplier?
if ( ((AnalysisEngineDescription) resourceSpecifier).getAnalysisEngineMetaData().getOperationalProperties().getOutputsNewCASes() )
{
+ casMultiplier = true;
System.out.println(getName()+"-Initializing CAS Pool for Context:"+getUimaContextAdmin().getQualifiedContextName());
System.out.println(getComponentName()+"-CasMultiplier Cas Pool Size="+aComponentCasPoolSize+" Cas Initialial Heap Size:"+anInitialCasHeapSize);
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
@@ -350,11 +361,6 @@
}
if ( this instanceof PrimitiveAnalysisEngineController )
{
-// if ( (statistic = getMonitor().getLongNumericStatistic("",Monitor.ProcessCount)) == null )
-// {
-// statistic = new LongNumericStatistic(Monitor.ProcessCount);
-// getMonitor().addStatistic("", statistic);
-// }
if ( (statistic = getMonitor().getLongNumericStatistic("",Monitor.ProcessErrorCount)) == null )
{
statistic = new LongNumericStatistic(Monitor.ProcessErrorCount);
@@ -451,7 +457,6 @@
String name = "";
int index = getIndex();
servicePerformance = new ServicePerformance();
-// name = getJMXDomain()+key_value_list+",name="+thisComponentName+"_"+servicePerformance.getLabel();
name = jmxManagement.getJmxDomain()+key_value_list+",name="+thisComponentName+"_"+servicePerformance.getLabel();
@@ -471,10 +476,7 @@
}
if ( pServiceInfo != null )
{
-// name = getJMXDomain()+key_value_list+",name="+thisComponentName+"_"+serviceInfo.getLabel();
name = jmxManagement.getJmxDomain()+key_value_list+",name="+thisComponentName+"_"+serviceInfo.getLabel();
-
-
if ( !isTopLevelComponent() )
{
pServiceInfo.setBrokerURL("Embedded Broker");
@@ -483,10 +485,7 @@
}
serviceErrors = new ServiceErrors();
-// name = getJMXDomain()+key_value_list+",name="+thisComponentName+"_"+serviceErrors.getLabel();
name = jmxManagement.getJmxDomain()+key_value_list+",name="+thisComponentName+"_"+serviceErrors.getLabel();
-
-
registerWithAgent(serviceErrors, name );
}
@@ -661,23 +660,12 @@
registeredWithJMXServer = true;
registerServiceWithJMX(jmxContext, false);
}
-/*
- if ( this instanceof AggregateAnalysisEngineController )
- {
- AggregateAnalysisEngineController aC = (AggregateAnalysisEngineController)this;
- if ( aC.requestForMetaSentToRemotes() == false && allDelegatesAreRemote )
- {
- aC.setRequestForMetaSentToRemotes();
- aC.sendRequestForMetadataToRemoteDelegates();
- }
-*/
}
public void addInputChannel( InputChannel anInputChannel )
{
if ( !inputChannelMap.containsKey(anInputChannel.getInputQueueName()))
{
inputChannelMap.put(anInputChannel.getInputQueueName(), anInputChannel);
-
}
}
public InputChannel getInputChannel()
@@ -1111,53 +1099,6 @@
}
/**
- * Logs controller statistics in a uima log.
- *
- * @param aComponentName
- * @param aStatsMap
- */
-/*
- protected void logStats(String aComponentName, Map aStatsMap)
- {
- float totalIdleTime = 0;
- long numberCASesProcessed = 0;
- float totalDeserializeTime = 0;
- float totalSerializeTime = 0;
-
- if ( aStatsMap.containsKey(Monitor.IdleTime))
- {
- totalIdleTime = ((Float)aStatsMap.get(Monitor.IdleTime)).floatValue();
- }
- if ( aStatsMap.containsKey(Monitor.ProcessCount))
- {
- numberCASesProcessed = ((Long)aStatsMap.get(Monitor.ProcessCount)).longValue();
- }
- if ( aStatsMap.containsKey(Monitor.TotalDeserializeTime))
- {
- totalDeserializeTime = ((Float)aStatsMap.get(Monitor.TotalDeserializeTime)).floatValue();
- }
- if ( aStatsMap.containsKey(Monitor.TotalDeserializeTime))
- {
- totalSerializeTime = ((Float)aStatsMap.get(Monitor.TotalSerializeTime)).floatValue();
- }
- float totalAEProcessTime=0;
- if ( aStatsMap.containsKey(Monitor.TotalAEProcessTime))
- {
- totalAEProcessTime = ((Float)aStatsMap.get(Monitor.TotalAEProcessTime)).floatValue();
- }
-
- if ( totalAEProcessTime > 0 )
- {
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, getClass().getName(), "logStats", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_dump_primitive_stats__INFO", new Object[] { aComponentName, totalIdleTime, numberCASesProcessed, totalDeserializeTime, totalSerializeTime, totalAEProcessTime });
- }
- else
- {
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, getClass().getName(), "logStats", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_dump_aggregate_stats__INFO", new Object[] { aComponentName, totalIdleTime, numberCASesProcessed, totalDeserializeTime, totalSerializeTime });
- }
-
- }
-*/
- /**
* Clears controller statistics.
*
*/
@@ -1366,14 +1307,14 @@
((AggregateAnalysisEngineController_impl)this).stopTimers();
// Stops ALL input channels of this service including the reply channels
stopInputChannels();
- int childControllerListSize = ((AggregateAnalysisEngineController_impl)this).childControllerList.size();
+ int childControllerListSize = ((AggregateAnalysisEngineController_impl)this).getChildControllerList().size();
// send terminate event to all collocated child controllers
if ( childControllerListSize > 0 )
{
for( int i=0; i < childControllerListSize; i++ )
{
AnalysisEngineController childController =
- (AnalysisEngineController)((AggregateAnalysisEngineController_impl)this).childControllerList.get(i);
+ (AnalysisEngineController)((AggregateAnalysisEngineController_impl)this).getChildControllerList().get(i);
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, getClass().getName(), "stop", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_stop_delegate__INFO", new Object[] { getComponentName(), childController.getComponentName() });
childController.stop();
@@ -1488,23 +1429,6 @@
// fully processed.
stopCasMultiplier();
stop();
-/*
- // If the InProcessCache is not empty ( CASes are still in play), register self
- // (the top level controller) to receive a callback when all CASes are fully
- // processed and the cache becomes empty.
- if ( !getInProcessCache().isEmpty() )
- {
- System.out.println("Controller:"+getComponentName()+" Cache Not Empty. Registering Self For Callback");
- getInProcessCache().dumpContents();
- getInProcessCache().registerCallbackWhenCacheEmpty(this.getEventListener());
- }
- else
- {
- // Cache is already empty - trigger shutdown. If this controller is an
- // aggregate, it will propagate stop() down the delegate hierarchy
- getEventListener().onCacheEmpty();
- }
-*/
}
}
@@ -1621,16 +1545,14 @@
public AnalysisEngineController getCasMultiplierController()
{
- int childControllerListSize = ((AggregateAnalysisEngineController_impl)this).childControllerList.size();
+ int childControllerListSize = ((AggregateAnalysisEngineController_impl)this).getChildControllerList().size();
if ( childControllerListSize > 0 )
{
for( int i=0; i < childControllerListSize; i++ )
{
AnalysisEngineController childController =
- (AnalysisEngineController)((AggregateAnalysisEngineController_impl)this).childControllerList.get(i);
- if ( childController instanceof PrimitiveAnalysisEngineController &&
- ((PrimitiveAnalysisEngineController)childController).isMultiplier()
- )
+ (AnalysisEngineController)((AggregateAnalysisEngineController_impl)this).getChildControllerList().get(i);
+ if ( childController.isCasMultiplier() )
{
return childController;
}
@@ -1651,7 +1573,6 @@
}
return null;
}
-
/**
* Callback method called the InProcessCache becomes empty meaning ALL CASes are processed.
@@ -1742,4 +1663,77 @@
perCasStatistics.remove(aCasReferenceId);
}
}
+
+ public boolean isCasMultiplier()
+ {
+ return casMultiplier;
+ }
+
+ public void releaseNextCas(String casReferenceId)
+ {
+ synchronized(syncObject)
+ {
+ // Check if the CAS is in the list of outstanding CASes and also exists in the cache
+ if ( cmOutstandingCASes.size() > 0 && cmOutstandingCASes.containsKey(casReferenceId) && getInProcessCache().entryExists(casReferenceId))
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "releaseNextCas", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_release_cas_req__FINE",
+ new Object[] { getComponentName(), casReferenceId });
+ try
+ {
+ CacheEntry cacheEntry = getInProcessCache().getCacheEntryForCAS(casReferenceId);
+ String parentCasReferenceId = cacheEntry.getInputCasReferenceId();
+ Endpoint freeCasEndpoint = cacheEntry.getFreeCasEndpoint();
+ // If the CAS was created by a remote Cas Multiplier, send a Free CAS Notification
+ // to the CM.
+ if ( freeCasEndpoint != null )
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "releaseNextCas", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_sending_fcq_req__FINE",
+ new Object[] { getComponentName(), casReferenceId, cacheEntry.getCasMultiplierKey(), freeCasEndpoint.getDestination() });
+ freeCasEndpoint.setReplyEndpoint(true);
+ getOutputChannel().sendRequest(AsynchAEMessage.ReleaseCAS, casReferenceId, freeCasEndpoint);
+ }
+ cacheEntry = null;
+ // Release the CAS and remove it from the InProcess cache
+ dropCAS(casReferenceId, true);
+ // Check if the CAS has a parent CAS
+ if ( parentCasReferenceId != null )
+ {
+ // Fetch the parent CAS from the InProcess Cache
+ cacheEntry = getInProcessCache().getCacheEntryForCAS(parentCasReferenceId);
+ if ( cacheEntry != null )
+ {
+ // Decrement number of child CASes in play
+ cacheEntry.decrementSubordinateCasInPlayCount();
+ if ( cacheEntry.isPendingReply() && cacheEntry.getSubordinateCasInPlayCount() == 0)
+ {
+ if ( this instanceof AggregateAnalysisEngineController )
+ {
+ ((AggregateAnalysisEngineController)this).finalStep( cacheEntry.getFinalStep(), parentCasReferenceId);
+ }
+ else // PrimitiveAnalysisEngineController
+ {
+ // Return an input CAS to the client. The input CAS is returned
+ // to the remote client only if all of the child CASes produced
+ // from the input CAS have been fully processed.
+ getOutputChannel().sendReply(cacheEntry.getCasReferenceId(), cacheEntry.getMessageOrigin());
+ dropCAS(cacheEntry.getCasReferenceId(), true);
+ }
+ }
+
+ }
+ }
+ }
+ catch( Exception e)
+ {
+ e.printStackTrace();
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
+ "releaseNextCas", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_exception__WARNING",
+ new Object[] { e});
+ }
+ }
+ }
+ }
+
}
Modified: incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/Endpoint_impl.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/Endpoint_impl.java?rev=661446&r1=661445&r2=661446&view=diff
==============================================================================
--- incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/Endpoint_impl.java (original)
+++ incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/Endpoint_impl.java Thu May 29 12:41:58 2008
@@ -100,9 +100,11 @@
private boolean tempReplyDestination;
private int initialHeapSize;
+
private volatile boolean replyDestinationFailed;
private long idleTime=0;
+
public int getCommand()
{
return command;
Modified: incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/PrimitiveAnalysisEngineController.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/PrimitiveAnalysisEngineController.java?rev=661446&r1=661445&r2=661446&view=diff
==============================================================================
--- incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/PrimitiveAnalysisEngineController.java (original)
+++ incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/PrimitiveAnalysisEngineController.java Thu May 29 12:41:58 2008
@@ -25,8 +25,7 @@
public interface PrimitiveAnalysisEngineController extends AnalysisEngineController
{
- public boolean isMultiplier();
- public void releaseNextCas();
+// public boolean isMultiplier();
public void setAnalysisEngineInstancePool( AnalysisEngineInstancePool aPool);
public PrimitiveServiceInfo getServiceInfo();
public void addAbortedCasReferenceId( String aCasReferenceId );
Modified: incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/PrimitiveAnalysisEngineController_impl.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/PrimitiveAnalysisEngineController_impl.java?rev=661446&r1=661445&r2=661446&view=diff
==============================================================================
--- incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/PrimitiveAnalysisEngineController_impl.java (original)
+++ incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/PrimitiveAnalysisEngineController_impl.java Thu May 29 12:41:58 2008
@@ -70,17 +70,13 @@
private List aeList = new ArrayList();
- private List cmOutstandingCASes = new ArrayList();
-
private int throttleWindow = 0;
private Object gater = new Object();
private long howManyBeforeReplySeen = 0;
- private boolean isMultiplier = false;
- private Object syncObject = new Object();
private PrimitiveServiceInfo serviceInfo = null;
@@ -148,10 +144,6 @@
if (i == 0)
{
analysisEngineMetadata = ae.getAnalysisEngineMetaData();
- if ( analysisEngineMetadata.getOperationalProperties().getOutputsNewCASes())
- {
- isMultiplier = true;
- }
}
}
@@ -163,7 +155,6 @@
serviceInfo.setAnalysisEngineInstanceCount(analysisEnginePoolSize);
aeInstancePool.intialize(aeList);
-
getMonitor().setThresholds(getErrorHandlerChain().getThresholds());
// Initialize Cas Manager
if (getCasManagerWrapper() != null)
@@ -182,6 +173,7 @@
// All internal components of this Primitive have been initialized. Open the latch
// so that this service can start processing requests.
latch.openLatch(getName(), isTopLevelComponent(), true);
+
}
catch ( Exception e)
{
@@ -197,20 +189,20 @@
}
catch ( AsynchAEException e)
{
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), "initialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_exception__WARNING", new Object[] { e });
+ e.printStackTrace();
throw e;
}
catch ( Exception e)
{
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), "initialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_exception__WARNING", new Object[] { e });
+ e.printStackTrace();
throw new AsynchAEException(e);
}
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "initialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_initialized_controller__INFO", new Object[] { getComponentName() });
super.serviceInitialized = true;
}
- public boolean isMultiplier()
- {
- return isMultiplier;
- }
/**
*
@@ -307,6 +299,7 @@
return;
}
*/
+ boolean inputCASReturned = false;
boolean processingFailed = false;
// This is a primitive controller. No more processing is to be done on the Cas. Mark the destination as final and return CAS in reply.
anEndpoint.setFinal(true);
@@ -333,35 +326,19 @@
return;
}
+ // Get input CAS entry from the InProcess cache
+ CacheEntry inputCASEntry = getInProcessCache().getCacheEntryForCAS(aCasReferenceId);
+
long time = System.nanoTime();
long totalProcessTime = 0; // stored total time spent producing ALL CASes
CasIterator casIterator = ae.processAndOutputNewCASes(aCAS);
// Store how long it took to call processAndOutputNewCASes()
totalProcessTime = ( System.nanoTime() - time);
long sequence = 1;
- String newCasReferenceId = null;
long hasNextTime = 0; // stores time in hasNext()
long getNextTime = 0; // stores time in next();
long timeToProcessCAS = 0; // stores time in hasNext() and next() for each CAS
boolean moreCASesToProcess = true;
-/*
-
- String parentCasReferenceId = null;
- CacheEntry inputCasCacheEntry = null;
- try
- {
- // Fetch cache entry for the input CAS
- inputCasCacheEntry = getInProcessCache().getCacheEntryForCAS(aCasReferenceId);
- parentCasReferenceId = inputCasCacheEntry.getInputCasReferenceId();
- }
- catch( Exception e )
- {
- // An exception be be thrown here if the service is being stopped.
- // The top level controller may have already cleaned up the cache
- // and the getCacheEntryForCAS() will throw an exception. Ignore it
- // here, we are shutting down.
- }
-*/
while (moreCASesToProcess)
{
hasNextTime = System.nanoTime();
@@ -412,35 +389,37 @@
}
OutOfTypeSystemData otsd = getInProcessCache().getOutOfTypeSystemData(aCasReferenceId);
MessageContext mContext = getInProcessCache().getMessageAccessorByReference(aCasReferenceId);
- sequence++;
- newCasReferenceId = getInProcessCache().register( casProduced, mContext, otsd);
- CacheEntry newEntry = getInProcessCache().getCacheEntryForCAS(newCasReferenceId);
-/*
- if ( parentCasReferenceId == null )
- {
- newEntry.setInputCasReferenceId(aCasReferenceId);
- }
- else
- {
- newEntry.setInputCasReferenceId(parentCasReferenceId);
- }
-*/
+ CacheEntry newEntry = getInProcessCache().register( casProduced, mContext, otsd);
+ // Associate input CAS with the new CAS
newEntry.setInputCasReferenceId(aCasReferenceId);
+ newEntry.setCasSequence(sequence);
// Add to the cache how long it took to process the generated (subordinate) CAS
- getCasStatistics(newCasReferenceId).incrementAnalysisTime(timeToProcessCAS);
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, getClass().getName(), "process", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_produced_new_cas__FINE", new Object[] { Thread.currentThread().getName(),getComponentName(),newCasReferenceId, aCasReferenceId });
+ getCasStatistics(newEntry.getCasReferenceId()).incrementAnalysisTime(timeToProcessCAS);
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, getClass().getName(), "process", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_produced_new_cas__FINE", new Object[] { Thread.currentThread().getName(),getComponentName(),newEntry.getCasReferenceId(), aCasReferenceId });
+ // Add the generated CAS to the outstanding CAS Map. Client notification will release
+ // this CAS back to its pool
synchronized(syncObject)
{
- cmOutstandingCASes.add(newCasReferenceId);
- getOutputChannel().sendReply(casProduced, aCasReferenceId, newCasReferenceId, anEndpoint, sequence);
+ if ( isTopLevelComponent() )
+ {
+ inputCASEntry.incrementSubordinateCasInPlayCount();
+ // Add the id of the generated CAS to the map holding outstanding CASes. This
+ // map will be referenced when a client sends Free CAS Notification. The map
+ // stores the id of the CAS both as a key and a value. Map is used to facilitate
+ // quick lookup
+ cmOutstandingCASes.put(newEntry.getCasReferenceId(),newEntry.getCasReferenceId());
+ }
+ // Send generated CAS to the client
+ getOutputChannel().sendReply(newEntry, anEndpoint);
}
// Remove Stats from the global Map associated with the new CAS
- // This stats for this CAS were added to the response message
+ // These stats for this CAS were added to the response message
// and are no longer needed
- dropCasStatistics(newCasReferenceId);
+ dropCasStatistics(newEntry.getCasReferenceId());
// Increment number of CASes processed by this service
getServicePerformance().incrementNumberOfCASesProcessed();
getServicePerformance().incrementAnalysisTime(timeToProcessCAS);
+ sequence++;
}
LongNumericStatistic statistic = null;
@@ -449,23 +428,28 @@
// Increment how long it took to process the input CAS. This timer is exposed via JMX
statistic.increment(totalProcessTime);
}
-/*
- if (newCasReferenceId != null)
- {
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, getClass().getName(), "process", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_completed_analysis__FINEST", new Object[] { Thread.currentThread().getName(), getComponentName(), newCasReferenceId, (double) (System.nanoTime() - time) / (double) 1000000 });
- }
- else
- {
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, getClass().getName(), "process", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_completed_analysis__FINEST", new Object[] { Thread.currentThread().getName(), getComponentName(), aCasReferenceId, (double) (System.nanoTime() - time) / (double) 1000000 });
-
- }
-*/
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, getClass().getName(), "process", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_completed_analysis__FINEST", new Object[] { Thread.currentThread().getName(), getComponentName(), aCasReferenceId, (double) (System.nanoTime() - time) / (double) 1000000 });
getMonitor().resetCountingStatistic("", Monitor.ProcessErrorCount);
getCasStatistics(aCasReferenceId).incrementAnalysisTime(totalProcessTime);
// Aggregate total time spent processing the input CAS
getServicePerformance().incrementAnalysisTime(totalProcessTime);
- getOutputChannel().sendReply(aCasReferenceId, anEndpoint);
+ synchronized( cmOutstandingCASes )
+ {
+ if ( cmOutstandingCASes.size() == 0)
+ {
+ inputCASReturned = true;
+ // Return an input CAS to the client if there are no outstanding child CASes in play
+ getOutputChannel().sendReply(aCasReferenceId, anEndpoint);
+ }
+ else
+ {
+ // Change the state of the input CAS. Since the input CAS is not returned to the client
+ // until all children of this CAS has been fully processed we keep the input in the cache.
+ // The client will send Free CAS Notifications to release CASes produced here. When the
+ // last child CAS is freed, the input CAS is allowed to be returned to the client.
+ inputCASEntry.setPendingReply(true);
+ }
+ }
}
catch ( Throwable e)
{
@@ -507,8 +491,9 @@
getInProcessCache().releaseCASesProducedFromInputCAS(aCasReferenceId);
}
- else
+ else if ( inputCASReturned )
{
+ // Remove input CAS cache entry if the CAS has been sent to the client
dropCAS(aCasReferenceId, true);
}
}
@@ -620,32 +605,6 @@
{
return super.getMetaData().getName();
}
- public void releaseNextCas()
- {
- synchronized(syncObject)
- {
- if ( cmOutstandingCASes.size() > 0 )
- {
- try
- {
- String casReferenceId = (String)cmOutstandingCASes.remove(0);
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
- "releaseNextCas", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_release_cas_req__FINE",
- new Object[] { getName(), casReferenceId });
- if ( casReferenceId != null && getInProcessCache().entryExists(casReferenceId))
- {
- dropCAS(casReferenceId, true);
- }
- }
- catch( Exception e)
- {
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(),
- "releaseNextCas", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_exception__WARNING",
- new Object[] { e});
- }
- }
- }
- }
public void setAnalysisEngineInstancePool( AnalysisEngineInstancePool aPool)
{
Modified: incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/error/handler/ProcessCasErrorHandler.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/error/handler/ProcessCasErrorHandler.java?rev=661446&r1=661445&r2=661446&view=diff
==============================================================================
--- incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/error/handler/ProcessCasErrorHandler.java (original)
+++ incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/error/handler/ProcessCasErrorHandler.java Thu May 29 12:41:58 2008
@@ -392,8 +392,7 @@
getCacheEntryForCAS(parentCasReferenceId);
synchronized( parentCasCacheEntry )
{
- ((AggregateAnalysisEngineController)aController).
- decrementCasSubordinateCount( parentCasCacheEntry);
+ parentCasCacheEntry.decrementSubordinateCasInPlayCount();
if ( parentCasCacheEntry.getSubordinateCasInPlayCount() == 0 &&
parentCasCacheEntry.isPendingReply())
{
Modified: incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/HandlerBase.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/HandlerBase.java?rev=661446&r1=661445&r2=661446&view=diff
==============================================================================
--- incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/HandlerBase.java (original)
+++ incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/HandlerBase.java Thu May 29 12:41:58 2008
@@ -165,6 +165,7 @@
public Handler getDelegate()
{
+// System.out.println("In getDelegate() - Returning:"+delegateHandler.getName());
return delegateHandler;
}
Modified: incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/input/MetadataRequestHandler_impl.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/input/MetadataRequestHandler_impl.java?rev=661446&r1=661445&r2=661446&view=diff
==============================================================================
--- incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/input/MetadataRequestHandler_impl.java (original)
+++ incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/input/MetadataRequestHandler_impl.java Thu May 29 12:41:58 2008
@@ -78,7 +78,6 @@
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(),
"handle", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_msg_for_next_handler__FINEST",
new Object[] { messageType });
-
super.getDelegate().handle(anObjectToHandle);
}
else
@@ -92,6 +91,7 @@
}
catch( Exception e)
{
+ e.printStackTrace();
getController().getErrorHandlerChain().handle(e, HandlerBase.populateErrorContext( (MessageContext)anObjectToHandle ), getController());
}
}
Modified: incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/input/ProcessRequestHandler_impl.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/input/ProcessRequestHandler_impl.java?rev=661446&r1=661445&r2=661446&view=diff
==============================================================================
--- incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/input/ProcessRequestHandler_impl.java (original)
+++ incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/handler/input/ProcessRequestHandler_impl.java Thu May 29 12:41:58 2008
@@ -58,89 +58,265 @@
entry.incrementTimeWaitingForCAS( aTimeWaitingForCAS);
entry.incrementTimeToDeserializeCAS(aTimeToDeserializeCAS);
}
-
- private void handleProcessRequestWithXMI(MessageContext aMessageContext) throws AsynchAEException
+ private boolean messageContainsXMI(MessageContext aMessageContext, String casReferenceId) throws Exception
+ {
+ // Fetch serialized CAS from the message
+ String xmi = aMessageContext.getStringMessage();
+ // *****************************************************************
+ // ***** NO XMI In Message. Kick this back to sender with exception
+ // *****************************************************************
+ if ( xmi == null )
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
+ "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_message_has_no_cargo__INFO",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint() });
+ getController().
+ getOutputChannel().
+ sendReply(new InvalidMessageException("No XMI data in message"), casReferenceId, aMessageContext.getEndpoint(),AsynchAEMessage.Process);
+ // Dont process this empty message
+ return false;
+ }
+ return true;
+ }
+ private CAS getCAS( boolean fetchCASFromShadowCasPool, String shadowCasPoolKey, String casReceivedFrom )
{
CAS cas = null;
- String casReferenceId = null;
+ if ( fetchCASFromShadowCasPool )
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(),
+ "getCAS", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_request_cas_cm__FINE",
+ new Object[] { shadowCasPoolKey });
+ // If this is a new CAS (generated by a CM), fetch a CAS from a Shadow Cas Pool associated with a CM that
+ // produced the CAS. Each CM will have its own Shadow Cas Pool
+ cas = getController().getCasManagerWrapper().getNewCas(shadowCasPoolKey);
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "getCAS", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_request_cas_granted_cm__FINE",
+ new Object[] { shadowCasPoolKey });
+ }
+ else
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(),
+ "getCAS", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_request_cas__FINE",
+ new Object[] { casReceivedFrom });
+ cas = getController().getCasManagerWrapper().getNewCas();
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "getCAS", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_request_cas_granted__FINE",
+ new Object[] { casReceivedFrom });
+ }
+ return cas;
+ }
+ /**
+ *
+ * @param casReferenceId
+ * @param freeCasEndpoint
+ * @param shadowCasPoolKey
+ * @param aMessageContext
+ * @return
+ * @throws Exception
+ */
+ private CacheEntry deserializeCASandRegisterWithCache( String casReferenceId, Endpoint freeCasEndpoint, String shadowCasPoolKey, MessageContext aMessageContext)
+ throws Exception
+ {
long inTime = System.nanoTime();
boolean casRegistered = false;
- boolean requestToFreeCasSent = false;
+
+ // Fetch serialized CAS from the message
+ String xmi = aMessageContext.getStringMessage();
+
+ // Time how long we wait on Cas Pool to fetch a new CAS
+ long t1 = System.nanoTime();
+ // *************************************************************************
+ // Fetch CAS from a Cas Pool. If the CAS came from a Cas Multiplier
+ // fetch the CAS from a shadow CAS pool. Otherwise, fetch the CAS
+ // from the service CAS Pool.
+ // *************************************************************************
+ CAS cas = getCAS(aMessageContext.propertyExists(AsynchAEMessage.CasSequence), shadowCasPoolKey, aMessageContext.getEndpoint().getEndpoint());
+ long timeWaitingForCAS = System.nanoTime() - t1;
+ // Check if we are still running
+ if ( getController().isStopped() )
+ {
+ // The Controller is in shutdown state.
+ getController().dropCAS(cas);
+ return null;
+ }
+ // *************************************************************************
+ // Deserialize the CAS from the message
+ // *************************************************************************
+ t1 = System.nanoTime();
+ XmiSerializationSharedData deserSharedData = new XmiSerializationSharedData();
+ UimaSerializer.deserializeCasFromXmi(xmi, cas, deserSharedData, true, -1);
+ long timeToDeserializeCAS = System.nanoTime() - t1;
+
+ LongNumericStatistic statistic;
+ if ( (statistic = getController().getMonitor().getLongNumericStatistic("",Monitor.TotalDeserializeTime)) != null )
+ {
+ statistic.increment(timeToDeserializeCAS);
+ }
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_deserialize_cas_time_FINE",
+ new Object[] { timeToDeserializeCAS / 1000 });
+ // *************************************************************************
+ // Register the CAS with a local cache
+ // *************************************************************************
+ CacheEntry entry = getController().getInProcessCache().register(cas, aMessageContext, deserSharedData, casReferenceId);
+
+ // Update Stats
+ ServicePerformance casStats = getController().getCasStatistics(casReferenceId);
+ casStats.incrementCasDeserializationTime(timeToDeserializeCAS);
+ if ( getController().isTopLevelComponent() )
+ {
+ getController().getServicePerformance().incrementCasDeserializationTime(timeToDeserializeCAS);
+ }
+ getController().saveTime(inTime, casReferenceId, getController().getName());
+
+ if ( getController() instanceof AggregateAnalysisEngineController )
+ {
+ // If the message came from a Cas Multiplier, associate the input/parent CAS id with this CAS
+ if ( aMessageContext.propertyExists(AsynchAEMessage.CasSequence) )
+ {
+ // Fetch parent CAS id
+ String inputCasReferenceId =
+ aMessageContext.getMessageStringProperty(AsynchAEMessage.InputCasReference);
+ if ( shadowCasPoolKey != null )
+ {
+ // Save the key of the Cas Multiplier in the cache. It will be now known which Cas Multiplier produced this CAS
+ entry.setCasMultiplierKey(shadowCasPoolKey);
+ }
+ // associate this subordinate CAS with the parent CAS
+ entry.setInputCasReferenceId(inputCasReferenceId);
+ // Save a Cas Multiplier endpoint where a Free CAS notification will be sent
+ entry.setFreeCasEndpoint(freeCasEndpoint);
+ cacheStats( inputCasReferenceId, timeWaitingForCAS, timeToDeserializeCAS);
+ }
+ else
+ {
+ cacheStats( casReferenceId, timeWaitingForCAS, timeToDeserializeCAS);
+ }
+ DelegateStats stats = new DelegateStats();
+ if ( entry.getStat() == null )
+ {
+ entry.setStat(stats);
+ // Add entry for self (this aggregate). MessageContext.getEndpointName()
+ // returns the name of the queue receiving the message.
+ stats.put(getController().getServiceEndpointName(), new TimerStats());
+ }
+ else
+ {
+ if (!stats.containsKey(getController().getServiceEndpointName()))
+ {
+ stats.put(getController().getServiceEndpointName(), new DelegateStats());
+ }
+ }
+ }
+ else
+ {
+ cacheStats( casReferenceId, timeWaitingForCAS, timeToDeserializeCAS);
+ }
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_deserialized_cas_ready_to_process_FINE",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint() });
+ cacheProcessCommandInClientEndpoint();
+ return entry;
+ }
+ private String getCasReferenceId( MessageContext aMessageContext ) throws Exception
+ {
+ if ( !aMessageContext.propertyExists(AsynchAEMessage.CasReference) )
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
+ "handleProcessRequestWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_message_has_cas_refid__INFO",
+ new Object[] { aMessageContext.getEndpoint().getEndpoint() });
+
+ getController().
+ getOutputChannel().
+ sendReply(new InvalidMessageException("No Cas Reference Id Received From Delegate In message"), null, aMessageContext.getEndpoint(),AsynchAEMessage.Process);
+ return null;
+ }
+ return aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
+ }
+ /**
+ * Handles process request from a remote client
+ *
+ * @param aMessageContext - contains a message from UIMA-AS Client
+ * @throws AsynchAEException
+ */
+ private void handleProcessRequestWithXMI(MessageContext aMessageContext) throws AsynchAEException
+ {
+ CacheEntry entry = null;
+ String casReferenceId = null;
+ // Check if there is a cargo in the message
+ if ( aMessageContext.getStringMessage() == null )
+ {
+ return; // No XMI just return
+ }
+
try
{
- boolean isNewCAS = false;
String newCASProducedBy = null;
- String remoteCasReferenceId = null;
- // This is only used when handling CASes produced by CAS Multiplier
-
// Get the CAS Reference Id of the input CAS
- casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
+ // Fetch id of the CAS from the message. If it doesnt exist the method will create an entry in the log file and return null
+ casReferenceId = getCasReferenceId(aMessageContext);
+ if ( casReferenceId == null )
+ {
+ return; // Invalid message. Nothing to do
+ }
// Initially make both equal
String inputCasReferenceId = casReferenceId;
- // CASes generated in a Cas Multiplier will have a CasSequence property set. If such property exists
- // it means that the CAS has been generated by a CM.
+ // Destination where Free Cas Notification will be sent if the CAS came from a Cas Multiplier
+ Endpoint freeCasEndpoint = null;
+ // CASes generated by a Cas Multiplier will have a CasSequence property set.
if ( aMessageContext.propertyExists(AsynchAEMessage.CasSequence) )
{
- // Remote CM?
- if ( aMessageContext.getEndpoint().isRemote())
- {
- remoteCasReferenceId = casReferenceId;
- }
- // Set the flag to indicate that the CAS been generated by CM
- isNewCAS = true;
- // Fetch the actual input CAS Reference Id from which the CAS being processed was generated from
+ // Fetch an ID of the parent CAS
inputCasReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.InputCasReference);
- // Fetch input CAS Cache entry
- CacheEntry inputCasCacheEntry = getController().
- getInProcessCache().
- getCacheEntryForCAS(inputCasReferenceId);
- // This CAS came in from the CAS Multiplier. Treat it differently than the
- // input CAS. First, in case the Aggregate needs to send this CAS to the
+ // Fetch Cache entry for the parent CAS
+ CacheEntry inputCasCacheEntry = getController().getInProcessCache().getCacheEntryForCAS(inputCasReferenceId);
+ // Fetch an endpoint where Free CAS Notification must be sent.
+ // This endpoint is unique per CM instance. Meaning, each
+ // instance of CM will have an endpoint where it expects Free CAS
+ // notifications.
+ freeCasEndpoint = aMessageContext.getEndpoint();
+ // This CAS came in from a CAS Multiplier. Treat it differently than the
+ // input CAS. In case the Aggregate needs to send this CAS to the
// client, retrieve the client destination by looking up the client endpoint
// using input CAS reference id. CASes generated by the CAS multiplier will have
// the same Cas Reference id.
Endpoint replyToEndpoint = inputCasCacheEntry.getMessageOrigin();
- if ( getController() instanceof AggregateAnalysisEngineController )
- {
- newCASProducedBy = inputCasCacheEntry.getCasMultiplierKey();
- // increment number of subordinate CASes that are currently being processed
- // The input CAS (parent) will be held by the aggregate until all of its
- // subordinate CASes are fully processed. Only then, the aggregate can return
- // it back to the client
- synchronized( inputCasCacheEntry )
- {
- inputCasCacheEntry.incrementSubordinateCasInPlayCount();
- }
- if ( ((AggregateAnalysisEngineController)getController()).sendRequestToReleaseCas() )
- {
- try
- {
- // Change the name of the queue where the request to free a CAS will be sent.
- aMessageContext.getEndpoint().setEndpoint(aMessageContext.getEndpoint().getEndpoint()+"CasSync");
-
- getController().getOutputChannel().sendRequest(AsynchAEMessage.ReleaseCAS, casReferenceId, aMessageContext.getEndpoint());
- requestToFreeCasSent = true;
- }
- catch( Exception e){}
- }
- }
- // MessageContext contains endpoint set by the CAS Multiplier service. Overwrite
- // this with the endpoint of the client who sent the input CAS. In case this
- // aggregate is configured to send new CASes to the client we know where to send them.
+ // The message context contains a Cas Multiplier endpoint. Since
+ // we dont want to send a generated CAS back to the CM, override
+ // with an endpoint provided by the client of
+ // this service. Client endpoint is attached to an input Cas cache entry.
aMessageContext.getEndpoint().setEndpoint(replyToEndpoint.getEndpoint());
aMessageContext.getEndpoint().setServerURI(replyToEndpoint.getServerURI());
- // Set this to null so that the new CAS gets its own Cas Reference Id below
- casReferenceId = null;
- }
- else if ( getController().isTopLevelComponent())
- {
- Endpoint replyToEndpoint = aMessageContext.getEndpoint();
- if ( getController() instanceof AggregateAnalysisEngineController )
- {
- ((AggregateAnalysisEngineController)getController()).addMessageOrigin(casReferenceId, replyToEndpoint);
+ // Before sending a CAS to Cas Multiplier, the aggregate has
+ // saved the CM key in the CAS cache entry. Fetch the key
+ // of the CM so that we can ask the right Shadow Cas Pool for
+ // a new CAS. Every Shadow Cas Pool has a unique id which
+ // corresponds to a Cas Multiplier key.
+ newCASProducedBy = inputCasCacheEntry.getCasMultiplierKey();
+ if ( getController() instanceof AggregateAnalysisEngineController )
+ {
+ Endpoint casMultiplierEndpoint =
+ ((AggregateAnalysisEngineController)getController()).lookUpEndpoint(newCASProducedBy, false);
+ if ( casMultiplierEndpoint != null )
+ {
+ // Save the URL of the broker managing the Free Cas Notification queue.
+ // This is needed when we try to establish a connection to the broker.
+ freeCasEndpoint.setServerURI(casMultiplierEndpoint.getServerURI());
+ }
+ }
+ // increment number of CASes produced from an input CAS
+ // The input CAS (parent) will be held by
+ // the aggregate until all of its subordinate CASes are
+ // fully processed. Only then, the aggregate can return
+ // it back to the client
+ synchronized (inputCasCacheEntry) {
+ inputCasCacheEntry.incrementSubordinateCasInPlayCount();
}
-
+ }
+ else if ( getController().isTopLevelComponent() && getController() instanceof AggregateAnalysisEngineController )
+ {
+ ((AggregateAnalysisEngineController)getController()).addMessageOrigin(casReferenceId, aMessageContext.getEndpoint());
}
// To prevent processing multiple messages with the same CasReferenceId, check the CAS cache
// to see if the message with a given CasReferenceId is already being processed. It is, the
@@ -151,168 +327,23 @@
// CasReferenceId. When the service finally comes back up, it will have multiple messages in
// its queue possibly from the same client. Only the first message for any given CasReferenceId
// should be processed.
- if ( casReferenceId == null || !getController().getInProcessCache().entryExists(casReferenceId) )
+ if ( !getController().getInProcessCache().entryExists(casReferenceId) )
{
- String xmi = aMessageContext.getStringMessage();
-
- // *****************************************************************
- // ***** NO XMI In Message. Kick this back to sender with exception
- // *****************************************************************
- if ( xmi == null )
- {
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
- "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_message_has_no_cargo__INFO",
- new Object[] { aMessageContext.getEndpoint().getEndpoint() });
- getController().
- getOutputChannel().
- sendReply(new InvalidMessageException("No XMI data in message"), casReferenceId, aMessageContext.getEndpoint(),AsynchAEMessage.Process);
- // Dont process this empty message
- return;
- }
-
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, CLASS_NAME.getName(),
- "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_request_cas__FINEST",
- new Object[] { aMessageContext.getEndpoint().getEndpoint() });
- long t1 = System.nanoTime();
- if ( isNewCAS )
- {
- cas = getController().getCasManagerWrapper().getNewCas(newCASProducedBy);
- }
- else
- {
- cas = getController().getCasManagerWrapper().getNewCas();
- }
- long timeWaitingForCAS = System.nanoTime() - t1;
-
- if ( getController().isStopped() )
- {
- // The Controller is in shutdown state.
- getController().dropCAS(cas);
- return;
- }
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
- "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_request_cas_granted__FINE",
- new Object[] { aMessageContext.getEndpoint().getEndpoint() });
- t1 = System.nanoTime();
- XmiSerializationSharedData deserSharedData = new XmiSerializationSharedData();
- UimaSerializer.deserializeCasFromXmi(xmi, cas, deserSharedData, true, -1);
-
- long timeToDeserializeCAS = System.nanoTime() - t1;
- LongNumericStatistic statistic;
- if ( (statistic = getController().getMonitor().getLongNumericStatistic("",Monitor.TotalDeserializeTime)) != null )
- {
- statistic.increment(timeToDeserializeCAS);
- }
-
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
- "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_deserialize_cas_time_FINE",
- new Object[] { timeToDeserializeCAS / 1000 });
- ServicePerformance casStats = null;
-
-
- if (casReferenceId == null)
- {
- if (getController() instanceof PrimitiveAnalysisEngineController)
- {
- inputCasReferenceId = getController().getInProcessCache().register(cas, aMessageContext, deserSharedData);
- }
- else
- {
- casReferenceId = getController().getInProcessCache().register(cas, aMessageContext, deserSharedData);
- if ( inputCasReferenceId == null )
- {
- inputCasReferenceId = casReferenceId;
- }
- }
- casStats = getController().getCasStatistics(inputCasReferenceId);
-
- }
- else
- {
- getController().getInProcessCache().register(cas, aMessageContext, deserSharedData, casReferenceId);
-/*
- if ( aMessageContext.propertyExists(AsynchAEMessage.InputCasReference))
- {
- CacheEntry cacheEntry = getController().getInProcessCache().getCacheEntryForCAS(casReferenceId);
- String parentCasId = aMessageContext.getMessageStringProperty(AsynchAEMessage.InputCasReference);
- cacheEntry.setInputCasReferenceId(parentCasId);
- }
-*/
- casStats = getController().getCasStatistics(casReferenceId);
- }
- casStats.incrementCasDeserializationTime(timeToDeserializeCAS);
- if ( getController().isTopLevelComponent() )
- {
- getController().getServicePerformance().incrementCasDeserializationTime(timeToDeserializeCAS);
- }
- // Set a local flag to indicate that the CAS has been added to the cache. This will be usefull when handling an exception
- // If an exception happens before the CAS is added to the cache, the CAS needs to be dropped immediately.
- casRegistered = true;
- if ( casReferenceId == null )
- {
- getController().saveTime(inTime, inputCasReferenceId, getController().getName());
- }
- else
- {
- getController().saveTime(inTime, casReferenceId, getController().getName());
- }
-
- CacheEntry entry = null;
- if ( getController() instanceof AggregateAnalysisEngineController )
- {
- entry = getController().getInProcessCache().getCacheEntryForCAS(casReferenceId);
- if ( isNewCAS )
- {
- if ( newCASProducedBy != null )
- {
- entry.setCasMultiplierKey(newCASProducedBy);
- }
- if ( remoteCasReferenceId != null )
- {
- entry.setRemoteCMCasReferenceId(remoteCasReferenceId);
- }
- if ( requestToFreeCasSent )
- {
- entry.setSendRequestToFreeCas(false);
- }
- // associate this subordinate CAS with the parent CAS
- entry.setInputCasReferenceId(inputCasReferenceId);
- }
- DelegateStats stats = new DelegateStats();
- if ( entry.getStat() == null )
- {
- entry.setStat(stats);
- // Add entry for self (this aggregate). MessageContext.getEndpointName()
- // returns the name of the queue receiving the message.
- stats.put(getController().getServiceEndpointName(), new TimerStats());
- }
- else
- {
- if (!stats.containsKey(getController().getServiceEndpointName()))
- {
- stats.put(getController().getServiceEndpointName(), new DelegateStats());
- }
- }
- }
-
- cacheStats( inputCasReferenceId, timeWaitingForCAS, timeToDeserializeCAS);
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
- "handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_deserialized_cas_ready_to_process_FINE",
- new Object[] { aMessageContext.getEndpoint().getEndpoint() });
-
-
- cacheProcessCommandInClientEndpoint();
-
- if ( getController().isStopped() )
+ entry = deserializeCASandRegisterWithCache( casReferenceId, freeCasEndpoint, newCASProducedBy, aMessageContext);
+ if ( getController().isStopped() || entry == null || entry.getCas() == null)
{
if ( entry != null )
{
- // The Controller is in shutdown state.
+ // The Controller is in shutdown state, release the CAS
getController().dropCAS( entry.getCasReferenceId(), true);
- return;
+ entry = null;
}
+ return;
}
- invokeProcess(cas, inputCasReferenceId, casReferenceId, aMessageContext, newCASProducedBy);
+ // *****************************************************************
+ // Process the CAS
+ // *****************************************************************
+ invokeProcess(entry.getCas(), inputCasReferenceId, casReferenceId, aMessageContext, newCASProducedBy);
}
else
{
@@ -320,7 +351,6 @@
"handleProcessRequestWithXMI", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_duplicate_request__INFO",
new Object[] { casReferenceId});
}
-
}
catch ( Exception e)
{
@@ -332,31 +362,122 @@
errorContext.add(AsynchAEMessage.Endpoint, aMessageContext.getEndpoint());
errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.Process);
errorContext.add(AsynchAEMessage.CasReference, casReferenceId );
- if ( casRegistered == false )
+ if ( entry != null )
{
- getController().dropCAS(cas);
+ getController().dropCAS(entry.getCas());
}
getController().getErrorHandlerChain().handle(e, errorContext, getController());
}
}
- private void handleProcessRequestWithXCAS(MessageContext aMessageContext) throws AsynchAEException
+ private void handleProcessRequestWithCASReference(MessageContext aMessageContext) throws AsynchAEException
{
-
+ boolean isNewCAS = false;
+ String newCASProducedBy = null;
+
+
try
{
- boolean isNewCAS = false;
- String newCASProducedBy = null;
-
// This is only used when handling CASes produced by CAS Multiplier
+ String inputCasReferenceId = null;
+ CAS cas = null;
+ String casReferenceId = getCasReferenceId(aMessageContext);
+ if ( aMessageContext.propertyExists(AsynchAEMessage.CasSequence) )
+ {
+ isNewCAS = true;
+ Endpoint casMultiplierEndpoint = aMessageContext.getEndpoint();
+
+ if ( casMultiplierEndpoint == null )
+ {
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
+ "handleProcessRequestWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_no_endpoint_for_reply__INFO",
+ new Object[] { casReferenceId });
+ return;
+ }
+ //
+ if ( getController() instanceof AggregateAnalysisEngineController )
+ {
+ getController().getInProcessCache().setCasProducer(casReferenceId, casMultiplierEndpoint.getEndpoint());
+ newCASProducedBy = ((AggregateAnalysisEngineController)getController()).lookUpDelegateKey(casMultiplierEndpoint.getEndpoint());
+ casMultiplierEndpoint.setIsCasMultiplier(true);
+ ((AggregateAnalysisEngineController)getController()).getServicePerformance(newCASProducedBy).incrementNumberOfCASesProcessed();
+ CacheEntry subordinateCasCacheEntry = getController().getInProcessCache().getCacheEntryForCAS(casReferenceId);
+ CacheEntry inputCasCacheEntry = getController().getInProcessCache().getCacheEntryForCAS(subordinateCasCacheEntry.getInputCasReferenceId());
+ if ( inputCasCacheEntry != null )
+ {
+ synchronized( inputCasCacheEntry )
+ {
+ inputCasCacheEntry.incrementSubordinateCasInPlayCount();
+ }
+ }
+ }
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleProcessRequestWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_new_cas__FINE",
+ new Object[] { casReferenceId, newCASProducedBy });
+
+ aMessageContext.getEndpoint().setEndpoint(casMultiplierEndpoint.getEndpoint());
+ aMessageContext.getEndpoint().setServerURI(casMultiplierEndpoint.getServerURI());
+ inputCasReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.InputCasReference);
+ }
+ else
+ {
+ if ( getController() instanceof AggregateAnalysisEngineController )
+ {
+ ((AggregateAnalysisEngineController)getController()).addMessageOrigin(casReferenceId, aMessageContext.getEndpoint());
+ }
+
+ }
+ cas = getController().getInProcessCache().getCasByReference(casReferenceId);
+ long arrivalTime = System.nanoTime();
+ getController().saveTime(arrivalTime, casReferenceId, getController().getName());//aMessageContext.getEndpointName());
+
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleProcessRequestWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_analyzing_cas__FINE",
+ new Object[] { casReferenceId});
+
+ // Save Process command in the client endpoint.
+ cacheProcessCommandInClientEndpoint();
+
+ if ( getController().isStopped() )
+ {
+ return;
+ }
+
+ if ( isNewCAS )
+ {
+ invokeProcess(cas, inputCasReferenceId, casReferenceId, aMessageContext, newCASProducedBy);
+ }
+ else
+ {
+ invokeProcess(cas, casReferenceId, null, aMessageContext, newCASProducedBy);
+ }
+ }
+ catch ( AsynchAEException e)
+ {
+ throw e;
+ }
+ catch ( Exception e)
+ {
+ throw new AsynchAEException(e);
+ }
+
+ }
+
+
+ private void handleProcessRequestWithXCAS(MessageContext aMessageContext) throws AsynchAEException
+ {
+
+ try
+ {
// Get the CAS Reference Id of the input CAS
- String casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
+ String casReferenceId = getCasReferenceId(aMessageContext);
String inputCasReferenceId = casReferenceId;
+ // This is only used when handling CASes produced by CAS Multiplier
+ String newCASProducedBy = null;
if ( aMessageContext.propertyExists(AsynchAEMessage.CasSequence) )
{
- isNewCAS = true;
// This CAS came in from the CAS Multiplier. Treat it differently than the
// input CAS. First, in case the Aggregate needs to send this CAS to the
// client, retrieve the client destination by looking up the client endpoint
@@ -429,7 +550,8 @@
if (casReferenceId == null)
{
- casReferenceId = getController().getInProcessCache().register(cas, aMessageContext, deserSharedData);
+ CacheEntry entry = getController().getInProcessCache().register(cas, aMessageContext, deserSharedData);
+ casReferenceId = entry.getCasReferenceId();
}
else
{
@@ -465,136 +587,6 @@
}
- private void handleProcessRequestWithCASReference(MessageContext aMessageContext) throws AsynchAEException
- {
- boolean isNewCAS = false;
- String newCASProducedBy = null;
-
- if ( getController().isStopped() )
- {
- return;
- }
-
- try
- {
- String casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
-
- if ( casReferenceId == null )
- {
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
- "handleProcessRequestWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_message_has_cas_refid__INFO",
- new Object[] { aMessageContext.getEndpoint().getEndpoint() });
-
- getController().
- getOutputChannel().
- sendReply(new InvalidMessageException("No Cas Reference Id Received From Delegate In message"), null, aMessageContext.getEndpoint(),AsynchAEMessage.Process);
- // Dont process this empty message
- return;
-
- }
-// Endpoint replyToEndpoint = aMessageContext.getEndpoint();
-
- // This is only used when handling CASes produced by CAS Multiplier
- String inputCasReferenceId = null;
- CAS cas = null;
-
- if ( aMessageContext.propertyExists(AsynchAEMessage.CasSequence) )
- {
- isNewCAS = true;
- Endpoint casMultiplierEndpoint = aMessageContext.getEndpoint();
-
- if ( casMultiplierEndpoint == null )
- {
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(),
- "handleProcessRequestWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_no_endpoint_for_reply__INFO",
- new Object[] { casReferenceId });
- return;
- }
- //
- if ( getController() instanceof AggregateAnalysisEngineController )
- {
- getController().getInProcessCache().setCasProducer(casReferenceId, casMultiplierEndpoint.getEndpoint());
- newCASProducedBy =
- ((AggregateAnalysisEngineController)getController()).lookUpDelegateKey(casMultiplierEndpoint.getEndpoint());
- casMultiplierEndpoint.setIsCasMultiplier(true);
- ((AggregateAnalysisEngineController)getController()).
- getServicePerformance(newCASProducedBy).
- incrementNumberOfCASesProcessed();
- CacheEntry subordinateCasCacheEntry = getController().getInProcessCache().getCacheEntryForCAS(casReferenceId);
- CacheEntry inputCasCacheEntry = getController().
- getInProcessCache().
- getCacheEntryForCAS(subordinateCasCacheEntry.getInputCasReferenceId());
- if ( inputCasCacheEntry != null )
- {
- synchronized( inputCasCacheEntry )
- {
-// System.out.println("++++++++ Incrementing Delegate:"+casReferenceId+" Count For Input Cas::"+subordinateCasCacheEntry.getInputCasReferenceId());
- inputCasCacheEntry.incrementSubordinateCasInPlayCount();
- }
- }
- }
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
- "handleProcessRequestWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_new_cas__FINE",
- new Object[] { casReferenceId, newCASProducedBy });
- // MessageContext contains endpoint set by the CAS Multiplier service. Overwrite
- // this with the endpoint of the client who sent the input CAS. In case this
- // aggregate is configured to send new CASes to the client we know where to send them.
-// if ( aMessageContext.getEndpoint() != null )
-// {
-// aMessageContext.getEndpoint().setEndpoint(casMultiplierEndpoint.getEndpoint());
-// aMessageContext.getEndpoint().setServerURI(casMultiplierEndpoint.getServerURI());
-// }
- aMessageContext.getEndpoint().setEndpoint(casMultiplierEndpoint.getEndpoint());
- aMessageContext.getEndpoint().setServerURI(casMultiplierEndpoint.getServerURI());
-
- inputCasReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.InputCasReference);
- }
- else
- {
- if ( getController() instanceof AggregateAnalysisEngineController )
- {
- ((AggregateAnalysisEngineController)getController()).addMessageOrigin(casReferenceId, aMessageContext.getEndpoint());
- }
-
- }
- cas = getController().getInProcessCache().getCasByReference(casReferenceId);
-
- long arrivalTime = System.nanoTime();
- getController().saveTime(arrivalTime, casReferenceId, getController().getName());//aMessageContext.getEndpointName());
-
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
- "handleProcessRequestWithCASReference", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_analyzing_cas__FINE",
- new Object[] { casReferenceId});
-
- // Save Process command in the client endpoint.
- cacheProcessCommandInClientEndpoint();
-
- if ( getController().isStopped() )
- {
- return;
- }
-
- if ( isNewCAS )
- {
- invokeProcess(cas, inputCasReferenceId, casReferenceId, aMessageContext, newCASProducedBy);
- }
- else
- {
- invokeProcess(cas, casReferenceId, null, aMessageContext, newCASProducedBy);
- }
- }
- catch ( AsynchAEException e)
- {
- throw e;
- }
- catch ( Exception e)
- {
- throw new AsynchAEException(e);
- }
-
- }
-
-
private void cacheProcessCommandInClientEndpoint()
{
@@ -612,32 +604,13 @@
getController().collectionProcessComplete(replyToEndpoint);
}
- private void handleReleaseCASRequest(MessageContext aMessageContext)
+ private void handleReleaseCASRequest(MessageContext aMessageContext) throws AsynchAEException
{
-
- if ( getController() instanceof PrimitiveAnalysisEngineController )
- {
- ( (PrimitiveAnalysisEngineController)getController()).releaseNextCas();
- }
-/*
- try
- {
- String casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
- "handleReleaseCASRequest", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_release_cas_req__FINE",
- new Object[] { getController().getName(), casReferenceId });
-System.out.println(getController().getName()+" ::::::: Processing Release CAS Request:"+casReferenceId);
-
- if ( casReferenceId != null && getController().getInProcessCache().entryExists(casReferenceId))
- {
- getController().dropCAS(casReferenceId, true);
- }
- }
- catch( Exception e)
- {
- getController().getErrorHandlerChain().handle(e, HandlerBase.populateErrorContext( (MessageContext)aMessageContext ), getController());
- }
-*/
+ String casReferenceId = aMessageContext.getMessageStringProperty(AsynchAEMessage.CasReference);
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(),
+ "handleReleaseCASRequest", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_release_cas_req__FINE",
+ new Object[] { getController().getName(), casReferenceId });
+ getController().releaseNextCas(casReferenceId);
}
private void handleStopRequest(MessageContext aMessageContext)
@@ -676,23 +649,43 @@
getController().getControllerLatch().waitUntilInitialized();
- // If a Process Request, increment number of docs processed
- if (messageContext.getMessageIntProperty(AsynchAEMessage.MessageType) == AsynchAEMessage.Request
- && command == AsynchAEMessage.Process) {
- // Increment number of CASes processed by this service
- getController().getServicePerformance().incrementNumberOfCASesProcessed();
- }
+ // If a Process Request, increment number of docs processed
+ if (messageContext.getMessageIntProperty(AsynchAEMessage.MessageType) == AsynchAEMessage.Request
+ && command == AsynchAEMessage.Process) {
+ // Increment number of CASes processed by this service
+ getController().getServicePerformance().incrementNumberOfCASesProcessed();
+ }
+ if ( getController().isStopped() )
+ {
+ return;
+ }
if (AsynchAEMessage.CASRefID == payload)
{
+ // Fetch id of the CAS from the message.
+ if ( getCasReferenceId(messageContext) == null )
+ {
+ return; // Invalid message. Nothing to do
+ }
+
handleProcessRequestWithCASReference(messageContext);
}
else if (AsynchAEMessage.XMIPayload == payload)
{
+ // Fetch id of the CAS from the message.
+ if ( getCasReferenceId(messageContext) == null )
+ {
+ return; // Invalid message. Nothing to do
+ }
handleProcessRequestWithXMI(messageContext);
}
else if (AsynchAEMessage.XCASPayload == payload)
{
+ // Fetch id of the CAS from the message.
+ if ( getCasReferenceId(messageContext) == null )
+ {
+ return; // Invalid message. Nothing to do
+ }
handleProcessRequestWithXCAS(messageContext);
}
else if ( AsynchAEMessage.None == payload && AsynchAEMessage.CollectionProcessComplete == command)
@@ -715,6 +708,7 @@
}
catch( Exception e)
{
+ e.printStackTrace();
getController().getErrorHandlerChain().handle(e, HandlerBase.populateErrorContext( (MessageContext)anObjectToHandle ), getController());
}
}
Modified: incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/message/AsynchAEMessage.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/message/AsynchAEMessage.java?rev=661446&r1=661445&r2=661446&view=diff
==============================================================================
--- incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/message/AsynchAEMessage.java (original)
+++ incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/java/org/apache/uima/aae/message/AsynchAEMessage.java Thu May 29 12:41:58 2008
@@ -23,7 +23,7 @@
public interface AsynchAEMessage
{
public static final String TotalTimeSpentInAnalytic = "TimeInAnalytic";
- public static final String SendTime = "SendTime";
+// public static final String SendTime = "SendTime";
public static final String TimeInService = "TimeInService";
public static final String Endpoint = "Endpoint";
public static final String DelegateStats = "DelegateStats";
Modified: incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/resources/uimaee_messages.properties
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/resources/uimaee_messages.properties?rev=661446&r1=661445&r2=661446&view=diff
==============================================================================
--- incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/resources/uimaee_messages.properties (original)
+++ incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-core/src/main/resources/uimaee_messages.properties Thu May 29 12:41:58 2008
@@ -61,6 +61,9 @@
UIMAEE_worker_thread_terminated__INFO = Worker Thread: {0} Terminated
UIMAEE_request_cas__FINE = Requesting CAS from CasPool For CAS Deserialization. Cas Sent From: {0}
UIMAEE_request_cas_granted__FINE = Obtained CAS from CasPool For CAS Deserialization. Cas Sent From: {0}
+UIMAEE_request_cas_cm__FINE = Requesting CAS from Cas Multipliers Shadow Cas Pool Identified By Key: {0} For CAS Deserialization.
+UIMAEE_request_cas_granted_cm__FINE = Obtained CAS from Cas Multipliers Shadow Cas Pool Identified By Key: {0} For CAS Deserialization.
+
UIMAEE_deserialized_cas_ready_to_process_FINE = Deserialized CAS from XMI. CAS Sent From: {0}
UIMAEE_got_cas__FINEST = Got Cas From: {0} Cas Received Contains: {1}
UIMAEE_duplicate_request__INFO = Duplicate Request With Cas Reference Id: {0} Received. Ignoring Duplicate.
@@ -146,3 +149,5 @@
UIMAEE_final_step_parent_cas_child_count__FINEST = Controller: {0} Final Step - Parent Cas Reference Id: {1} Has In-Play Subordinate CASes. Current Subordinate CAS Count: {2}
UIMAEE_final_step_parent_cas_no_children__FINEST = Controller: {0} Final Step - Parent Cas Reference Id: {1} Has No Subordinate CASes Being Processed.
UIMAEE_unable_to_check_ae_back_to_pool__WARNING = Controller: {0} Unable to Check In an Instance Of AE While Processing CPC. Exception {1}
+UIMAEE_sending_fcq_req__FINE = Controller: {0} Sending Request To Release CAS: {1} To Cas Multiplier:{2} Queue: {3}
+UIMAEE_remove_cache_entry__INFO = Controller: {0} Releasing CASes Produced From Input CAS: {1}
Modified: incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-jms/src/main/java/org/apache/uima/adapter/jms/client/BaseMessageSender.java
URL: http://svn.apache.org/viewvc/incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-jms/src/main/java/org/apache/uima/adapter/jms/client/BaseMessageSender.java?rev=661446&r1=661445&r2=661446&view=diff
==============================================================================
--- incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-jms/src/main/java/org/apache/uima/adapter/jms/client/BaseMessageSender.java (original)
+++ incubator/uima/sandbox/branches/uima-as-post1st/uimaj-as-jms/src/main/java/org/apache/uima/adapter/jms/client/BaseMessageSender.java Thu May 29 12:41:58 2008
@@ -21,6 +21,8 @@
import java.util.List;
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
import javax.jms.InvalidDestinationException;
import javax.jms.Message;
import javax.jms.MessageProducer;
@@ -295,4 +297,11 @@
engine.onException(e, aDestination);
}
+ /**
+ * @override
+ */
+ public MessageProducer getMessageProducer(Destination destination) throws Exception {
+ return null;
+ }
+
}