}
}
OutOfTypeSystemData otsd = getInProcessCache().getOutOfTypeSystemData(aCasReferenceId);
MessageContext mContext = getInProcessCache()
.getMessageAccessorByReference(aCasReferenceId);
CacheEntry newEntry = getInProcessCache().register(casProduced, mContext, otsd);
// if this Cas Multiplier is not Top Level service, add new Cas Id to the private
// cache of the parent aggregate controller. The Aggregate needs to know about
// all CASes it has in play that were generated from the input CAS.
CasStateEntry childCasStateEntry = null;
if (!isTopLevelComponent()) {
newEntry.setNewCas(true, parentController.getComponentName());
// Create CAS state entry in the aggregate's local cache
childCasStateEntry = parentController.getLocalCache().createCasStateEntry(
newEntry.getCasReferenceId());
// Fetch the parent CAS state entry from the aggregate's local cache. We need to increment
// number of child CASes associated with it.
parentCasStateEntry = parentController.getLocalCache().lookupEntry(aCasReferenceId);
} else {
childCasStateEntry = getLocalCache().createCasStateEntry(newEntry.getCasReferenceId());
}
// Associate parent CAS (input CAS) with the new CAS.
childCasStateEntry.setInputCasReferenceId(aCasReferenceId);
// Increment number of child CASes generated from the input CAS
parentCasStateEntry.incrementSubordinateCasInPlayCount();
parentCasStateEntry.incrementOutstandingFlowCounter();
// 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(newEntry.getCasReferenceId()).incrementAnalysisTime(timeToProcessCAS);
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) {
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(),
getUimaContextAdmin().getQualifiedContextName(),
newEntry.getCasReferenceId(), aCasReferenceId });
}
// Add the generated CAS to the outstanding CAS Map. Client notification will release
// this CAS back to its pool
synchronized (syncObject) {
if (isTopLevelComponent()) {
// 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());
}
// Increment number of CASes processed by this service
sequence++;
}
if (!anEndpoint.isRemote()) {
UimaTransport transport = getTransport(anEndpoint.getEndpoint());
UimaMessage message = transport.produceMessage(AsynchAEMessage.Process,
AsynchAEMessage.Request, getName());
message.addStringProperty(AsynchAEMessage.CasReference, newEntry.getCasReferenceId());
message.addStringProperty(AsynchAEMessage.InputCasReference, aCasReferenceId);
message.addLongProperty(AsynchAEMessage.CasSequence, sequence);
ServicePerformance casStats = getCasStatistics(aCasReferenceId);
message.addLongProperty(AsynchAEMessage.TimeToSerializeCAS, casStats
.getRawCasSerializationTime());
message.addLongProperty(AsynchAEMessage.TimeToDeserializeCAS, casStats
.getRawCasDeserializationTime());
message.addLongProperty(AsynchAEMessage.TimeInProcessCAS, casStats.getRawAnalysisTime());
long iT = getIdleTimeBetweenProcessCalls(AsynchAEMessage.Process);
message.addLongProperty(AsynchAEMessage.IdleTime, iT);
if (!stopped) {
transport.getUimaMessageDispatcher(anEndpoint.getEndpoint()).dispatch(message);
dropStats(newEntry.getCasReferenceId(), getName());
}
} else {
// Send generated CAS to the remote client
if (!stopped) {
getOutputChannel().sendReply(newEntry, anEndpoint);
// Check for delivery failure. The client may have terminated while an input CAS was being processed
if ( childCasStateEntry.deliveryToClientFailed() ) {
clientUnreachable = true;
if ( cmOutstandingCASes.containsKey(childCasStateEntry.getCasReferenceId())) {
cmOutstandingCASes.remove(childCasStateEntry.getCasReferenceId());
}
// Stop generating new CASes. We failed to send a CAS to a client. Most likely
// the client has terminated.
moreCASesToProcess = false; // exit the while loop
dropCAS(childCasStateEntry.getCasReferenceId(), true);
}
}
}
// Remove new CAS state entry from the local cache if this is a top level primitive.
// If not top level, the client (an Aggregate) will remove this entry when this new
// generated CAS reaches Final State.
if (isTopLevelComponent()) {
try {
localCache.lookupEntry(newEntry.getCasReferenceId()).setDropped(true);
} catch (Exception e) {
}
localCache.remove(newEntry.getCasReferenceId());
}
// Remove Stats from the global Map associated with the new CAS
// These stats for this CAS were added to the response message
// and are no longer needed
dropCasStatistics(newEntry.getCasReferenceId());
} // while
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) {
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) (super.getCpuTime() - time) / (double) 1000000 });
}
getMonitor().resetCountingStatistic("", Monitor.ProcessErrorCount);
// Set total number of children generated from this CAS
// Store total time spent processing this input CAS
getCasStatistics(aCasReferenceId).incrementAnalysisTime(totalProcessTime);
// Fetch AE's management information that includes per component performance stats
// These stats are internally maintained in a Map. If the AE is an aggregate
// the Map will contain AnalysisEngineManagement instance for each AE.
AnalysisEngineManagement aem = ae.getManagementInterface();
if ( aem.getComponents().size() > 0) {
// Flatten the hierarchy by recursively (if this AE is an aggregate) extracting
// primitive AE's AnalysisEngineManagement instance and placing it in
// afterAnalysisManagementObjects List.
getLeafManagementObjects(aem, afterAnalysisManagementObjects);
} else {
// Add the top level AnalysisEngineManagement instance.
afterAnalysisManagementObjects.add(deepCopyMetrics(aem));
}
// Create a List to hold per CAS analysisTime and total number of CASes processed
// by each AE. This list will be serialized and sent to the client
List<AnalysisEnginePerformanceMetrics> performanceList =
new ArrayList<AnalysisEnginePerformanceMetrics>();
// Diff the before process() performance metrics with post process performance
// metrics
for (AnalysisEnginePerformanceMetrics after : afterAnalysisManagementObjects) {
for( AnalysisEnginePerformanceMetrics before: beforeAnalysisManagementObjects) {
if ( before.getUniqueName().equals(after.getUniqueName())) {
AnalysisEnginePerformanceMetrics metrics =
new AnalysisEnginePerformanceMetrics(after.getName(),
after.getUniqueName(),
after.getAnalysisTime()- before.getAnalysisTime(),
after.getNumProcessed());
performanceList.add(metrics);
break;
}
}
}
// Save this component performance metrics
parentCasStateEntry.getAEPerformanceList().addAll(performanceList);
if (!anEndpoint.isRemote()) {
inputCASReturned = true;
UimaTransport transport = getTransport(anEndpoint.getEndpoint());
if (getInProcessCache() != null && getInProcessCache().getSize() > 0
&& getInProcessCache().entryExists(aCasReferenceId)) {
try {
CacheEntry ancestor =
getInProcessCache().
getTopAncestorCasEntry(getInProcessCache().getCacheEntryForCAS(aCasReferenceId));
if ( ancestor != null ) {
// Set a flag on the input CAS to indicate that the processing was aborted
ancestor.addDelegateMetrics(getKey(), performanceList);
}
} 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.
}
}
UimaMessage message = transport.produceMessage(AsynchAEMessage.Process,
AsynchAEMessage.Response, getName());
message.addStringProperty(AsynchAEMessage.CasReference, aCasReferenceId);
ServicePerformance casStats = getCasStatistics(aCasReferenceId);
message.addLongProperty(AsynchAEMessage.TimeToSerializeCAS, casStats
.getRawCasSerializationTime());
message.addLongProperty(AsynchAEMessage.TimeToDeserializeCAS, casStats
.getRawCasDeserializationTime());
message.addLongProperty(AsynchAEMessage.TimeInProcessCAS, casStats.getRawAnalysisTime());
long iT = getIdleTimeBetweenProcessCalls(AsynchAEMessage.Process);
message.addLongProperty(AsynchAEMessage.IdleTime, iT);
// Send reply back to the client. Use internal (non-jms) transport
if (!stopped) {
transport.getUimaMessageDispatcher(anEndpoint.getEndpoint()).dispatch(message);
dropStats(aCasReferenceId, getName());
}
} else {
try {
CacheEntry entry =
getInProcessCache().getCacheEntryForCAS(aCasReferenceId);
entry.addDelegateMetrics(getKey(), performanceList);
} 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.