in uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/engine/ProcessingUnit.java [758:1275]
protected boolean processNext(Object[] aCasObjectList, ProcessTrace pTrTemp)
throws ResourceProcessException, IOException, CollectionException, AbortCPMException,
KillPipelineException {
maybeLogFinest("UIMA_CPM_start_analysis__FINEST");
// String lastDocId = "";
CasProcessor processor = null;
// This is used to hold an index of the current CasObject
// int currentIndex = -1;
boolean doneAlready = false;
// If there are no CASes in the list, return false since there is nothing else to do
if (aCasObjectList == null || aCasObjectList[0] == null) {
maybeLogFinest("UIMA_CPM_invalid_cas_reference__SEVERE");
return false;
}
Object[] casObjects = null;
// Determine if the Cas'es contained in the CasList are of type CAS. Samples the first CAS in
// the list.
// The list contains CASes of the same type ( either CasData or CAS ). Mixed model not
// supported.
boolean isCasObject = aCasObjectList[0] instanceof CAS;
// String docid = "";
maybeLogFinest("UIMA_CPM_entering_pipeline__FINEST");
ProcessingContainer container = null;
// *******************************************
// ** P R O C E S S I N G P I P E L I N E **
// *******************************************
// Send CAS Object through the processing pipeline.
for (int i = 0; processContainers != null && i < processContainers.size(); i++) {
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
logFinest("UIMA_CPM_retrieve_container__FINEST", String.valueOf(i));
}
container = (ProcessingContainer) processContainers.get(i);
synchronized (container) {
// Check to see if the CasProcessor is available for processing
if (!isProcessorReady(container.getStatus())) {
maybeLogFinest("UIMA_CPM_container_not_ready__FINEST", container);
boolean breakTheLoop = (i == (processContainers.size() - 1));
if (breakTheLoop && isCasObject) {
releaseCases(aCasObjectList, true, container.getName());
break;
}
// Skip any CasProcessor that is not ready to process
continue;
}
}
// Check if any of the Cas'es in the set has a required feature structure.
if (!isCasObject && !container.processCas(aCasObjectList)) {
maybeLogFinest("UIMA_CPM_skip_CAS__FINEST", container);
container.incrementFilteredCount(aCasObjectList.length);
container.logAbortedCases(aCasObjectList);
continue;
}
long byteCount;
// Flag controlling do-while loop that facilitates retries. Retries are defined in the
// CasProcessor configuration.
boolean retry = false;
// Retry Loop.
do {
if (System.getProperty("SHOW_MEMORY") != null) {
maybeLogMemoryFinest();
}
maybeLogFinest("UIMA_CPM_checkout_cp_from_container__FINEST", container);
threadState = 2004; // Entering dequeue()
processor = container.getCasProcessor();
if (processor == null) {
maybeLogSevere("UIMA_CPM_checkout_null_cp_from_container__SEVERE", container.getName());
throw new ResourceProcessException(CpmLocalizedMessage.getLocalizedMessage(
CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
"UIMA_CPM_EXP_invalid_component_reference__WARNING",
new Object[] { Thread.currentThread().getName(), "CasProcessor", "NULL" }), null);
}
// Check to see if the CasProcessor is available for processing
// Container may have been disabled by another thread, so first check
if (!isProcessorReady(container.getStatus())) {
maybeLogFinest("UIMA_CPM_container_not_ready__FINEST", container);
if (container.getStatus() == Constants.CAS_PROCESSOR_KILLED) {
container.releaseCasProcessor(processor);
// Another thread has initiated CPM Abort. That Thread has already notified
// the application of the Abort. Here we just return as the CPM has been
// killed most likely due to excessive errors.
return false;
}
// Skip any CasProcessor that is not ready to process
break;
}
maybeLogFinest("UIMA_CPM_checkedout_cp_from_container__FINEST", container, processor);
try {
if (processor instanceof CasDataProcessor) {
maybeLogFinest("UIMA_CPM_cas_data_processor__FINEST", container, processor);
pTrTemp.startEvent(container.getName(), "Process", "");
if (isCasObject) {
CasData[] casDataObjects = new CasData[aCasObjectList.length];
for (int casIndex = 0; casIndex < aCasObjectList.length; casIndex++) {
casDataObjects[casIndex] = mConverter
.casContainerToCasData((CAS) aCasObjectList[casIndex]);
if ((CAS) aCasObjectList[casIndex] != null) {
((CAS) aCasObjectList[casIndex]).reset();
}
}
casCache = (CAS[]) aCasObjectList;
aCasObjectList = casDataObjects;
}
isCasObject = false;
byteCount = 0;
if (!retry) {
for (int casIndex = 0; casIndex < aCasObjectList.length; casIndex++) {
byteCount = getBytes(aCasObjectList[casIndex]);
container.addBytesIn(byteCount);
}
}
casObjects = aCasObjectList;
long pStart = System.currentTimeMillis();
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
logFinest("UIMA_CPM_call_process__FINEST", container, processor);
logFinest("UIMA_CPM_casObjects_class__FINEST", casObjects.getClass().getName());
}
if (!(casObjects instanceof CasData[])) {
maybeLogFinest("UIMA_CPM_expected_casdata__FINEST", casObjects.getClass().getName());
}
maybeLogFinest("UIMA_CPM_call_process__FINEST", container, processor);
casObjects = ((CasDataProcessor) processor).process((CasData[]) casObjects);
maybeLogFinest("UIMA_CPM_call_process_completed__FINEST", container, processor);
long pEnd = System.currentTimeMillis();
container.incrementTotalTime((pEnd - pStart));
if (casObjects != null) {
if (processor instanceof CasDataConsumer) {
container.addBytesOut(byteCount);
} else {
aCasObjectList = casObjects;
if (!retry) {
for (int casIndex = 0; casIndex < aCasObjectList.length; casIndex++) {
byteCount = getBytes(aCasObjectList[casIndex]);
container.addBytesOut(byteCount);
}
}
}
}
pTrTemp.endEvent(container.getName(), "Process", "success");
} else if (processor instanceof CasObjectProcessor) {
maybeLogFinest("UIMA_CPM_casobject_processor__FINEST", container, processor);
maybeLogMemoryFinest();
casList = new CAS[aCasObjectList.length];
for (int casIndex = 0; casIndex < aCasObjectList.length; casIndex++) {
maybeLogFinest("UIMA_CPM_initialize_cas__FINEST", container);
if (aCasObjectList[casIndex] == null) {
if (UIMAFramework.getLogger().isLoggable(Level.SEVERE)) {
logSevere("UIMA_CPM_casobjectlist_is_null__SEVERE", container.getName(),
String.valueOf(casIndex));
}
break;
}
if (!isCasObject) {
// The following may be true if the CollectionReader is CasData based and this is
// the first CasObject based annotator in the chain.
if (casCache == null || casCache[casIndex] == null) {
casList[casIndex] = null;
while (casList[casIndex] == null) {
maybeLogFinest("UIMA_CPM_get_cas_from_pool__FINEST", container);
// Retrieve a Cas from Cas Pool. Wait max 10 millis for an instance
casList[casIndex] = casPool.getCas(0);
maybeLogFinest("UIMA_CPM_got_cas_from_pool__FINEST", container);
}
if (casList[casIndex] != null) {
maybeLogFinest("UIMA_CPM_call_cas_reset__FINEST", container);
casList[casIndex].reset();
}
} else {
casList[casIndex] = casCache[casIndex];
casList[casIndex].reset();
maybeLogFinest("UIMA_CPM_nullify_cas__FINEST", container);
// Cas is used up
casCache[casIndex] = null;
}
// Convert CasData to CAS
mConverter.casDataToCasContainer((CasData) aCasObjectList[casIndex],
casList[casIndex], true);
} else {
casList[casIndex] = (CAS) aCasObjectList[casIndex];
}
// Set the type from CasData to CasObject. When an error occurs in the proces()
// we need to know what type of object we deal with.
isCasObject = true;
aCasObjectList = casList;
if (processor instanceof AnalysisEngine) {
maybeLogFinest("UIMA_CPM_call_process__FINEST", container, processor);
threadState = 2005;
pTrTemp.aggregate(((AnalysisEngine) processor).process(casList[casIndex]));
maybeLogFinest("UIMA_CPM_call_process_completed__FINEST", container, processor);
} else {
pTrTemp.startEvent(container.getName(), "Process", "");
threadState = 2006;
maybeLogFinest("UIMA_CPM_call_process__FINEST", container, processor);
((CasObjectProcessor) processor).processCas(casList[casIndex]);
maybeLogFinest("UIMA_CPM_call_process_completed__FINEST", container, processor);
pTrTemp.endEvent(container.getName(), "Process", "success");
}
}
}
// Release the CAS and notify listeners if the end of the
// pipeline is reached.
if ((releaseCAS) && (i == (processContainers.size() - 1))) {
// This flag is used to prevent multiple notifications
doneAlready = true;
EntityProcessStatus aEntityProcStatus = new EntityProcessStatusImpl(pTrTemp);
maybeLogFinest("UIMA_CPM_notify_listeners__FINEST");
threadState = 2007;
notifyListeners(aCasObjectList, isCasObject, aEntityProcStatus);
if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
logFinest("UIMA_CPM_done_notify_listeners__FINEST");
logFinest("UIMA_CPM_releasing_cases__FINEST", container.getName(),
String.valueOf(releaseCAS), "true");
}
if (casCache != null) {
clearCasCache();
}
// Release CAS's.
if (aCasObjectList instanceof CAS[]) {
cpm.releaseCASes((CAS[]) aCasObjectList);
}
maybeLogFinest("UIMA_CPM_done_releasing_cases__FINEST", container);
}
maybeLogFinest("UIMA_CPM_pipeline_completed__FINEST");
retry = false;
// On successfull processing reset the restart counter. Restart counter determines how
// many times to restart Cas Processor on the same CAS
// Do this conditionally. If the CAS is to be dropped on Exception this restart counter
// scope extends to the entire collection not just one CAS
if (!cpm.dropCasOnException()) {
container.resetRestartCount();
}
} catch (Exception e) {
e.printStackTrace();
if (UIMAFramework.getLogger().isLoggable(Level.SEVERE)) {
logSevere("UIMA_CPM_pipeline_exception__SEVERE", container.getName(), e.getMessage());
maybeLogSevereException(e);
logFinest("UIMA_CPM_pipeline_exception__FINEST", container.getName(),
String.valueOf(container.isPaused()));
}
EntityProcessStatusImpl enProcSt = new EntityProcessStatusImpl(pTrTemp);
enProcSt.addEventStatus("Process", "Failed", e);
threadState = 2008;
notifyListeners(aCasObjectList, isCasObject, enProcSt);
doneAlready = true;
threadState = 2009;
// Check the policy to determine what to do with the CAS on exception. Return the CAS back
// to the pool
// and stop the processing chain if required. The policy for what to do with the CAS on
// exception is
// defined in the CPE descriptor
if (cpm.dropCasOnException()) {
if (casCache != null) {
clearCasCache();
}
if (aCasObjectList instanceof CAS[]) {
cpm.invalidateCASes((CAS[]) aCasObjectList);
}
retry = false; // Dont retry. The CAS has been released
maybeLogWarning("UIMA_CPM_drop_cas__WARNING", container.getName(),
processor.getClass().getName());
} else {
retry = true; // default on Exception
}
// If the container is in pause state dont increment errors since one thread has already
// done this. While the container is in pause state the CPM is attempting to re-connect
// to a failed service. Once that is done, the container is going to be resumed. While
// in pause state ALL threads using the container will be suspended.
if (processor instanceof CasObjectNetworkCasProcessorImpl && container.isPaused()) {
threadState = 2010;
maybeLogFinest("UIMA_CPM_container_paused_do_retry__FINEST", container);
// Do conditional release of CAS instances. The release occurs only if the CasProcessor
// is the last one
// in processing chain. This only releases instances of CAS checked out from the Cas
// Pool. If not done, the
// pool gets depleted and no more work will be done.
releaseCases(casList, (i == (processContainers.size() - 1)), container.getName());
maybeLogFinest("UIMA_CPM_container_paused__FINEST", container);
// Release current Cas Processor before continuing with the next Cas Processor in the
// pipeline
if (processor != null) {
container.releaseCasProcessor(processor);
maybeLogFinest("UIMA_CPM_ok_released_cp__FINEST", container);
processor = null;
}
try {
pTrTemp.endEvent(container.getName(), "Process", "failed");
} catch (Exception exc) {
// Just ignore out-of-phase endEvent exceptions for now.
}
continue;
}
if (pauseContainer(container, e, threadId)) // container.isRemote() &&
// container.isSingleFencedService() &&
// threadId == null )
{
threadState = 2011;
// Pause the container while the CPM is re-connecting to un-managed service
// that is shared by all processing threads
container.pause();
maybeLogFinest("UIMA_CPM_pausing_container__FINEST", container);
threadId = Thread.currentThread().getName();
}
if (processor instanceof CasDataProcessor || (processor instanceof CasObjectProcessor
&& !(processor instanceof AnalysisEngine))) {
try {
pTrTemp.endEvent(container.getName(), "Process", "failed");
} catch (Exception exc) {
// Just ignore out-of-phase endEvent exceptions for now.
}
}
try {
// Increments error counter and determines if any threshold have been reached. If
// the max error rate is reached, the CasProcessor can be configured as follows:
// - terminates CPM when threshold is reached ( method below throws AbortCPMException)
// - disables CasProcessor ( method below throws AbortCasProcessorException )
// - continue, CasProcessor continues to run dispite error
container.incrementCasProcessorErrors(e);
container.releaseCasProcessor(processor);
processor = null;
if (cpm.dropCasOnException()) {
// Cas has already been returned to the CAS pool. The policy requires to stop the
// processing chain for this CAS and
// to get another CAS for processing.
return true;
} else {
container.incrementRetryCount(1);
continue;
}
} // check if the exception should terminate the CPM
catch (KillPipelineException ex) {
try {
handleKillPipeline(container);
processor = null;
} catch (Exception innerE) {
maybeLogWarning("UIMA_CPM_exception_on_pipeline_kill__WARNING", container.getName(),
innerE.getMessage());
}
// finally
// {
// // Throw Original Exception - Killing Pipeline
// throw ex;
// }
throw ex;
} catch (AbortCPMException ex) {
try {
handleAbortCPM(container, processor);
} catch (Exception innerE) {
maybeLogWarning("UIMA_CPM_exception_on_cpm_kill__WARNING", container.getName(),
innerE.getMessage());
}
// finally
// {
// throw new AbortCPMException("Aborting CPM. CasProcessor::" + container.getName() + "
// Configured to Abort the CPM.");
// }
throw ex;
} // check if the CasProcessor is to be disabled due to excessive errors
catch (AbortCasProcessorException ex) {
try {
handleAbortCasProcessor(container, processor);
if (cpm.dropCasOnException()) {
// Cas has already been returned to the CAS pool. The policy requires to stop the
// processing chain for this CAS and
// to get another CAS for processing.
return true;
} else {
// Do conditional release of CAS instances. The release occurs only if the
// CasProcessor is the last one
// in processing chain. This only releases instances of CAS checked out from the Cas
// Pool. If not done, the
// pool gets depleted and no more work will be done.
releaseCases(casList, (i == (processContainers.size() - 1)), container.getName());
}
} catch (ResourceProcessException rpe) {
throw rpe;
} catch (Exception rpe) {
throw new ResourceProcessException(rpe);
}
break; // CasProcessor disabled move on to the next one
} // check if need to redeploy the CasProcessor
catch (ServiceConnectionException ex) {
pTrTemp.startEvent(container.getName(), "Process", "");
try {
threadState = 2012;
handleServiceException(container, processor, pTrTemp, ex);
// processor = null;
if (cpm.dropCasOnException()) {
return true;
} else {
// Increment number of restarts
container.incrementRestartCount(1);
pTrTemp.endEvent(container.getName(), "Process", "success");
continue; // retry the same CAS'es
}
} catch (ResourceProcessException rpe) {
pTrTemp.endEvent(container.getName(), "Process", "failure");
throw rpe;
} catch (ResourceConfigurationException rpe) {
if (rpe.getCause() != null && rpe.getCause() instanceof KillPipelineException) {
try {
handleKillPipeline(container);
processor = null;
} catch (Exception excep) {
// Just log the exception. We are killing the pipeline
maybeLogWarning("UIMA_CPM_exception_on_pipeline_kill__WARNING",
container.getName(), excep.getMessage());
}
}
pTrTemp.endEvent(container.getName(), "Process", "failure");
throw new ResourceProcessException(rpe.getCause());
} catch (Exception rpe) {
pTrTemp.endEvent(container.getName(), "Process", "failure");
throw new ResourceProcessException(rpe);
}
} catch (SkipCasException ex) {
try {
// Release current Cas Processor before continuing with the next Cas Processor in the
// pipeline
if (processor != null) {
container.releaseCasProcessor(processor);
}
handleSkipCasProcessor(container, aCasObjectList,
(i == (processContainers.size() - 1)));
} catch (Exception sEx) {
throw new ResourceProcessException(sEx);
}
processor = null;
if (cpm.dropCasOnException()) {
return true;
} else {
break;
}
} catch (Exception ex) {
if (UIMAFramework.getLogger().isLoggable(Level.FINER)) {
logCPM(Level.FINER, "UIMA_CPM_exception__FINER", new Object[] { ex.getMessage() });
ex.printStackTrace();
}
}
// CAS may have already been dropped (released) due to dropCasOnException policy defined
// in the CPE Descriptor
if (!cpm.dropCasOnException()) {
// Do conditional release of CAS instances. The release occurs only if the CasProcessor
// is the last one
// in processing chain. This only releases instances of CAS checked out from the Cas
// Pool. If not done, the
// pool gets depleted and no more work will be done.
releaseCases(casList, (i == (processContainers.size() - 1)), container.getName());
container.incrementRetryCount(1);
}
} // catch
// Let the container take action if the end-of-batch marker has been reached.
// End-of-batch marker is defined in the cpm configuration for every CasProcessor.
// This marker is defined in the <checkpoint> section of the CasProcessor Definition
// and corresponds to the attribute "batch". If end-of-batch marker is reached the container
// invokes batchProcessComplete() on the CasProcessor
maybeLogFinest("UIMA_CPM_end_of_batch__FINEST", container, processor);
doEndOfBatchProcessing(container, processor, pTrTemp, aCasObjectList);
processor = null;
} while (retry); // retry loop
if (processor != null) {
maybeLogFinest("UIMA_CPM_release_cp__FINEST", container, processor, casCache);
container.releaseCasProcessor(processor);
processor = null;
maybeLogFinest("UIMA_CPM_ok_release_cp__FINEST", container, processor, casCache);
}
} // end of: For All CasProcessors
try {
postAnalysis(aCasObjectList, isCasObject, casObjects, pTrTemp, doneAlready);
} catch (ResourceProcessException rpe) {
throw rpe;
} catch (Exception rpe) {
throw new ResourceProcessException(rpe);
}
maybeLogFinest("UIMA_CPM_pipeline_completed__FINEST");
return true;
}