| /* |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| */ |
| |
| package org.apache.uima.collection.impl.cpm.engine; |
| |
| import java.util.ArrayList; |
| import java.util.LinkedList; |
| |
| import org.apache.uima.UIMAFramework; |
| import org.apache.uima.analysis_engine.AnalysisEngine; |
| import org.apache.uima.cas.CAS; |
| import org.apache.uima.cas_data.CasData; |
| import org.apache.uima.collection.CollectionException; |
| import org.apache.uima.collection.EntityProcessStatus; |
| import org.apache.uima.collection.StatusCallbackListener; |
| import org.apache.uima.collection.base_cpm.AbortCPMException; |
| import org.apache.uima.collection.base_cpm.AbortCasProcessorException; |
| import org.apache.uima.collection.base_cpm.BaseStatusCallbackListener; |
| import org.apache.uima.collection.base_cpm.CasDataConsumer; |
| import org.apache.uima.collection.base_cpm.CasDataProcessor; |
| import org.apache.uima.collection.base_cpm.CasDataStatusCallbackListener; |
| import org.apache.uima.collection.base_cpm.CasObjectProcessor; |
| import org.apache.uima.collection.base_cpm.CasProcessor; |
| import org.apache.uima.collection.base_cpm.SkipCasException; |
| import org.apache.uima.collection.impl.CasConverter; |
| import org.apache.uima.collection.impl.EntityProcessStatusImpl; |
| import org.apache.uima.collection.impl.base_cpm.container.ProcessingContainer; |
| import org.apache.uima.collection.impl.base_cpm.container.ServiceConnectionException; |
| import org.apache.uima.collection.impl.base_cpm.container.deployer.CasProcessorDeployer; |
| import org.apache.uima.collection.impl.cpm.Constants; |
| import org.apache.uima.collection.impl.cpm.container.CasObjectNetworkCasProcessorImpl; |
| import org.apache.uima.collection.impl.cpm.container.NetworkCasProcessorImpl; |
| import org.apache.uima.collection.impl.cpm.utils.CPMUtils; |
| import org.apache.uima.collection.impl.cpm.utils.CpmLocalizedMessage; |
| import org.apache.uima.collection.impl.cpm.vinci.DATACasUtils; |
| import org.apache.uima.collection.metadata.CpeConfiguration; |
| import org.apache.uima.resource.ResourceProcessException; |
| import org.apache.uima.util.Level; |
| import org.apache.uima.util.ProcessTrace; |
| import org.apache.uima.util.UimaTimer; |
| import org.apache.uima.util.impl.ProcessTrace_impl; |
| |
| /** |
| * |
| */ |
| public class NonThreadedProcessingUnit { |
| public int threadState = 0; |
| |
| protected CPECasPool casPool; |
| |
| protected boolean relaseCAS = false; |
| |
| protected CPMEngine cpm = null; |
| |
| protected BoundedWorkQueue workQueue = null; |
| |
| protected BoundedWorkQueue outputQueue = null; |
| |
| protected CasConverter mConverter; |
| |
| protected ProcessTrace processingUnitProcessTrace; |
| |
| protected LinkedList processContainers = new LinkedList(); |
| |
| protected long numToProcess = 0; |
| |
| protected CAS[] casList; |
| |
| protected ArrayList statusCbL = new ArrayList(); |
| |
| protected boolean notifyListeners = false; |
| |
| protected CAS conversionCas = null; |
| |
| protected Object[] artifact = null; |
| |
| protected CAS[] conversionCasArray; |
| |
| protected UimaTimer timer; |
| |
| protected String threadId = null; |
| |
| protected CpeConfiguration cpeConfiguration = null; |
| |
| private CAS[] casCache = null; |
| |
| /** |
| * |
| */ |
| /** |
| * Initialize the PU |
| * |
| * @param acpm - |
| * component managing life cycle of the CPE |
| * @param aInputQueue - |
| * queue to read from |
| * @param aOutputQueue - |
| * queue to write to |
| */ |
| public NonThreadedProcessingUnit(CPMEngine acpm, BoundedWorkQueue aInputQueue, |
| BoundedWorkQueue aOutputQueue) { |
| cpm = acpm; |
| try { |
| cpeConfiguration = cpm.getCpeConfig(); |
| } catch (Exception e) { |
| } |
| workQueue = aInputQueue; |
| conversionCasArray = new CAS[1]; |
| // Instantiate a class responsible for converting CasData to CasObject and vice versa |
| mConverter = new CasConverter(); |
| |
| outputQueue = aOutputQueue; |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_initialize_pipeline__FINEST", |
| new Object[] { Thread.currentThread().getName(), workQueue.getName(), |
| String.valueOf(workQueue.getCurrentSize()) }); |
| } |
| } |
| |
| public NonThreadedProcessingUnit(CPMEngine acpm) { |
| cpm = acpm; |
| try { |
| cpeConfiguration = cpm.getCpeConfig(); |
| } catch (Exception e) { |
| } |
| conversionCasArray = new CAS[1]; |
| // Instantiate a class responsible for converting CasData to CasObject and vice versa |
| mConverter = new CasConverter(); |
| } |
| |
| /** |
| * Alternative method of providing a queue from which this PU will read bundle of Cas |
| * |
| * @param aInputQueue - |
| * read queue |
| */ |
| public void setInputQueue(BoundedWorkQueue aInputQueue) { |
| workQueue = aInputQueue; |
| } |
| |
| /** |
| * Alternative method of providing a queue where this PU will deposit results of analysis |
| * |
| * @param aOutputQueue - |
| * queue to write to |
| */ |
| public void setOutputQueue(BoundedWorkQueue aOutputQueue) { |
| outputQueue = aOutputQueue; |
| } |
| |
| /** |
| * Alternative method of providing the reference to the component managing the lifecycle of the |
| * CPE |
| * |
| * @param acpm - |
| * reference to the contrlling engine |
| */ |
| public void setCPMEngine(CPMEngine acpm) { |
| cpm = acpm; |
| } |
| |
| /** |
| * Null out fields of this object. Call this only when this object is no longer needed. |
| */ |
| public void cleanup() { |
| this.casPool = null; |
| this.cpm = null; |
| this.workQueue = null; |
| this.outputQueue = null; |
| this.mConverter = null; |
| this.processingUnitProcessTrace = null; |
| this.processContainers.clear(); |
| this.processContainers = null; |
| this.casList = null; |
| this.conversionCas = null; |
| this.artifact = null; |
| this.statusCbL = null; |
| this.conversionCasArray = null; |
| } |
| |
| /** |
| * Set a flag indicating if notifications should be made via configured Listeners |
| * |
| * @param aDoNotify - |
| * true if notification is required, false otherwise |
| */ |
| public void setNotifyListeners(boolean aDoNotify) { |
| notifyListeners = aDoNotify; |
| } |
| |
| /** |
| * Plugs in Listener object used for notifications. |
| * |
| * @param aListener - |
| * {@link org.apache.uima.collection.base_cpm.BaseStatusCallbackListener} instance |
| */ |
| public void addStatusCallbackListener(BaseStatusCallbackListener aListener) { |
| statusCbL.add(aListener); |
| } |
| |
| /** |
| * Returns list of listeners used by this PU for callbacks. |
| * |
| * @return - lif of {@link org.apache.uima.collection.base_cpm.BaseStatusCallbackListener} |
| * instances |
| */ |
| public ArrayList getCallbackListeners() { |
| return statusCbL; |
| } |
| |
| /** |
| * Removes given listener from the list of listeners |
| * |
| * @param aListener - |
| * object to remove from the list |
| */ |
| public void removeStatusCallbackListener(BaseStatusCallbackListener aListener) { |
| statusCbL.remove(aListener); |
| } |
| |
| /** |
| * Plugs in ProcessTrace object used to collect statistics |
| * |
| * @param aProcessingUnitProcessTrace - |
| * object to compile stats |
| */ |
| public void setProcessingUnitProcessTrace(ProcessTrace aProcessingUnitProcessTrace) { |
| processingUnitProcessTrace = aProcessingUnitProcessTrace; |
| |
| } |
| |
| /** |
| * Plugs in custom timer used by the PU for getting time |
| * |
| * @param aTimer - |
| * custom timer to use |
| */ |
| public void setUimaTimer(UimaTimer aTimer) { |
| timer = aTimer; |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_timer_class__FINEST", |
| new Object[] { Thread.currentThread().getName(), timer.getClass().getName() }); |
| } |
| } |
| |
| /** |
| * Plugs in a list of Cas Processor containers. During processing Cas Processors in this list are |
| * called sequentially. Each Cas Processor is contained in the container that is managing errors, |
| * counts and totals, and restarts. |
| * |
| * @param aProcessor |
| * CASProcessor to be added to the processing pipeline |
| */ |
| public void setContainers(LinkedList processorList) { |
| processContainers = processorList; |
| } |
| |
| /** |
| * |
| * Disable a CASProcessor in the processing pipeline. Locate it by provided index. The disabled |
| * Cas Processor remains in the Processing Pipeline, however it is not used furing processing. |
| * |
| * @param aCasProcessorIndex - |
| * location in the pipeline of the Cas Processor to delete |
| */ |
| public void disableCasProcessor(int aCasProcessorIndex) { |
| if (aCasProcessorIndex < 0 || aCasProcessorIndex > processContainers.size()) { |
| return; |
| } |
| // Retrive container with a reference to the CasProcessor |
| ProcessingContainer pc = ((ProcessingContainer) processContainers.get(aCasProcessorIndex)); |
| if (pc != null) { |
| pc.setStatus(Constants.CAS_PROCESSOR_DISABLED); |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_disabled_cp__FINEST", |
| new Object[] { Thread.currentThread().getName(), pc.getName() }); |
| } |
| } |
| } |
| |
| /** |
| * |
| * Alternative method to disable Cas Processor. Uses a name to locate it. |
| * |
| * @param aCasProcessorName - |
| * a name of the Cas Processor to disable |
| */ |
| public void disableCasProcessor(String aCasProcessorName) { |
| for (int i = 0; i < processContainers.size(); i++) { |
| ProcessingContainer pc = ((ProcessingContainer) processContainers.get(i)); |
| if (pc.getName().equals(aCasProcessorName)) { |
| pc.setStatus(Constants.CAS_PROCESSOR_DISABLED); |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_disabled_cp__FINEST", |
| new Object[] { Thread.currentThread().getName(), pc.getName() }); |
| } |
| } |
| } |
| } |
| |
| /** |
| * |
| * Enables Cas Processor with a given name. Enabled Cas Processor will immediately begin to |
| * receive bundles of Cas. |
| * |
| * @param aCasProcessorName - |
| * name of the Cas Processor to enable |
| */ |
| public void enableCasProcessor(String aCasProcessorName) { |
| for (int i = 0; i < processContainers.size(); i++) { |
| ProcessingContainer pc = ((ProcessingContainer) processContainers.get(i)); |
| if (pc.getName().equals(aCasProcessorName)) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_enabled_cp__FINEST", |
| new Object[] { Thread.currentThread().getName(), pc.getName() }); |
| } |
| pc.setStatus(Constants.CAS_PROCESSOR_RUNNING); |
| } |
| } |
| } |
| |
| protected boolean analyze(Object[] aCasObjectList, ProcessTrace pTrTemp) throws Exception // throws |
| // ResourceProcessException, |
| // IOException, |
| // CollectionException, |
| // AbortCPMException |
| { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_start_analysis__FINEST", |
| new Object[] { Thread.currentThread().getName() }); |
| } |
| // String lastDocId = ""; |
| CasProcessor processor = null; |
| // This is used to hold an index of the current CasObject |
| 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) { |
| if (UIMAFramework.getLogger().isLoggable(Level.SEVERE)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.SEVERE, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_invalid_cas_reference__SEVERE", |
| new Object[] { Thread.currentThread().getName() }); |
| } |
| 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 = ""; |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_entering_pipeline__FINEST", |
| new Object[] { Thread.currentThread().getName() }); |
| } |
| |
| ProcessingContainer container = null; |
| String containerName = ""; |
| // ******************************************* |
| // ** 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)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_retrieve_container__FINEST", |
| new Object[] { Thread.currentThread().getName(), String.valueOf(i) }); |
| } |
| // Retrieve the container. Container manages one or more instances of CAS Processor |
| container = (ProcessingContainer) processContainers.get(i); |
| // container can be disabled in multi-processing pipeline configurations. The container is |
| // disabled |
| // when one of the processing threads is in the process of restarting/reconnecting to a shared |
| // fenced service. Shared, meaning that all processing pipelines use the same service for |
| // invocations. |
| // Container must be disabled to prevent concurrent restarts. |
| if (containerDisabled(container) || filterOutTheCAS(container, isCasObject, aCasObjectList)) { |
| continue; |
| } |
| containerName = container.getName(); |
| |
| // Flag controlling do-while loop that facilitates retries. Retries are defined in the |
| // CasProcessor configuration. |
| boolean retry = false; |
| do // Retry |
| { |
| |
| try { |
| if (System.getProperty("SHOW_MEMORY") != null) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_show_memory__FINEST", |
| new Object[] { Thread.currentThread().getName(), |
| String.valueOf(Runtime.getRuntime().totalMemory() / 1024), |
| String.valueOf(Runtime.getRuntime().freeMemory() / 1024) }); |
| } |
| } |
| |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_checkout_cp_from_container__FINEST", |
| new Object[] { Thread.currentThread().getName(), containerName }); |
| } |
| threadState = 2004; |
| // Get the CasProcessor from the pool managed by the container |
| processor = container.getCasProcessor(); |
| if (processor == null) { |
| if (UIMAFramework.getLogger().isLoggable(Level.SEVERE)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.SEVERE, |
| this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_checkout_null_cp_from_container__SEVERE", |
| new Object[] { Thread.currentThread().getName(), containerName }); |
| } |
| 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 |
| // The CasProcessor may have been disabled due to excessive errors and error policy |
| // defined |
| // in the CPE descriptor. |
| if (!isProcessorReady(container.getStatus())) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, |
| this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_container_not_ready__FINEST", |
| new Object[] { Thread.currentThread().getName(), containerName }); |
| } |
| 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; |
| } |
| |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_checkedout_cp_from_container__FINEST", |
| new Object[] { Thread.currentThread().getName(), containerName, |
| processor.getClass().getName() }); |
| } |
| // ************************* P E R F O R M A N A L Y S I S ************************* |
| if (processor instanceof CasDataProcessor) { |
| invokeCasDataCasProcessor(container, processor, aCasObjectList, pTrTemp, isCasObject, |
| retry); |
| isCasObject = false; |
| } else if (processor instanceof CasObjectProcessor) { |
| invokeCasObjectCasProcessor(container, processor, aCasObjectList, pTrTemp, isCasObject); |
| isCasObject = true; |
| } |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_analysis_successfull__FINEST", |
| new Object[] { Thread.currentThread().getName(), containerName, |
| processor.getClass().getName() }); |
| } |
| 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) { |
| retry = handleErrors(e, container, processor, pTrTemp, aCasObjectList, isCasObject); |
| if (cpm.dropCasOnException()) { |
| retry = false; // override |
| return false; // Dont pass the CAS to the CasConsumer. CAS has been dropped |
| } |
| } finally { |
| if (retry == false) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_end_of_batch__FINEST", |
| new Object[] { Thread.currentThread().getName(), containerName, |
| processor.getClass().getName() }); |
| } |
| if (isProcessorReady(container.getStatus())) { |
| |
| // 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 |
| doEndOfBatch(container, processor, pTrTemp, aCasObjectList.length); |
| } |
| } else { |
| container.incrementRetryCount(1); |
| } |
| // Release current Cas Processor before continuing with the next Cas Processor in the |
| // pipeline |
| if (processor != null) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_release_cp__FINEST", |
| new Object[] { Thread.currentThread().getName(), containerName, |
| processor.getClass().getName(), String.valueOf(casCache == null) }); |
| } |
| doReleaseCasProcessor(container, processor); |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_ok_release_cp__FINEST", |
| new Object[] { Thread.currentThread().getName(), containerName, |
| processor.getClass().getName(), String.valueOf(casCache == null) }); |
| } |
| processor = null; |
| } |
| |
| } |
| |
| } while (retry); |
| } // end of: For All CasProcessors |
| |
| postAnalysis(aCasObjectList, isCasObject, casObjects, pTrTemp, doneAlready); |
| casObjects = null; |
| return true; |
| } |
| |
| /** |
| * |
| * @param aFlag |
| */ |
| public void setReleaseCASFlag(boolean aFlag) { |
| relaseCAS = aFlag; |
| } |
| |
| /** |
| * @param tcasPool |
| */ |
| public void setCasPool(CPECasPool aPool) { |
| casPool = aPool; |
| } |
| |
| /** |
| * |
| * @param aCasObjectList |
| * @param isCasObject |
| * @param casObjects |
| * @param aProcessTr |
| * @param doneAlready |
| * @throws Exception |
| */ |
| private void postAnalysis(Object[] aCasObjectList, boolean isCasObject, Object[] casObjects, |
| ProcessTrace aProcessTr, boolean doneAlready) throws Exception { |
| try { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_pipeline_completed__FINEST", |
| new Object[] { Thread.currentThread().getName() }); |
| } |
| // Notify Listeners that the entity has been processed. |
| if (!doneAlready && notifyListeners) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_notify_listeners__FINEST", |
| new Object[] { Thread.currentThread().getName() }); |
| } |
| threadState = 2013; |
| |
| EntityProcessStatus aEntityProcStatus = new EntityProcessStatusImpl(aProcessTr); |
| notifyListeners(aCasObjectList, isCasObject, aEntityProcStatus); |
| threadState = 2014; |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_done_notify_listeners__FINEST", |
| new Object[] { Thread.currentThread().getName() }); |
| } |
| } |
| // enqueue CASes. If the CPM is in shutdown mode due to hard kill dont allow enqueue of CASes |
| if (outputQueue != null |
| && (cpm.isRunning() == true || (cpm.isRunning() == false && cpm.isHardKilled() == false))) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_add_cas_to_queue__FINEST", |
| new Object[] { Thread.currentThread().getName(), outputQueue.getName(), |
| String.valueOf(outputQueue.getCurrentSize()) }); |
| } |
| WorkUnit workUnit = new WorkUnit(aCasObjectList); |
| if (casCache != null && casCache[0] != null) { |
| workUnit.setCas(casCache); |
| } |
| threadState = 2015; |
| |
| outputQueue.enqueue(workUnit); |
| |
| casCache = null; |
| |
| // synchronized (outputQueue) { // redundant - the above enqueue call does this |
| // outputQueue.notifyAll(); |
| // } |
| |
| } |
| return; |
| } catch (Exception e) { |
| throw e; |
| } finally { |
| // if (casObjects != null && casObjects instanceof CasData[]) |
| if (outputQueue == null && casObjects != null && casObjects instanceof CasData[]) { |
| if (System.getProperty("DEBUG_RELEASE") != null) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_done_with_cas__FINEST", |
| new Object[] { Thread.currentThread().getName(), |
| String.valueOf(Runtime.getRuntime().freeMemory() / 1024) }); |
| } |
| } |
| for (int i = 0; i < casObjects.length; i++) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_show_local_cache__FINEST", |
| new Object[] { Thread.currentThread().getName(), |
| String.valueOf(casCache == null) }); |
| } |
| casObjects[i] = null; |
| aCasObjectList[i] = null; |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_show_local_cache__FINEST", |
| new Object[] { Thread.currentThread().getName(), |
| String.valueOf(casCache == null) }); |
| } |
| } |
| if (System.getProperty("DEBUG_RELEASE") != null) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_show_total_memory__FINEST", |
| new Object[] { Thread.currentThread().getName(), |
| String.valueOf(Runtime.getRuntime().freeMemory() / 1024) }); |
| } |
| } |
| } |
| } |
| } |
| |
| private void doReleaseCasProcessor(ProcessingContainer aContainer, CasProcessor aCasProcessor) { |
| if (aCasProcessor != null && aContainer != null) { |
| aContainer.releaseCasProcessor(aCasProcessor); |
| } |
| } |
| |
| private void doEndOfBatch(ProcessingContainer aContainer, CasProcessor aProcessor, |
| ProcessTrace aProcessTr, int howManyCases) { |
| String containerName = aContainer.getName(); |
| try { |
| aContainer.isEndOfBatch(aProcessor, howManyCases); |
| |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_end_of_batch_completed__FINEST", |
| new Object[] { Thread.currentThread().getName(), containerName }); |
| } |
| } catch (Exception ex) { |
| if (UIMAFramework.getLogger().isLoggable(Level.SEVERE)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.SEVERE, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_end_of_batch_exception__SEVERE", |
| new Object[] { Thread.currentThread().getName(), containerName, ex.getMessage() }); |
| } |
| } |
| |
| } |
| |
| /** |
| * Main routine that handles errors occuring in the processing loop. |
| * |
| * @param e - |
| * exception in the main processing loop |
| * @param aContainer - |
| * current container of the Cas Processor |
| * @param aProcessor - |
| * current Cas Processor |
| * @param aProcessTrace - |
| * an object containing stats for this procesing loop |
| * @param aCasObjectList - |
| * list of CASes being analyzed |
| * @param isCasObject - |
| * determines type of CAS in the aCasObjectList ( CasData or CasObject) |
| * @return boolean |
| * @throws Exception |
| */ |
| private boolean handleErrors(Throwable e, ProcessingContainer aContainer, |
| CasProcessor aProcessor, ProcessTrace aProcessTrace, Object[] aCasObjectList, |
| boolean isCasObject) throws Exception { |
| boolean retry = true; |
| |
| String containerName = aContainer.getName(); |
| e.printStackTrace(); |
| if (UIMAFramework.getLogger().isLoggable(Level.SEVERE)) { |
| UIMAFramework.getLogger(this.getClass()).log(Level.SEVERE, Thread.currentThread().getName(), |
| e); |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.SEVERE, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_handle_exception__SEVERE", |
| new Object[] { Thread.currentThread().getName(), containerName, |
| aProcessor.getClass().getName(), e.getMessage() }); |
| } |
| |
| EntityProcessStatusImpl enProcSt = new EntityProcessStatusImpl(aProcessTrace); |
| enProcSt.addEventStatus("Process", "Failed", e); |
| threadState = 2008; |
| // Send exception notifications to all registered listeners |
| notifyListeners(aCasObjectList, isCasObject, enProcSt); |
| 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(); |
| } |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.WARNING, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_drop_cas__WARNING", |
| new Object[] { Thread.currentThread().getName(), containerName, |
| aProcessor.getClass().getName() }); |
| |
| // Release CASes and notify listeners |
| cpm.invalidateCASes((CAS[]) aCasObjectList); |
| retry = false; // Dont retry. The CAS has been released |
| } |
| // 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 (aProcessor instanceof CasObjectNetworkCasProcessorImpl && aContainer.isPaused()) { |
| threadState = 2010; |
| |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_container_paused_do_retry__FINEST", |
| new Object[] { Thread.currentThread().getName(), containerName }); |
| |
| } |
| return true; // retry |
| } |
| if (e instanceof Exception && pauseContainer(aContainer, (Exception) e, threadId)) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_pausing_container__FINEST", |
| new Object[] { Thread.currentThread().getName(), containerName }); |
| } |
| threadState = 2011; |
| |
| // New Code 02/23/05 |
| // Pause the container while the CPM is re-connecting to un-managed service |
| // that is shared by all processing threads |
| aContainer.pause(); |
| threadId = Thread.currentThread().getName(); |
| } |
| |
| 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 |
| aContainer.incrementCasProcessorErrors(e); |
| // End of new code |
| } // check if the exception should terminate the CPM |
| catch (AbortCPMException ex) { |
| retry = false; |
| if (aContainer.isPaused()) { |
| aContainer.resume(); |
| } |
| aContainer.setStatus(Constants.CAS_PROCESSOR_KILLED); |
| if (UIMAFramework.getLogger().isLoggable(Level.SEVERE)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.SEVERE, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_abort_cpm__SEVERE", |
| new Object[] { Thread.currentThread().getName(), aProcessor.getClass().getName() }); |
| } |
| throw new AbortCPMException(CpmLocalizedMessage.getLocalizedMessage( |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_EXP_configured_to_abort__WARNING", |
| new Object[] { Thread.currentThread().getName(), containerName })); |
| } // check if the CasProcessor is to be disabled due to excessive errors |
| catch (AbortCasProcessorException ex) { |
| retry = false; |
| |
| if (aContainer.isPaused()) { |
| aContainer.resume(); |
| } |
| if (UIMAFramework.getLogger().isLoggable(Level.SEVERE)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.SEVERE, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_disable_cp__SEVERE", |
| new Object[] { Thread.currentThread().getName(), aProcessor.getClass().getName() }); |
| } |
| aContainer.setStatus(Constants.CAS_PROCESSOR_DISABLED); |
| |
| } // check if need to redeploy the CasProcessor |
| catch (ServiceConnectionException ex) { |
| aProcessTrace.startEvent(containerName, "Process", ""); |
| String status = "failure"; |
| try { |
| threadState = 2012; |
| |
| handleServiceException(aContainer, aProcessor, aProcessTrace, ex); |
| // Increment number of restarts |
| // aContainer.incrementRestartCount(1); |
| status = "success"; |
| } catch (ResourceProcessException rpe) { |
| throw rpe; |
| } catch (Exception rpe) { |
| throw new ResourceProcessException(rpe); |
| } finally { |
| aProcessTrace.endEvent(containerName, "Process", status); |
| } |
| } catch (SkipCasException ex) { |
| try { |
| handleSkipCasProcessor(aContainer, aCasObjectList, false); |
| retry = false; |
| } catch (Exception sEx) { |
| throw new ResourceProcessException(sEx); |
| } |
| } catch (Exception ex) { |
| if (UIMAFramework.getLogger().isLoggable(Level.SEVERE)) { |
| UIMAFramework.getLogger(this.getClass()).log(Level.SEVERE, |
| Thread.currentThread().getName(), e); |
| } |
| retry = false; |
| ex.printStackTrace(); |
| } |
| return retry; |
| } |
| |
| /** |
| * |
| * @param container |
| * @param processor |
| * @param aCasObjectList |
| * @param pTrTemp |
| * @param isCasObject |
| * @throws Exception |
| */ |
| private void invokeCasObjectCasProcessor(ProcessingContainer container, CasProcessor processor, |
| Object[] aCasObjectList, ProcessTrace pTrTemp, boolean isCasObject) throws Exception { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_show_memory__FINEST", |
| new Object[] { Thread.currentThread().getName(), |
| String.valueOf(Runtime.getRuntime().totalMemory() / 1024), |
| String.valueOf(Runtime.getRuntime().freeMemory() / 1024) }); |
| |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_invoke_cp_process__FINEST", |
| new Object[] { Thread.currentThread().getName(), container.getName(), |
| processor.getClass().getName() }); |
| |
| } |
| casList = new CAS[aCasObjectList.length]; |
| for (int casIndex = 0; casIndex < aCasObjectList.length; casIndex++) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_initialize_cas__FINEST", |
| new Object[] { Thread.currentThread().getName(), container.getName() }); |
| } |
| if (aCasObjectList[casIndex] == null) { |
| if (UIMAFramework.getLogger().isLoggable(Level.SEVERE)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.SEVERE, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_casobjectlist_is_null__SEVERE", |
| new Object[] { Thread.currentThread().getName(), container.getName(), |
| String.valueOf(casIndex) }); |
| } |
| break; |
| } |
| if (isCasObject == false) { |
| convertCasDataToCasObject(casIndex, container.getName(), aCasObjectList); |
| } else { |
| casList[casIndex] = (CAS) aCasObjectList[casIndex]; |
| } |
| if (processor instanceof AnalysisEngine) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_call_process__FINEST", |
| new Object[] { Thread.currentThread().getName(), container.getName(), |
| processor.getClass().getName() }); |
| } |
| threadState = 2005; |
| |
| pTrTemp.aggregate(((AnalysisEngine) processor).process(casList[casIndex])); |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_call_process_completed__FINEST", |
| new Object[] { Thread.currentThread().getName(), container.getName(), |
| processor.getClass().getName() }); |
| } |
| } else { |
| pTrTemp.startEvent(container.getName(), "Process", ""); |
| threadState = 2006; |
| |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_call_process__FINEST", |
| new Object[] { Thread.currentThread().getName(), container.getName(), |
| processor.getClass().getName() }); |
| } |
| ((CasObjectProcessor) processor).processCas(casList[casIndex]); |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_call_process_completed__FINEST", |
| new Object[] { Thread.currentThread().getName(), container.getName(), |
| processor.getClass().getName() }); |
| } |
| pTrTemp.endEvent(container.getName(), "Process", "success"); |
| } |
| } |
| aCasObjectList = casList; |
| |
| } |
| |
| /** |
| * |
| * @param casIndex |
| * @param aContainerName |
| * @param aCasObjectList |
| * @throws Exception |
| */ |
| private void convertCasDataToCasObject(int casIndex, String aContainerName, |
| Object[] aCasObjectList) throws Exception { |
| // 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] = casPool.getCas(); |
| casList[casIndex] = null; |
| |
| while (casList[casIndex] == null) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_get_cas_from_pool__FINEST", |
| new Object[] { Thread.currentThread().getName(), aContainerName }); |
| } |
| |
| // Retrieve a Cas from Cas Pool. Wait max 10 millis for an instance |
| casList[casIndex] = casPool.getCas(0); |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_got_cas_from_pool__FINEST", |
| new Object[] { Thread.currentThread().getName(), aContainerName }); |
| } |
| } |
| if (casList[casIndex] != null) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_call_cas_reset__FINEST", |
| new Object[] { Thread.currentThread().getName(), aContainerName }); |
| } |
| casList[casIndex].reset(); |
| } |
| } else { |
| casList[casIndex] = casCache[casIndex]; |
| casList[casIndex].reset(); |
| // Cas is used up |
| casCache[casIndex] = null; |
| } |
| // Convert CasData to CAS |
| mConverter.casDataToCasContainer((CasData) aCasObjectList[casIndex], casList[casIndex], true); |
| } |
| |
| /** |
| * |
| * @param container |
| * @param processor |
| * @param aCasObjectList |
| * @param pTrTemp |
| * @param isCasObject |
| * @param retry |
| * @throws Exception |
| */ |
| private void invokeCasDataCasProcessor(ProcessingContainer container, CasProcessor processor, |
| Object[] aCasObjectList, ProcessTrace pTrTemp, boolean isCasObject, boolean retry) |
| throws Exception { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_invoke_cp_process__FINEST", |
| new Object[] { Thread.currentThread().getName(), container.getName(), |
| processor.getClass().getName() }); |
| } |
| pTrTemp.startEvent(container.getName(), "Process", ""); |
| // Check if the CasObject to CasData conversion is necessary |
| if (isCasObject == true) { |
| CasData[] casDataObjects = new CasData[aCasObjectList.length]; |
| for (int casIndex = 0; casIndex < aCasObjectList.length; casIndex++) { |
| casDataObjects[casIndex] = mConverter.casContainerToCasData((CAS) aCasObjectList[casIndex]); |
| // After the conversion reset the CAS so that it can be reused |
| if ((CAS) aCasObjectList[casIndex] != null) { |
| ((CAS) aCasObjectList[casIndex]).reset(); |
| } |
| } |
| // Cache the CAS list for possible reuse |
| casCache = (CAS[]) aCasObjectList; |
| // aCasObjectList is working list. |
| aCasObjectList = casDataObjects; |
| } |
| long byteCount = 0; |
| if (!retry) { |
| for (int casIndex = 0; casIndex < aCasObjectList.length; casIndex++) { |
| byteCount = getBytes(aCasObjectList[casIndex]); |
| container.addBytesIn(byteCount); |
| } |
| } |
| Object[] casObjects = aCasObjectList; |
| long pStart = System.currentTimeMillis(); |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_casobject_class__FINEST", |
| new Object[] { Thread.currentThread().getName(), container.getName(), |
| casObjects.getClass().getName() }); |
| } |
| if (!(casObjects instanceof CasData[])) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_expected_casdata_class__FINEST", |
| new Object[] { Thread.currentThread().getName(), container.getName(), |
| casObjects.getClass().getName() }); |
| } |
| } |
| |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_call_process__FINEST", |
| new Object[] { Thread.currentThread().getName(), container.getName(), |
| processor.getClass().getName() }); |
| } |
| casObjects = ((CasDataProcessor) processor).process((CasData[]) casObjects); |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_call_process_completed__FINEST", |
| new Object[] { Thread.currentThread().getName(), container.getName(), |
| processor.getClass().getName() }); |
| } |
| 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"); |
| |
| } |
| |
| private boolean containerDisabled(ProcessingContainer aContainer) { |
| synchronized (aContainer) { |
| // Check to see if the CasProcessor is available for processing |
| if (!isProcessorReady(aContainer.getStatus())) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_container_not_ready__FINEST", |
| new Object[] { Thread.currentThread().getName(), aContainer.getName() }); |
| } |
| // Skip any CasProcessor that is not ready to process. Cas Processors may be disabled during |
| // processing |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| /** |
| * Check if the CASProcessor status is available for processing |
| */ |
| protected boolean isProcessorReady(int aStatus) { |
| if (aStatus == Constants.CAS_PROCESSOR_READY || aStatus == Constants.CAS_PROCESSOR_RUNNING) { |
| return true; |
| } |
| |
| return false; |
| } |
| |
| private boolean filterOutTheCAS(ProcessingContainer aContainer, boolean isCasObject, |
| Object[] aCasObjectList) { |
| // Check if any of the Cas'es in the set has a required feature structure. |
| // This is currently only supported for the CasData instances and provides |
| // filtering mechanism |
| if (!isCasObject && !aContainer.processCas(aCasObjectList)) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_skip_CAS__FINEST", |
| new Object[] { Thread.currentThread().getName(), aContainer.getName() }); |
| } |
| aContainer.incrementFilteredCount(aCasObjectList.length); |
| aContainer.logAbortedCases(aCasObjectList); |
| return true; // skipped this CAS |
| } |
| return false; |
| |
| } |
| |
| /** |
| * Notifies Listeners of the fact that the pipeline has finished processing the current set Cas'es |
| * |
| * @param aCas - |
| * object containing an array of OR a single instance of Cas |
| * @param isCasObject - |
| * true if instance of Cas is of type Cas, false otherwise |
| * @param aEntityProcStatus - |
| * status object that may contain exceptions and trace |
| */ |
| protected void notifyListeners(Object aCas, boolean isCasObject, |
| EntityProcessStatus aEntityProcStatus) { |
| if (aCas instanceof Object[]) { |
| for (int i = 0; i < ((Object[]) aCas).length; i++) { |
| doNotifyListeners(((Object[]) aCas)[i], isCasObject, aEntityProcStatus); |
| } |
| } else { |
| doNotifyListeners(aCas, isCasObject, aEntityProcStatus); |
| } |
| } |
| |
| /** |
| * Notifies all configured listeners. Makes sure that appropriate type of Cas is sent to the |
| * listener. Convertions take place to ensure compatibility. |
| * |
| * @param aCas - |
| * Cas to pass to listener |
| * @param isCasObject - |
| * true is Cas is of type CAS |
| * @param aEntityProcStatus - |
| * status object containing exceptions and trace info |
| */ |
| protected void doNotifyListeners(Object aCas, boolean isCasObject, |
| EntityProcessStatus aEntityProcStatus) { |
| // Notify Listener that the entity has been processed |
| Object casObjectCopy = aCas; |
| // Notify ALL listeners |
| for (int j = 0; j < statusCbL.size(); j++) { |
| BaseStatusCallbackListener statCL = (BaseStatusCallbackListener) statusCbL.get(j); |
| // Based on type of listener do appropriate conversions of Cas if necessary |
| if (statCL instanceof CasDataStatusCallbackListener) { |
| // The Cas is of type CAS, need to convert it to CasData |
| if (isCasObject == true) { |
| // Convert CAS to CasData object |
| casObjectCopy = mConverter.casContainerToCasData((CAS) casObjectCopy); |
| } |
| // Notify the listener that the Cas has been processed |
| ((CasDataStatusCallbackListener) statCL).entityProcessComplete((CasData) casObjectCopy, |
| aEntityProcStatus); |
| } else if (statCL instanceof StatusCallbackListener) { |
| boolean casFromPool = false; |
| // The cas is of type CasData, need to convert it to CAS |
| if (isCasObject == false) { |
| conversionCas = null; |
| if (casCache != null && casCache[0] != null) { |
| conversionCas = casCache[0]; |
| } else { |
| while (conversionCas == null) { |
| conversionCas = casPool.getCas(0); |
| } |
| casFromPool = true; |
| } |
| try { |
| mConverter.casDataToCasContainer((CasData) casObjectCopy, conversionCas, true); |
| } catch (CollectionException e) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.WARNING, |
| this.getClass().getName(), "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_exception_converting_CAS__WARNING", |
| new Object[] { Thread.currentThread().getName() }); |
| } |
| casObjectCopy = conversionCas; |
| } |
| // Notify the listener that the Cas has been processed |
| // ((StatusCallbackListener) statCL).entityProcessComplete((CAS) casObjectCopy, |
| // aEntityProcStatus); |
| CPMEngine.callEntityProcessCompleteWithCAS( |
| (StatusCallbackListener) statCL, (CAS) casObjectCopy, aEntityProcStatus); |
| if (conversionCas != null) { |
| if (casFromPool) { |
| conversionCasArray[0] = conversionCas; |
| cpm.releaseCASes(conversionCasArray); |
| } |
| conversionCas = null; |
| if (casCache != null && casCache[0] != null) { |
| casCache[0].reset(); |
| } |
| } |
| |
| } |
| } |
| |
| } |
| |
| private void clearCasCache() { |
| if (casCache != null) { |
| for (int index = 0; index < casCache.length; index++) { |
| if (casCache[index] != null) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_release_cas_from_cache__FINEST", |
| new Object[] { Thread.currentThread().getName() }); |
| } |
| casPool.releaseCas(casCache[index]); |
| // synchronized (casPool) { // redundant - the above releaseCas call does this |
| // casPool.notifyAll(); |
| // } |
| |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_release_cas_from_cache_done__FINEST", |
| new Object[] { Thread.currentThread().getName() }); |
| } |
| } |
| } |
| casCache = null; |
| } |
| // End New Code 05/30/05 |
| } |
| |
| /** |
| * Determines if the thread should be paused. Pausing container effectively pauses ALL Cas |
| * Processors that are managed by the container. The pause is needed when there are multiple |
| * pipelines shareing a common service. If this service dies (Socket Down), only one thread should |
| * initiate service restart. While the service is being restarted no invocations on the service |
| * should be done. Containers will be resumed on successfull service restart. |
| * |
| * @param aContainer - |
| * a container that manages the current Cas Processor. |
| * @param aProcessor - |
| * a Cas Processor to be disabled |
| * @param aThreadId - |
| * id of the current thread |
| * |
| * @throws Exception - |
| * exception |
| */ |
| private boolean pauseContainer(ProcessingContainer aContainer, Exception aException, |
| String aThreadId) { |
| if (aContainer.isRemote() && aContainer.isSingleFencedService() |
| && aException.getCause() instanceof ServiceConnectionException && aThreadId == null) { |
| return true; |
| } |
| |
| return false; |
| } |
| |
| /** |
| * |
| * @param aContainer |
| * @param aProcessor |
| * @param aProcessTr |
| * @param ex |
| * @throws Exception |
| */ |
| private void handleServiceException(ProcessingContainer aContainer, CasProcessor aProcessor, |
| ProcessTrace aProcessTr, Exception ex) throws Exception { |
| if (aProcessor instanceof NetworkCasProcessorImpl) { |
| ((NetworkCasProcessorImpl) aProcessor).collectionProcessComplete(aProcessTr); |
| } |
| // New Code 07/13/04 |
| // Release current Cas Processor before continuing with the next Cas Processor in the pipeline |
| aContainer.releaseCasProcessor(aProcessor); |
| |
| // End New Code 07/13/04 |
| |
| // New Code 2/25/2005 JC Container is paused ONLY iff CasProcessor is Remote (unmanaged) and |
| // all proxies share the same CasProcessor service AND a previous thread initiated service |
| // restart. Only one thread can initiate a restart, other threads sharing the same service |
| // will block until all connections are re-established |
| |
| if (aContainer.isRemote() && aContainer.isSingleFencedService()) { |
| if (Thread.currentThread().getName().equals(threadId)) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_service_connection_exception__FINEST", |
| new Object[] { Thread.currentThread().getName(), aContainer.getName(), |
| aProcessor.getClass().getName() }); |
| } |
| aProcessTr.startEvent(aContainer.getName(), "Process", ""); |
| // Redeploy the CasProcessor |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_redeploy_cp__FINEST", |
| new Object[] { Thread.currentThread().getName(), aContainer.getName(), |
| aProcessor.getClass().getName() }); |
| } |
| // Reconnect the CPM to CasProcessor running in fenced mode |
| cpm.redeployAnalysisEngine(aContainer); |
| |
| // New Code 02/23/05 |
| // Resume the container |
| aContainer.resume(); |
| threadId = null; |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_redeploy_cp_done__FINEST", |
| new Object[] { Thread.currentThread().getName(), aContainer.getName(), |
| aProcessor.getClass().getName() }); |
| } |
| } |
| |
| } else { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_service_connection_exception__FINEST", |
| new Object[] { Thread.currentThread().getName(), aContainer.getName(), |
| aProcessor.getClass().getName() }); |
| } |
| aProcessTr.startEvent(aContainer.getName(), "Process", ""); |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_redeploy_cp__FINEST", |
| new Object[] { Thread.currentThread().getName(), aContainer.getName(), |
| aProcessor.getClass().getName() }); |
| } |
| // Reconnect the CPM to CasProcessor running in fenced mode |
| cpm.redeployAnalysisEngine(aContainer); |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_redeploy_cp_done__FINEST", |
| new Object[] { Thread.currentThread().getName(), aContainer.getName(), |
| aProcessor.getClass().getName() }); |
| } |
| } |
| } |
| |
| /** |
| * |
| * @param aContainer |
| * @param aCasObjectList |
| * @param isLastCP |
| * @throws Exception |
| */ |
| private void handleSkipCasProcessor(ProcessingContainer aContainer, Object[] aCasObjectList, |
| boolean isLastCP) throws Exception { |
| |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_skip_CAS__FINEST", |
| new Object[] { Thread.currentThread().getName(), aContainer.getName() }); |
| } |
| if (aContainer.isPaused()) { |
| aContainer.resume(); |
| } |
| |
| aContainer.incrementAbortCount(aCasObjectList.length); |
| |
| // CAS may have already been dropped (released) due to dropCasOnException policy defined in the |
| // CPE Descriptor |
| if (!cpm.dropCasOnException()) { |
| try { |
| aContainer.logAbortedCases(aCasObjectList); |
| } catch (Exception e) { |
| throw e; |
| } finally { |
| // 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. |
| try { |
| releaseCases(casList, isLastCP, aContainer.getName()); |
| } catch (Exception ex2) { |
| if (UIMAFramework.getLogger().isLoggable(Level.SEVERE)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.SEVERE, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_exception_releasing_cas__SEVERE", |
| new Object[] { Thread.currentThread().getName(), aContainer.getName() }); |
| throw ex2; |
| } |
| } |
| } |
| } |
| |
| } |
| |
| /** |
| * Returns the size of the CAS object. Currently only CASData is supported. |
| * |
| * @param aCas - |
| * Cas to get the size for |
| * |
| * @return |
| */ |
| protected long getBytes(Object aCas) { |
| try { |
| if (aCas instanceof CasData) { |
| return DATACasUtils.getByteCount((CasData) aCas); |
| } |
| } catch (Exception e) { |
| e.printStackTrace(); |
| } |
| return 0; |
| } |
| |
| /** |
| * Conditionally, releases CASes back to the CAS pool. The release only occurs if the Cas |
| * Processor is the last in the processing chain. |
| * |
| * @param aCasList - |
| * list of CASes to release |
| * @param lastProcessor - |
| * determines if the release takes place |
| * @param aContainer - |
| * current container |
| */ |
| private void releaseCases(Object aCasList, boolean lastProcessor, String aName) // ProcessingContainer |
| // aContainer) |
| { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_releasing_cases__FINEST", |
| new Object[] { Thread.currentThread().getName(), aName, String.valueOf(relaseCAS), |
| String.valueOf(lastProcessor) }); |
| } |
| if (aCasList == null) { |
| return; |
| } |
| |
| if (relaseCAS && lastProcessor) { |
| if (aCasList instanceof CAS[]) { |
| // New Code 06/01/05 JC |
| if (casCache != null) { |
| // First release any CASes from the Cas Cache |
| cpm.releaseCASes(casCache); |
| casCache = null; |
| } |
| // End New Code 06/01/05 JC |
| cpm.releaseCASes((CAS[]) aCasList); |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_done_releasing_cases__FINEST", |
| new Object[] { Thread.currentThread().getName(), aName }); |
| } |
| } else { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_casobject_class__FINEST", |
| new Object[] { Thread.currentThread().getName(), aName, |
| aCasList.getClass().getName() }); |
| } |
| } |
| } |
| |
| } |
| |
| /** |
| * Stops all Cas Processors that are part of this PU. |
| * |
| * @param kill - |
| * true if CPE has been stopped before finishing processing during external stop |
| * |
| */ |
| public void stopCasProcessors(boolean kill) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_stop_containers__FINEST", |
| new Object[] { Thread.currentThread().getName() }); |
| } |
| // while (consumeQueue()); |
| // |
| // Object[] eofToken = new Object[1]; |
| // // only need to one member in the array |
| // eofToken[0] = new EOFToken(); |
| // |
| // workQueue.enqueue(eofToken); |
| |
| // Stop all running CASProcessors |
| for (int i = 0; processContainers != null && i < processContainers.size(); i++) { |
| ProcessingContainer container = (ProcessingContainer) processContainers.get(i); |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_show_container_time__FINEST", |
| new Object[] { Thread.currentThread().getName(), container.getName(), |
| String.valueOf(container.getTotalTime()) }); |
| } |
| synchronized (container) { |
| // Change the status of this container to KILLED if the CPM has been stopped |
| // before completing the collection and current status of CasProcessor is |
| // either READY or RUNNING |
| if (kill || (!cpm.isRunning() && isProcessorReady(container.getStatus()))) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), |
| "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_kill_cp__FINEST", |
| new Object[] { Thread.currentThread().getName(), container.getName() }); |
| } |
| container.setStatus(Constants.CAS_PROCESSOR_KILLED); |
| } else { |
| // If the CasProcessor has not been disabled during processing change its |
| // status to COMPLETED. |
| if (container.getStatus() != Constants.CAS_PROCESSOR_DISABLED) { |
| container.setStatus(Constants.CAS_PROCESSOR_COMPLETED); |
| } |
| } |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_container_status__FINEST", |
| new Object[] { Thread.currentThread().getName(), container.getName(), |
| String.valueOf(container.getStatus()) }); |
| } |
| ProcessTrace pTrTemp = new ProcessTrace_impl(cpm.getPerformanceTuningSettings()); |
| pTrTemp.startEvent(container.getName(), "End of Batch", ""); |
| try { |
| CasProcessorDeployer deployer = container.getDeployer(); |
| |
| if (deployer != null) { |
| if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(this.getClass()).logrb( |
| Level.FINEST, |
| this.getClass().getName(), |
| "process", |
| CPMUtils.CPM_LOG_RESOURCE_BUNDLE, |
| "UIMA_CPM_undeploy_cp_instances__FINEST", |
| new Object[] { Thread.currentThread().getName(), container.getName(), |
| deployer.getClass().getName() }); |
| } |
| deployer.undeploy(); |
| } |
| container.destroy(); |
| } catch (Exception e) { |
| e.printStackTrace(); |
| } finally { |
| pTrTemp.endEvent(container.getName(), "End of Batch", ""); |
| if (processingUnitProcessTrace != null) { |
| this.processingUnitProcessTrace.aggregate(pTrTemp); |
| } |
| } |
| } |
| } |
| } |
| |
| } |
