You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@uima.apache.org by al...@apache.org on 2006/11/22 17:44:00 UTC
svn commit: r478226 [2/32] - in /incubator/uima/uimaj/trunk/uimaj-cpe/src:
main/java/org/apache/uima/cas_data/impl/vinci/
main/java/org/apache/uima/collection/impl/
main/java/org/apache/uima/collection/impl/base_cpm/container/
main/java/org/apache/uima...
Modified: incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/BaseCPMImpl.java
URL: http://svn.apache.org/viewvc/incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/BaseCPMImpl.java?view=diff&rev=478226&r1=478225&r2=478226
==============================================================================
--- incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/BaseCPMImpl.java (original)
+++ incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/BaseCPMImpl.java Wed Nov 22 08:43:54 2006
@@ -63,1178 +63,1113 @@
import org.apache.uima.util.impl.ProcessTrace_impl;
/**
- * Main thread that launches CPE and manages it. An application interacts with the running
- * CPE via this object. Through an API, an application may start, pause, resume, and stop a
- * CPE.
+ * Main thread that launches CPE and manages it. An application interacts with the running CPE via
+ * this object. Through an API, an application may start, pause, resume, and stop a CPE.
+ *
*
- *
*/
-public class BaseCPMImpl implements BaseCPM, Runnable
-{
- private boolean defaultProcessTrace;
- private CPMEngine cpEngine = null;
- private ProcessTrace procTr = null;
- BaseCollectionReader collectionReader = null;
- private Checkpoint checkpoint = null;
- private CheckpointData checkpointData = null;
- private long num2Process = -1; // Default ALL
- private boolean killed = false;
- private boolean completed = false;
- private CPEFactory cpeFactory = null;
- private boolean useJediiReport = true;
- private Map mEventTypeMap;
- public CPMThreadGroup cpmThreadGroup = null;
- /**
- * Instantiates and initializes CPE Factory with a given CPE Descriptor and defaults.
- *
- * @param aDescriptor - parsed CPE descriptor
- * @throws Exception
- */
- public BaseCPMImpl(CpeDescription aDescriptor) throws Exception
- {
- this(aDescriptor, null, true, UIMAFramework.getDefaultPerformanceTuningProperties());
- cpmThreadGroup = new CPMThreadGroup("CPM Thread Group");
- }
- /**
- * Instantiates and initializes CPE Factory responsible for creating individual components
- * that are part of the processing pipeline.
- *
- * @param aDescriptor - parsed CPE descriptor
- * @param aResourceManager - ResourceManager instance to be used by the CPE
- * @param aDefaultProcessTrace - ProcessTrace instance to capture events and stats
- * @throws Exception
- */
- public BaseCPMImpl(CpeDescription aDescriptor, ResourceManager aResourceManager, boolean aDefaultProcessTrace, Properties aProps) throws Exception
- {
- cpeFactory = new CPEFactory(aDescriptor, aResourceManager);
- defaultProcessTrace = aDefaultProcessTrace;
- cpmThreadGroup = new CPMThreadGroup("CPM Thread Group");
- init(false, aProps);
- }
- /**
- * Parses CPE descriptor
- *
- * @param mode - indicates if the CPM should use a static descriptor or one provided
- * @param aDescriptor - provided descriptor path
- * @param aResourceManager ResourceManager to be used by CPM
- *
- * @throws Exception
- */
- public BaseCPMImpl(Boolean mode, String aDescriptor, ResourceManager aResourceManager) throws Exception
- {
- cpmThreadGroup = new CPMThreadGroup("CPM Thread Group");
- cpeFactory = new CPEFactory(aResourceManager);
- if (mode == null)
- {
- defaultProcessTrace = true;
- cpeFactory.parse();
- }
- else
- {
- defaultProcessTrace = mode.booleanValue();
- cpeFactory.parse(aDescriptor);
- }
- init(mode == null,UIMAFramework.getDefaultPerformanceTuningProperties() );
- }
- /**
- * Plugs in custom perfomance tunning parameters
- *
- * @param aPerformanceTuningSettings
- */
- public void setPerformanceTuningSettings(Properties aPerformanceTuningSettings)
- {
- cpEngine.setPerformanceTuningSettings(aPerformanceTuningSettings);
- }
- /**
- * Plugs in a given {@link ProcessControllerAdapter}. The CPM uses this adapter to
- * request Cas Processor restarts and shutdown.
- *
- * @param aPca - instance of the ProcessControllerAdapter
- */
- public void setProcessControllerAdapter(ProcessControllerAdapter aPca)
- {
- cpEngine.setProcessControllerAdapter(aPca);
- }
-
- /**
- * Sets Jedii-style reporting resources and sets the global flag to indicate what report-style to use at
- * the end of processing. Jedii-style reporting shows a summary for this run. The CPM default report
- * shows more detail information.
- *
- * @param aUseJediiReport
- */
- public void setJediiReport(boolean aUseJediiReport )
- {
- mEventTypeMap = new HashMap();
- mEventTypeMap.put(ProcessTraceEvent.ANALYSIS_ENGINE, "TAE");
- mEventTypeMap.put(ProcessTraceEvent.ANALYSIS, "Annotator");
- mEventTypeMap.put("CAS_PROCESSOR", "CAS Consumer");
- useJediiReport = aUseJediiReport;
- }
- /**
- * Instantiates and initializes a CPE.
- *
- * @param aDummyCasProcessor -
- *
- * @throws Exception
- */
- public void init(boolean aDummyCasProcessor, Properties aProps) throws Exception
- {
- String uimaTimerClass = cpeFactory.getCPEConfig().getTimerImpl();
- try
- {
- new TimerFactory(uimaTimerClass);
- }
- catch( Exception e)
- {
- //e.printStackTrace();
- if ( UIMAFramework.getLogger().isLoggable(Level.FINEST) )
- {
- UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), "process",
- CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_use_default_timer__FINEST",
- new Object[] {Thread.currentThread().getName()});
-
- }
- }
- UimaTimer uimaTimer = TimerFactory.getTimer();
- if ( UIMAFramework.getLogger().isLoggable(Level.FINEST) )
- {
- UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(), "process",
- CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_use_custom_timer__FINEST",
- new Object[] {Thread.currentThread().getName(),uimaTimer.getClass().getName()});
- }
-
- procTr = new ProcessTrace_impl(uimaTimer, aProps);
- String checkpointFileName = null;
- if ( cpeFactory.getCPEConfig().getCheckpoint() != null && cpeFactory.getCPEConfig().getCheckpoint().getFilePath() != null)
- {
- // Retrieve from CPM configuration a name of the checkpoint file where the
- // CPM's runtime stats be deposited.
- checkpointFileName = cpeFactory.getCPEConfig().getCheckpoint().getFilePath();
- }
- if (checkpointFileName != null && checkpointFileName.trim().length() > 0)
- {
- File checkpointFile = new File(checkpointFileName);
- checkpoint = new Checkpoint(this, checkpointFileName, cpeFactory.getCPEConfig().getCheckpoint().getFrequency());
- // Check if the checkpoint file already exists. If it does, the CPM did not complete
- // successfully during the previous run and CPM will start in recovery mode, restoring all
- // totals and status's from the recovered checkpoint. The processing pipeline state will
- // restored to the state as of before foreced shutdown. All CasProcessors that were disabled
- // during that run will remain disabled. The CollectionReader will be advanced to the
- // entity last processed by the previous CPM.
- if (checkpointFile.exists())
- {
- try
- {
- Object restoredObject = checkpoint.restoreFromCheckpoint();
- if ( restoredObject != null && restoredObject instanceof CheckpointData )
- {
- checkpointData = (CheckpointData) restoredObject;
- }
- }
- catch (Exception e)
- {
- e.printStackTrace();
- }
- }
- }
- // Instantiate class responsible for processing
- cpEngine = new CPMEngine(cpmThreadGroup,cpeFactory, procTr, checkpointData);
- if (!aDummyCasProcessor)
- {
- int concurrentThreadCount = cpeFactory.getCpeDescriptor().getCpeCasProcessors().getConcurrentPUCount();
- for (int threadCount = 0; threadCount < concurrentThreadCount; threadCount++)
- {
- CasProcessor[] casProcessors = cpeFactory.getCasProcessors();
- for (int i = 0; i < casProcessors.length; i++)
- {
- if ( UIMAFramework.getLogger().isLoggable(Level.CONFIG) )
- {
- UIMAFramework.getLogger(this.getClass()).logrb(Level.CONFIG, this.getClass().getName(), "process",
- CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_add_cp__CONFIG",
- new Object[] {Thread.currentThread().getName(),casProcessors[i].getProcessingResourceMetaData().getName()});
- }
- addCasProcessor(casProcessors[i]);
- }
- }
- }
- int casPoolSize = 0;
- try
- {
- casPoolSize = cpeFactory.getCpeDescriptor().getCpeCasProcessors().getCasPoolSize();
- casPoolSize = (casPoolSize == -1) ? 0 : casPoolSize;
- cpEngine.setPoolSize(casPoolSize);
- }
- catch (NumberFormatException e)
- {
- // PoolSize is currently an optional parameter. If it does not exist in CPE descriptor a default
- // value will be derived from defined queue sizes
- }
-
- try
- {
- int iqSize = 0;
- if ( casPoolSize == 0 )
- {
- iqSize = cpeFactory.getCpeDescriptor().getCpeCasProcessors().getInputQueueSize();
- }
- cpEngine.setInputQueueSize(casPoolSize == 0 ? iqSize : casPoolSize);
- }
- catch (NumberFormatException e)
- {
- throw new Exception(CpmLocalizedMessage.getLocalizedMessage(CPMUtils.CPM_LOG_RESOURCE_BUNDLE,"UIMA_CPM_EXP_queue_size_not_defined__WARNING",new Object[] {Thread.currentThread().getName(),"inputQueueSize"}) );
- }
- try
- {
- int oqSize = 0;
- if ( casPoolSize == 0)
- {
- oqSize = cpeFactory.getCpeDescriptor().getCpeCasProcessors().getOutputQueueSize();
- }
- cpEngine.setOutputQueueSize(casPoolSize == 0 ? oqSize : casPoolSize+2);
- }
- catch (NumberFormatException e)
- {
- throw new Exception(CpmLocalizedMessage.getLocalizedMessage(CPMUtils.CPM_LOG_RESOURCE_BUNDLE,"UIMA_CPM_EXP_queue_size_not_defined__WARNING",new Object[] {Thread.currentThread().getName(),"outputQueueSize"}) );
- }
- try
- {
- int threadCount = cpeFactory.getCpeDescriptor().getCpeCasProcessors().getConcurrentPUCount();
- cpEngine.setConcurrentThreadSize(threadCount);
- }
- catch (NumberFormatException e)
- {
- throw new Exception(CpmLocalizedMessage.getLocalizedMessage(CPMUtils.CPM_LOG_RESOURCE_BUNDLE,"UIMA_CPM_EXP_invalid_component_reference__WARNING",new Object[] {Thread.currentThread().getName(),"casProcessors","processingUnitThreadCount"}) );
- }
- }
- /**
- * Returns {@link CPEConfig} object holding current CPE configuration
- *
- * @return CPEConfig instance
- *
- * @throws Exception
- */
- public CpeConfiguration getCPEConfig() throws Exception
- {
- return cpeFactory.getCPEConfig();
- }
- /*
- * Returns All CasProcessors currently in the processing pipeline
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#getCasProcessors()
- */
- public CasProcessor[] getCasProcessors()
- {
- CasProcessor[] casProcs = cpEngine.getCasProcessors();
- return casProcs == null ? new CasProcessor[0] : casProcs;
- }
- /*
- * Adds given CasProcessor to the processing pipeline. A new CasProcessor is appended to the current list.
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#addCasProcessor(org.apache.uima.collection.base_cpm.CasProcessor)
- */
- public void addCasProcessor(CasProcessor aCasProcessor) throws ResourceConfigurationException
- {
- cpEngine.addCasProcessor(aCasProcessor);
- }
-
- /*
- * Adds given CasProcessor to the processing pipeline. A new CasProcessor is inserted into a given spot in the current list.
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#addCasProcessor(org.apache.uima.collection.base_cpm.CasProcessor, int)
- */
- public void addCasProcessor(CasProcessor aCasProcessor, int aIndex) throws ResourceConfigurationException
- {
- cpEngine.addCasProcessor(aCasProcessor, aIndex);
- }
-
- /*
- * Removes given CasProcessor from the processing pipeline.
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#removeCasProcessor(org.apache.uima.collection.base_cpm.CasProcessor)
- */
- public void removeCasProcessor(CasProcessor aCasProcessor)
- {
- cpEngine.removeCasProcessor(0);
- }
- /*
- * Disables given CasProcessor in the existing processing pipeline.
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#disableCasProcessor(org.apache.uima.collection.base_cpm.CasProcessor)
- */
- public void disableCasProcessor(String aCasProcessorName)
- {
-
- cpEngine.disableCasProcessor(aCasProcessorName);
- }
- /*
- * Disables given CasProcessor in the existing processing pipeline.
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#disableCasProcessor(org.apache.uima.collection.base_cpm.CasProcessor)
- */
- public void enableCasProcessor(String aCasProcessorName)
- {
-
- cpEngine.enableCasProcessor(aCasProcessorName);
- }
-
- /*
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#isSerialProcessingRequired()
- */
- public boolean isSerialProcessingRequired()
- {
- return false;
- }
-
- /* (non-Javadoc)
- * @see org.apache.uima.collection.base_cpm.BaseCPM#setSerialProcessingRequired(boolean)
- */
- public void setSerialProcessingRequired(boolean aRequired)
- {
- }
-
- /*
- * Returns true if this cpEngine pauses on exception. False otherwise.
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#isPauseOnException()
- */
- public boolean isPauseOnException()
- {
- return cpEngine.isPauseOnException();
- }
-
- /*
- * Defines if cpEngine should pause on exception
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#setPauseOnException(boolean)
- */
- public void setPauseOnException(boolean aPause)
- {
- cpEngine.setPauseOnException(aPause);
- }
-
- /*
- * Adds Event Listener. Important events like, end of entity processing, exceptions, etc will
- * be sent to the registered listeners.
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#addStatusCallbackListener(org.apache.uima.collection.base_cpm.BaseStatusCallbackListener)
- */
- public void addStatusCallbackListener(BaseStatusCallbackListener aListener)
- {
- cpEngine.addStatusCallbackListener(aListener);
- }
-
- /*
- * Remoces named listener from the listener list.
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#removeStatusCallbackListener(org.apache.uima.collection.base_cpm.BaseStatusCallbackListener)
- */
- public void removeStatusCallbackListener(BaseStatusCallbackListener aListener)
- {
- cpEngine.removeStatusCallbackListener(aListener);
- }
-
- /*
- * Starting point for the CPE. Before starting processing, the CPE must deploy all CasProcessors.
- * Once all are deployed the processing begins in Worker Thread called CPEngine. This thread
- * blocks until the CPEngine thread completes.
- * CPMWorker Thread finishes.
- *
- * @see java.lang.Runnable#run()
- */
- public void run()
- {
- long start, end;
- // name this thread
- Thread.currentThread().setName("BaseCPMImpl-Thread");
-
- start = System.currentTimeMillis();
- if ( !useJediiReport )
- {
- procTr.startEvent("CPM", "CPM PROCESSING TIME", "");
- }
-
- // Specify how docs to process
- cpEngine.setNumToProcess(num2Process);
- try
- {
- // Deploy all CAS processors
- cpEngine.deployCasProcessors();
- cpEngine.setCollectionReader(collectionReader);
-
- // Start the checkpoint thread
- if (checkpoint != null)
- {
- new Thread(checkpoint).start();
- }
- cpEngine.start();
- // Joing the CPMWorker Thread and wait until it finishes
- cpEngine.join();
-
- completed = true;
- // If the entire collection has been processed there is no need for a checkpoint.
- // Delete it, there is no need to recover anything. Otherwise, the CPM has been killed
- // and may need to be recovered. Checkpoint file contains the status of the CPM, including
- // last document processed, status of all CasProcessors along with all counts and totals.
- if (!killed && checkpoint != null)
- {
- checkpoint.stop();
- checkpoint.delete();
- checkpoint = null;
- }
- // Terminate all threads and running services
- cpEngine.stopCasProcessors(false);
- // Notify Listeners that the processing pipeline has finished
- if ( !useJediiReport )
- {
- procTr.endEvent("CPM", "CPM PROCESSING TIME", "success");
- }
- end = 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_show_total_time_in_cpm__FINEST",
- new Object[] {Thread.currentThread().getName(),String.valueOf(end - start)});
- }
- }
- catch (AbortCPMException e)
- {
- if ( !useJediiReport )
- {
- procTr.endEvent("CPM", "CPM PROCESSING TIME", "failed");
- }
- // Terminate all threads and running services
- try
- {
- cpEngine.stopCasProcessors(true);
- }
- catch( Exception ex )
- {
- UIMAFramework.getLogger(this.getClass()).log(Level.SEVERE, "" + ex);
- }
- killed = true;
- }
- catch (Exception e)
- {
- if ( !useJediiReport )
- {
- procTr.endEvent("CPM", "CPM PROCESSING TIME", "failed");
- }
- UIMAFramework.getLogger(this.getClass()).log(Level.SEVERE, "" + e);
- killed = true;
- ArrayList statusCbL = cpEngine.getCallbackListeners();
- EntityProcessStatusImpl enProcSt = new EntityProcessStatusImpl(procTr, true);
- // e is the actual exception.
- enProcSt.addEventStatus("CPM", "Failed", e);
-
-
- // Notify all listeners that the CPM has finished processing
- for (int j = 0; j < statusCbL.size(); j++)
- {
- BaseStatusCallbackListener st = (BaseStatusCallbackListener) statusCbL.get(j);
- if ( st instanceof StatusCallbackListener )
- {
- ((StatusCallbackListener) st).entityProcessComplete(null, enProcSt);
- }
- }
- }
-
- if ( cpEngine.isKilled() )
- {
- killed = 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_cpm_stopped__FINEST",
- new Object[] {Thread.currentThread().getName(),String.valueOf(killed)});
- }
- ArrayList statusCbL = cpEngine.getCallbackListeners();
- // Notify all listeners that the CPM has finished processing
- for (int j = 0; j < statusCbL.size(); j++)
- {
- BaseStatusCallbackListener st = (BaseStatusCallbackListener) statusCbL.get(j);
- if (!killed)
- {
- st.collectionProcessComplete();
- }
- else
- {
- st.aborted();
- }
- }
- }
-
- /**
- * Called to cleanup CPE on shutdown
- *
- */
- public void finalizeIt()
- {
- // Do cleanup before terminating self
- cpEngine.cleanup();
-
- }
- /**
- * This method is called by an applications to begin CPM processing with a given Collection.
- * It just creates a new thread and starts it.
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#process(org.apache.uima.collection.base_cpm.BaseCollectionReader)
- * @deprecated
- */
- public void process(BaseCollectionReader aCollectionReader) throws ResourceInitializationException
- {
- // Retrieve number of entities to process from the CPM configuration
- try
- {
- num2Process = cpeFactory.getCPEConfig().getNumToProcess();
- }
- catch (InstantiationException e)
- {
- throw new ResourceInitializationException(e);
- }
- collectionReader = aCollectionReader;
- if (cpeFactory.isDefault())
- {
- cpeFactory.addCollectionReader(collectionReader);
- }
- cpmThreadGroup.setProcessTrace(procTr);
- cpmThreadGroup.setListeners(cpEngine.getCallbackListeners());
- new Thread(this).start();
- }
- /*
- * This method is called by an application to begin processing given Collection.
- * It creates a new thread, adds it to a ThreadGroup and starts it.
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#process(org.apache.uima.collection.base_cpm.BaseCollectionReader)
- */
- public void process() throws ResourceInitializationException
- {
- // Retrieve number of entities to process from the CPM configuration
- try
- {
- num2Process = cpeFactory.getCPEConfig().getNumToProcess();
- if (collectionReader == null)
- {
- collectionReader = cpeFactory.getCollectionReader();
- }
- }
- catch (InstantiationException e)
- {
- throw new ResourceInitializationException(e);
- }
- catch (Exception e)
- {
- throw new ResourceInitializationException(e);
- }
- if (cpeFactory.isDefault())
- {
- cpeFactory.addCollectionReader(collectionReader);
- }
- cpmThreadGroup.setProcessTrace(procTr);
- cpmThreadGroup.setListeners(cpEngine.getCallbackListeners());
-
- new Thread(this).start();
- }
-
- /**
- *
- * This method is called by an applications to begin CPM processing with a given Collection.
- * It just creates a new thread and starts it.
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#process(org.apache.uima.collection.base_cpm.BaseCollectionReader, int)
- * @deprecated
- */
- public void process(BaseCollectionReader aCollectionReader, int aBatchSize) throws ResourceInitializationException
- {
- // Let the application define the size of Collection.
- num2Process = aBatchSize;
- collectionReader = aCollectionReader;
- if (cpeFactory.isDefault())
- {
- cpeFactory.addCollectionReader(collectionReader);
- }
- cpmThreadGroup.setProcessTrace(procTr);
- cpmThreadGroup.setListeners(cpEngine.getCallbackListeners());
-
- new Thread(cpmThreadGroup, this).start();
- }
- /**
- * Sets the Collection Reader for this CPE.
- *
- * @param aCollectionReader the collection reader
- */
- public void setCollectionReader(BaseCollectionReader aCollectionReader)
- {
- collectionReader = aCollectionReader;
- if (cpeFactory.isDefault())
- {
- cpeFactory.addCollectionReader(collectionReader);
- }
- }
-
- /**
- * Returns a Collection Reader for this CPE.
- *
- * @return the collection reader
- */
- public BaseCollectionReader getCollectionReader()
- {
- try
- {
- if (this.collectionReader == null) {
- this.collectionReader = this.cpeFactory.getCollectionReader();
- }
- return this.collectionReader;
- }
- catch (ResourceConfigurationException e)
- {
- throw new UIMARuntimeException(e);
- }
- }
-
- /*
- * Returns current state of the CPM.
- *
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#isProcessing()
- */
- public boolean isProcessing()
- {
- return cpEngine.isRunning();
- }
-
- /*
- * Pauses the CPM
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#pause()
- */
- public void pause()
- {
- cpEngine.pauseIt();
- if (checkpoint != null)
- {
- checkpoint.doCheckpoint();
- checkpoint.pause();
- }
- }
-
- /*
- * Returns true if the CPM is in pause state
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#isPaused()
- */
- public boolean isPaused()
- {
- return cpEngine.isPaused();
- }
-
- /*
- * Resumes the CPM
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#resume(boolean)
- */
- public void resume(boolean aRetryFailed)
- {
- resume();
- }
-
- /*
- * Resumes the CPM
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#resume()
- */
- public void resume()
- {
- cpEngine.resumeIt();
- if (checkpoint != null)
- {
- checkpoint.resume();
- }
- }
- /**
- * Kills the CPM hard. CASes in transit are not processed.
- *
- *
- */
- public void kill()
- {
- if ( UIMAFramework.getLogger().isLoggable(Level.WARNING) )
- {
- UIMAFramework.getLogger(this.getClass()).logrb(Level.WARNING, this.getClass().getName(), "process",
- CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_killing_cpm__WARNING",
- new Object[] {Thread.currentThread().getName()});
- }
- killed = true;
- // Stop processing pipeline. The CPMWorker will finish processing of the current
- // entity through the processing pipeline and than will terminate processing.
- cpEngine.killIt();
-
- // If valid checkpoint reference and not already deleted do the next checkpoint and stop
- // the checkpoint thread. Checkpoint file may be deleted if the CPM successfully completes
- // its run. See BaseCPM#run().
- if (checkpoint != null && !completed)
- {
- checkpoint.doCheckpoint();
- checkpoint.stop();
- }
-
- }
- /*
- * Stops the CPM and all of its processing components. The CPM finishes processing of all CASes that are
- * in its queues.
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#stop()
- */
- public void stop()
- {
- if ( UIMAFramework.getLogger().isLoggable(Level.WARNING) )
- {
- UIMAFramework.getLogger(this.getClass()).logrb(Level.WARNING, this.getClass().getName(), "process",
- CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_stop_cpm__WARNING",
- new Object[] {Thread.currentThread().getName()});
- }
- killed = true;
- // Stop processing pipeline. The CPMWorker will finish processing of the current
- // entity through the processing pipeline and than will terminate processing.
- cpEngine.stopIt();
-
- // If valid checkpoint reference and not already deleted do the next checkpoint and stop
- // the checkpoint thread. Checkpoint file may be deleted if the CPM successfully completes
- // its run. See BaseCPM#run().
- if (checkpoint != null && !completed)
- {
- checkpoint.doCheckpoint();
- checkpoint.stop();
- }
- }
- /*
- * Stops/kills the CPM and all of its processing components.
- * @deprecated
- *
- * @see org.apache.uima.collection.base_cpm.BaseCPM#stop()
- */
- public void asynchStop()
- {
- if ( UIMAFramework.getLogger().isLoggable(Level.INFO) )
- {
- UIMAFramework.getLogger(this.getClass()).logrb(Level.WARNING, this.getClass().getName(), "process",
- CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_asynch_stop_cpm__WARNING",
- new Object[] {Thread.currentThread().getName()});
- }
- killed = true;
- // Stop processing pipeline. The CPMWorker will finish processing of the current
- // entity through the processing pipeline and than will terminate processing.
- cpEngine.asynchStop();
-
- // If valid checkpoint reference and not already deleted do the next checkpoint and stop
- // the checkpoint thread. Checkpoint file may be deleted if the CPM successfully completes
- // its run. See BaseCPM#run().
- if (checkpoint != null && !completed)
- {
- checkpoint.doCheckpoint();
- checkpoint.stop();
- }
- }
- /*
- * Returns a String describing a given CASProcessor state.
- *
- * @param aStatus - status of the CasProcessor
- * @return - String corresponding to a given state of the CasProcessor
- */
- private String decodeStatus(int aStatus)
- {
- try
- {
- switch (aStatus)
- {
- case Constants.CAS_PROCESSOR_COMPLETED :
- return Constants.COMPLETED;
- case Constants.CAS_PROCESSOR_DISABLED :
- return Constants.DISABLED;
- case Constants.CAS_PROCESSOR_READY :
- return Constants.READY;
- case Constants.CAS_PROCESSOR_RUNNING :
- return Constants.RUNNING;
- case Constants.CAS_PROCESSOR_KILLED :
- return Constants.KILLED;
- }
- }
- catch (NumberFormatException e)
- {
- e.printStackTrace();
- }
- return Constants.UNKNOWN;
- }
- /*
- * Copies events of a given type found in the list to a provided ProcessTrace instance
- *
- * @param - aEvType, event type to copy from the list
- * @param - List, list of events
- * @param ProcessTrace, where to copy events of a given type
- *
- */
- private void copyComponentEvents(String aEvType, List aList, ProcessTrace aPTr) throws IOException
- {
- for (int i = 0; i < aList.size(); i++)
- {
- ProcessTraceEvent prEvent = (ProcessTraceEvent) aList.get(i);
- if (aEvType != null && aEvType.equals(prEvent.getType()))
- {
- aPTr.addEvent(prEvent);
- }
- }
-
- }
- /**
- * Helper method to display stats and totals
- *
- * @param aProcessTrace - trace containing stats
- *
- * @param aNumDocsProcessed - number of entities processed so far
- */
- public void displayStats(ProcessTrace aProcessTrace, int aNumDocsProcessed)
- {
- if ( UIMAFramework.getLogger().isLoggable(Level.FINEST) )
- {
- UIMAFramework.getLogger(this.getClass()).log(Level.FINEST, "Documents Processed: " + aNumDocsProcessed);
- }
- //count total time
- int totalTime = 0;
- Iterator it = aProcessTrace.getEvents().iterator();
- while (it.hasNext())
- {
- ProcessTraceEvent event = (ProcessTraceEvent)it.next();
- // Dont add total time the CPM ran for. Just add all of the times of all components to
- // get the time.
- if ( "CPM".equals(event.getComponentName()) )
- {
- if ( UIMAFramework.getLogger().isLoggable(Level.FINEST) )
- {
- UIMAFramework.getLogger(this.getClass()).log(Level.FINEST, "Current Component::"+event.getComponentName()+" Time::"+event.getDuration());
- }
- continue;
- }
- if ( UIMAFramework.getLogger().isLoggable(Level.FINEST) )
- {
- UIMAFramework.getLogger(this.getClass()).log(Level.FINEST, "Current Component::"+event.getComponentName());
- }
- totalTime += event.getDuration();
- }
- float totalTimeSeconds = (float)totalTime / 1000;
- if ( UIMAFramework.getLogger().isLoggable(Level.FINEST) )
- {
- UIMAFramework.getLogger(this.getClass()).log(Level.FINEST, "Total Time: " + totalTimeSeconds + " seconds");
- }
-
- //create root tree node
- if ( UIMAFramework.getLogger().isLoggable(Level.FINEST) )
- {
- UIMAFramework.getLogger(this.getClass()).log(Level.FINEST, "100% (" + totalTime + "ms) - Collection Processing Engine");
- }
- //build tree
- it = aProcessTrace.getEvents().iterator();
- while (it.hasNext())
- {
- ProcessTraceEvent event = (ProcessTraceEvent)it.next();
- buildEventTree(event, totalTime);
- }
- }
- /**
- * Helper method to help build the CPM report
- *
- * @param aEvent
- * @param aTotalTime
- */
- public void buildEventTree(ProcessTraceEvent aEvent, int aTotalTime)
- {
- // Skip reporting the CPM time.This time has already been acquired by summing up
- // times from all individual components
- if ( "CPM".equals(aEvent.getComponentName()) )
- {
- return;
- }
-
- int duration = aEvent.getDuration();
- float pct = (float)((duration * 100 * 10) / aTotalTime) / 10;
-
- String type = (String)mEventTypeMap.get(aEvent.getType());
- if (type == null)
- {
- type = aEvent.getType();
- }
-
- if (System.getProperty("DEBUG") != null)
- UIMAFramework.getLogger(this.getClass()).log(Level.FINEST, "" + pct + "% (" + duration + "ms) - " + aEvent.getComponentName() + " (" + type + ")");
- Iterator it = aEvent.getSubEvents().iterator();
- while (it.hasNext())
- {
- ProcessTraceEvent event = (ProcessTraceEvent)it.next();
- buildEventTree(event, aTotalTime);
- }
- }
-
-
- /**
- * Returns PerformanceReport for the CPM. This report contains a snapshot of the CPM state.
- *
- */
- public ProcessTrace getPerformanceReport()
- {
- Map perfReport = cpEngine.getStats();
- Progress[] colReaderProgress = (Progress[]) perfReport.get("COLLECTION_READER_PROGRESS");
-
- ProcessTrace processTrace = new ProcessTrace_impl(cpEngine.getPerformanceTuningSettings());
- if ( UIMAFramework.getLogger().isLoggable(Level.FINEST) )
- {
- UIMAFramework.getLogger(this.getClass()).log(Level.FINEST, "-------------------------------------------");
- }
- if ( useJediiReport )
- {
- try
- {
- synchronized (procTr)
- {
- List eventList = procTr.getEvents();
-
- for (int j = 0; eventList != null && j < eventList.size(); j++)
- {
- ProcessTraceEvent prEvent = (ProcessTraceEvent) eventList.get(j);
- processTrace.addEvent(prEvent);
- }
- }
- return processTrace;
- }
- catch( Exception e)
- {
- e.printStackTrace();
- }
- }
- if (defaultProcessTrace)
- {
- createDefaultProcessTrace(getCasProcessors(), procTr, processTrace);
-
- return processTrace;
- }
- try
- {
-
- // To facilitate recovery from CPM untimely shutdown ( due to external STOP), it must
- // have access to last entity id. CollectionReader will use this marker to synch itself
- // up to the last known entity before the CPM shut itself down. More complicated
- // recovery mechanism may be supported as long as CAS METADATA contains information
- // about the last known point. For example, in case WF Large Store the cas must
- // hold the entire EDATA frame, containing restart information. Its up to the
- // CollectionReader to know what part of the CAS Metadata should be used for recovery.
- // The "last cas" CasMetaData is added after each successfull read from the CollectionReader
- // by the cpEngine in its run() processing loop.
- CasMetaData casMetaData = (CasMetaData) perfReport.get("CPM_LAST_CAS_METADATA");
- if (casMetaData != null)
- {
- NameValuePair[] nvp = casMetaData.getCasMetaData();
- StringBuffer sb = new StringBuffer();
- for (int i = 0; i < nvp.length && nvp[i] != null; i++)
- {
- if (i != 0)
- {
- // Add separator between name-value pairs. StringTokenizer will parse this string
- // and will use the separator to extract nvp.
- sb.append(",");
- }
- sb.append(nvp[i].getName() + "=" + (String) nvp[i].getValue());
- }
- processTrace.addEvent("CPM", "CPM_LAST_CAS_METADATA", sb.toString(), 0, null);
- }
- List eList = null;
- synchronized (procTr)
- {
- eList = procTr.getEventsByComponentName("CPM", true);
- }
- if ( !useJediiReport )
- {
- copyComponentEvents("CPM PROCESSING TIME", eList, processTrace);
- }
- eList.clear();
- if (colReaderProgress != null)
- {
- Long totalCollectionReaderTime = (Long)perfReport.get("COLLECTION_READER_TIME");
- String readerName = collectionReader.getProcessingResourceMetaData().getName();
- if (totalCollectionReaderTime != null )
- {
- processTrace.addEvent(readerName, "COLLECTION_READER_TIME",String.valueOf(totalCollectionReaderTime), 0, null);
- }
- for (int i = 0; i < colReaderProgress.length; i++)
- {
- if (Progress.BYTES.equals(colReaderProgress[i].getUnit()))
- {
- processTrace.addEvent(readerName, Constants.COLLECTION_READER_BYTES_PROCESSED, String.valueOf(colReaderProgress[i].getCompleted()), 0, null);
- }
- else if (Progress.ENTITIES.equals(colReaderProgress[i].getUnit()))
- {
- processTrace.addEvent(readerName, Constants.COLLECTION_READER_DOCS_PROCESSED, String.valueOf(colReaderProgress[i].getCompleted()), 0, null);
- }
- }
-
- synchronized (procTr)
- {
- eList = procTr.getEventsByComponentName(readerName, true);
- }
- copyComponentEvents("COLLECTION READER PROCESSING TIME", eList, processTrace);
- eList.clear();
- processTrace.addEvent(readerName, "Last Entity ID Read", cpEngine.getLastProcessedDocId(), 0, null);
- }
-
- LinkedList processors = cpEngine.getAllProcessingContainers();
- for (int i = 0; i < processors.size(); i++)
- {
- ProcessingContainer container = (ProcessingContainer) processors.get(i);
- synchronized (procTr)
- {
- eList = procTr.getEventsByComponentName(container.getName(), true);
- }
- copyComponentEvents("Process", eList, processTrace);
-
-
- processTrace.addEvent(container.getName(), "Documents Processed", String.valueOf(container.getProcessed()), 0, null);
- String status = decodeStatus(container.getStatus());
- processTrace.addEvent(container.getName(), "Processor Status", status, 0, null);
-
- long bytesIn = container.getBytesIn();
- processTrace.addEvent(container.getName(), "Processor BYTESIN", String.valueOf(bytesIn), 0, null);
-
- long bytesOut = container.getBytesOut();
- processTrace.addEvent(container.getName(), "Processor BYTESOUT", String.valueOf(bytesOut), 0, null);
-
- int restartCount = container.getRestartCount();
- processTrace.addEvent(container.getName(), "Processor Restarts", String.valueOf(restartCount), 0, null);
-
- int retryCount = container.getRetryCount();
- processTrace.addEvent(container.getName(), "Processor Retries", String.valueOf(retryCount), 0, null);
-
- int filteredCount = container.getFilteredCount();
- processTrace.addEvent(container.getName(), "Filtered Entities", String.valueOf(filteredCount), 0, null);
-
- long remainingCount = container.getRemaining();
- processTrace.addEvent(container.getName(), "Processor Remaining", String.valueOf(remainingCount), 0, null);
-
- HashMap aMap = container.getAllStats();
-
- if ( aMap.keySet() != null )
- {
- if (System.getProperty("SHOW_CUSTOM_STATS") != null )
- UIMAFramework.getLogger(this.getClass()).log(Level.FINEST,"Adding Custom Stats");
- Iterator it = aMap.keySet().iterator();
- while( it != null && it.hasNext() )
- {
-
- String key = (String)it.next();
- if ( key != null )
- {
- Object o = aMap.get(key);
- if ( o instanceof String )
- {
- processTrace.addEvent(container.getName(), key, (String)o, 0, null);
- if (System.getProperty("SHOW_CUSTOM_STATS") != null )
- UIMAFramework.getLogger(this.getClass()).log(Level.FINEST,"Custom String Stat-"+key+" Value="+(String)o);
- }
- else if ( o instanceof Integer )
- {
- processTrace.addEvent(container.getName(), key, String.valueOf( ((Integer)o).intValue()), 0, null);
- if (System.getProperty("SHOW_CUSTOM_STATS") != null )
- UIMAFramework.getLogger(this.getClass()).log(Level.FINEST,"Custom Integer Stat-"+key+" Value="+((Integer)o).intValue());
- }
- else
- {
- if ( UIMAFramework.getLogger().isLoggable(Level.FINEST) )
- {
- UIMAFramework.getLogger(this.getClass()).log(Level.FINEST, "Invalid Type Found When Generating Status For "+key+". Type::"+o.getClass().getName()+" Not supported. Use Integer or String instead.");
- }
- }
- }
- }
- }
- try
- {
- String lastDocId = container.getLastProcessedEntityId();
- if (lastDocId != null)
- {
- processTrace.addEvent(container.getName(), "Processor Last EntityId", lastDocId, 0, null);
- }
- }
- catch (Exception e)
- {
- e.printStackTrace();
- }
- }
- }
- catch (Exception e)
- {
- e.printStackTrace();
- }
- return processTrace;
- }
-
- /**
- * @param processors
- * @param procTr
- * @param processTrace
- */
- private void createDefaultProcessTrace(CasProcessor[] aProcessors, ProcessTrace srcProcTr, ProcessTrace aProcessTrace)
- {
- for (int i = 0; aProcessors != null && i < aProcessors.length; i++)
- {
- String name = aProcessors[i].getProcessingResourceMetaData().getName();
- if (name == null)
- {
- name = aProcessors[i].getClass().getName();
- }
- synchronized (srcProcTr)
- {
-
- List eventList = srcProcTr.getEventsByComponentName(name, false);
- for (int j = 0; j < eventList.size(); j++)
- {
- ProcessTraceEvent prEvent = (ProcessTraceEvent) eventList.get(j);
- aProcessTrace.addEvent(prEvent);
- }
- }
- }
- }
- /**
- * Returns current CPE progress. How many entities processed and bytes processed.
- */
- public Progress[] getProgress()
- {
- return cpEngine.getProgress();
- }
- /**
- * Returns a CPE descriptor as a String
- *
- * @param aList - list of components
- *
- * @return - descriptor populated with a given components
- */
- public String getDescriptor(List aList) throws ResourceConfigurationException
- {
- return cpeFactory.getDescriptor(aList);
- }
- /**
- * Returns a {@link SynchPoint} object initialized by the Collection Reader if the Collection Reader
- * implements {@link RecoverableCollectionReader}. The synchpoint object contains
- * the current snapshot that includes the last document processed.
- *
- * @return - instance of SynchPoint if the Collection Reader is recoverable, null otherwise
- */
- public SynchPoint getSynchPoint()
- {
- SynchPoint synchPoint = null;
- // Check if the CR is recoverable
- if ( collectionReader != null && collectionReader instanceof RecoverableCollectionReader )
- {
- synchPoint = ((RecoverableCollectionReader)collectionReader).getSynchPoint();
- }
- return synchPoint;
- }
+public class BaseCPMImpl implements BaseCPM, Runnable {
+ private boolean defaultProcessTrace;
+
+ private CPMEngine cpEngine = null;
+
+ private ProcessTrace procTr = null;
+
+ BaseCollectionReader collectionReader = null;
+
+ private Checkpoint checkpoint = null;
+
+ private CheckpointData checkpointData = null;
+
+ private long num2Process = -1; // Default ALL
+
+ private boolean killed = false;
+
+ private boolean completed = false;
+
+ private CPEFactory cpeFactory = null;
+
+ private boolean useJediiReport = true;
+
+ private Map mEventTypeMap;
+
+ public CPMThreadGroup cpmThreadGroup = null;
+
+ /**
+ * Instantiates and initializes CPE Factory with a given CPE Descriptor and defaults.
+ *
+ * @param aDescriptor -
+ * parsed CPE descriptor
+ * @throws Exception
+ */
+ public BaseCPMImpl(CpeDescription aDescriptor) throws Exception {
+ this(aDescriptor, null, true, UIMAFramework.getDefaultPerformanceTuningProperties());
+ cpmThreadGroup = new CPMThreadGroup("CPM Thread Group");
+ }
+
+ /**
+ * Instantiates and initializes CPE Factory responsible for creating individual components that
+ * are part of the processing pipeline.
+ *
+ * @param aDescriptor -
+ * parsed CPE descriptor
+ * @param aResourceManager -
+ * ResourceManager instance to be used by the CPE
+ * @param aDefaultProcessTrace -
+ * ProcessTrace instance to capture events and stats
+ * @throws Exception
+ */
+ public BaseCPMImpl(CpeDescription aDescriptor, ResourceManager aResourceManager,
+ boolean aDefaultProcessTrace, Properties aProps) throws Exception {
+ cpeFactory = new CPEFactory(aDescriptor, aResourceManager);
+ defaultProcessTrace = aDefaultProcessTrace;
+ cpmThreadGroup = new CPMThreadGroup("CPM Thread Group");
+ init(false, aProps);
+ }
+
+ /**
+ * Parses CPE descriptor
+ *
+ * @param mode -
+ * indicates if the CPM should use a static descriptor or one provided
+ * @param aDescriptor -
+ * provided descriptor path
+ * @param aResourceManager
+ * ResourceManager to be used by CPM
+ *
+ * @throws Exception
+ */
+ public BaseCPMImpl(Boolean mode, String aDescriptor, ResourceManager aResourceManager)
+ throws Exception {
+ cpmThreadGroup = new CPMThreadGroup("CPM Thread Group");
+ cpeFactory = new CPEFactory(aResourceManager);
+ if (mode == null) {
+ defaultProcessTrace = true;
+ cpeFactory.parse();
+ } else {
+ defaultProcessTrace = mode.booleanValue();
+ cpeFactory.parse(aDescriptor);
+ }
+ init(mode == null, UIMAFramework.getDefaultPerformanceTuningProperties());
+ }
+
+ /**
+ * Plugs in custom perfomance tunning parameters
+ *
+ * @param aPerformanceTuningSettings
+ */
+ public void setPerformanceTuningSettings(Properties aPerformanceTuningSettings) {
+ cpEngine.setPerformanceTuningSettings(aPerformanceTuningSettings);
+ }
+
+ /**
+ * Plugs in a given {@link ProcessControllerAdapter}. The CPM uses this adapter to request Cas
+ * Processor restarts and shutdown.
+ *
+ * @param aPca -
+ * instance of the ProcessControllerAdapter
+ */
+ public void setProcessControllerAdapter(ProcessControllerAdapter aPca) {
+ cpEngine.setProcessControllerAdapter(aPca);
+ }
+
+ /**
+ * Sets Jedii-style reporting resources and sets the global flag to indicate what report-style to
+ * use at the end of processing. Jedii-style reporting shows a summary for this run. The CPM
+ * default report shows more detail information.
+ *
+ * @param aUseJediiReport
+ */
+ public void setJediiReport(boolean aUseJediiReport) {
+ mEventTypeMap = new HashMap();
+ mEventTypeMap.put(ProcessTraceEvent.ANALYSIS_ENGINE, "TAE");
+ mEventTypeMap.put(ProcessTraceEvent.ANALYSIS, "Annotator");
+ mEventTypeMap.put("CAS_PROCESSOR", "CAS Consumer");
+ useJediiReport = aUseJediiReport;
+ }
+
+ /**
+ * Instantiates and initializes a CPE.
+ *
+ * @param aDummyCasProcessor -
+ *
+ * @throws Exception
+ */
+ public void init(boolean aDummyCasProcessor, Properties aProps) throws Exception {
+ String uimaTimerClass = cpeFactory.getCPEConfig().getTimerImpl();
+ try {
+ new TimerFactory(uimaTimerClass);
+ } catch (Exception e) {
+ // e.printStackTrace();
+ if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
+ UIMAFramework.getLogger(this.getClass()).logrb(Level.FINEST, this.getClass().getName(),
+ "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
+ "UIMA_CPM_use_default_timer__FINEST",
+ new Object[] { Thread.currentThread().getName() });
+
+ }
+ }
+ UimaTimer uimaTimer = TimerFactory.getTimer();
+ if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
+ UIMAFramework.getLogger(this.getClass()).logrb(
+ Level.FINEST,
+ this.getClass().getName(),
+ "process",
+ CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
+ "UIMA_CPM_use_custom_timer__FINEST",
+ new Object[] { Thread.currentThread().getName(),
+ uimaTimer.getClass().getName() });
+ }
+
+ procTr = new ProcessTrace_impl(uimaTimer, aProps);
+ String checkpointFileName = null;
+ if (cpeFactory.getCPEConfig().getCheckpoint() != null
+ && cpeFactory.getCPEConfig().getCheckpoint().getFilePath() != null) {
+ // Retrieve from CPM configuration a name of the checkpoint file where the
+ // CPM's runtime stats be deposited.
+ checkpointFileName = cpeFactory.getCPEConfig().getCheckpoint().getFilePath();
+ }
+ if (checkpointFileName != null && checkpointFileName.trim().length() > 0) {
+ File checkpointFile = new File(checkpointFileName);
+ checkpoint = new Checkpoint(this, checkpointFileName, cpeFactory.getCPEConfig()
+ .getCheckpoint().getFrequency());
+ // Check if the checkpoint file already exists. If it does, the CPM did not complete
+ // successfully during the previous run and CPM will start in recovery mode, restoring all
+ // totals and status's from the recovered checkpoint. The processing pipeline state will
+ // restored to the state as of before foreced shutdown. All CasProcessors that were disabled
+ // during that run will remain disabled. The CollectionReader will be advanced to the
+ // entity last processed by the previous CPM.
+ if (checkpointFile.exists()) {
+ try {
+ Object restoredObject = checkpoint.restoreFromCheckpoint();
+ if (restoredObject != null && restoredObject instanceof CheckpointData) {
+ checkpointData = (CheckpointData) restoredObject;
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ }
+ // Instantiate class responsible for processing
+ cpEngine = new CPMEngine(cpmThreadGroup, cpeFactory, procTr, checkpointData);
+ if (!aDummyCasProcessor) {
+ int concurrentThreadCount = cpeFactory.getCpeDescriptor().getCpeCasProcessors()
+ .getConcurrentPUCount();
+ for (int threadCount = 0; threadCount < concurrentThreadCount; threadCount++) {
+ CasProcessor[] casProcessors = cpeFactory.getCasProcessors();
+ for (int i = 0; i < casProcessors.length; i++) {
+ if (UIMAFramework.getLogger().isLoggable(Level.CONFIG)) {
+ UIMAFramework.getLogger(this.getClass()).logrb(
+ Level.CONFIG,
+ this.getClass().getName(),
+ "process",
+ CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
+ "UIMA_CPM_add_cp__CONFIG",
+ new Object[] { Thread.currentThread().getName(),
+ casProcessors[i].getProcessingResourceMetaData().getName() });
+ }
+ addCasProcessor(casProcessors[i]);
+ }
+ }
+ }
+ int casPoolSize = 0;
+ try {
+ casPoolSize = cpeFactory.getCpeDescriptor().getCpeCasProcessors().getCasPoolSize();
+ casPoolSize = (casPoolSize == -1) ? 0 : casPoolSize;
+ cpEngine.setPoolSize(casPoolSize);
+ } catch (NumberFormatException e) {
+ // PoolSize is currently an optional parameter. If it does not exist in CPE descriptor a
+ // default
+ // value will be derived from defined queue sizes
+ }
+
+ try {
+ int iqSize = 0;
+ if (casPoolSize == 0) {
+ iqSize = cpeFactory.getCpeDescriptor().getCpeCasProcessors().getInputQueueSize();
+ }
+ cpEngine.setInputQueueSize(casPoolSize == 0 ? iqSize : casPoolSize);
+ } catch (NumberFormatException e) {
+ throw new Exception(CpmLocalizedMessage.getLocalizedMessage(CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
+ "UIMA_CPM_EXP_queue_size_not_defined__WARNING", new Object[] {
+ Thread.currentThread().getName(), "inputQueueSize" }));
+ }
+ try {
+ int oqSize = 0;
+ if (casPoolSize == 0) {
+ oqSize = cpeFactory.getCpeDescriptor().getCpeCasProcessors().getOutputQueueSize();
+ }
+ cpEngine.setOutputQueueSize(casPoolSize == 0 ? oqSize : casPoolSize + 2);
+ } catch (NumberFormatException e) {
+ throw new Exception(CpmLocalizedMessage.getLocalizedMessage(CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
+ "UIMA_CPM_EXP_queue_size_not_defined__WARNING", new Object[] {
+ Thread.currentThread().getName(), "outputQueueSize" }));
+ }
+ try {
+ int threadCount = cpeFactory.getCpeDescriptor().getCpeCasProcessors().getConcurrentPUCount();
+ cpEngine.setConcurrentThreadSize(threadCount);
+ } catch (NumberFormatException e) {
+ throw new Exception(CpmLocalizedMessage.getLocalizedMessage(CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
+ "UIMA_CPM_EXP_invalid_component_reference__WARNING", new Object[] {
+ Thread.currentThread().getName(), "casProcessors",
+ "processingUnitThreadCount" }));
+ }
+ }
+
+ /**
+ * Returns {@link CPEConfig} object holding current CPE configuration
+ *
+ * @return CPEConfig instance
+ *
+ * @throws Exception
+ */
+ public CpeConfiguration getCPEConfig() throws Exception {
+ return cpeFactory.getCPEConfig();
+ }
+
+ /*
+ * Returns All CasProcessors currently in the processing pipeline
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#getCasProcessors()
+ */
+ public CasProcessor[] getCasProcessors() {
+ CasProcessor[] casProcs = cpEngine.getCasProcessors();
+ return casProcs == null ? new CasProcessor[0] : casProcs;
+ }
+
+ /*
+ * Adds given CasProcessor to the processing pipeline. A new CasProcessor is appended to the
+ * current list.
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#addCasProcessor(org.apache.uima.collection.base_cpm.CasProcessor)
+ */
+ public void addCasProcessor(CasProcessor aCasProcessor) throws ResourceConfigurationException {
+ cpEngine.addCasProcessor(aCasProcessor);
+ }
+
+ /*
+ * Adds given CasProcessor to the processing pipeline. A new CasProcessor is inserted into a given
+ * spot in the current list.
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#addCasProcessor(org.apache.uima.collection.base_cpm.CasProcessor,
+ * int)
+ */
+ public void addCasProcessor(CasProcessor aCasProcessor, int aIndex)
+ throws ResourceConfigurationException {
+ cpEngine.addCasProcessor(aCasProcessor, aIndex);
+ }
+
+ /*
+ * Removes given CasProcessor from the processing pipeline.
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#removeCasProcessor(org.apache.uima.collection.base_cpm.CasProcessor)
+ */
+ public void removeCasProcessor(CasProcessor aCasProcessor) {
+ cpEngine.removeCasProcessor(0);
+ }
+
+ /*
+ * Disables given CasProcessor in the existing processing pipeline.
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#disableCasProcessor(org.apache.uima.collection.base_cpm.CasProcessor)
+ */
+ public void disableCasProcessor(String aCasProcessorName) {
+
+ cpEngine.disableCasProcessor(aCasProcessorName);
+ }
+
+ /*
+ * Disables given CasProcessor in the existing processing pipeline.
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#disableCasProcessor(org.apache.uima.collection.base_cpm.CasProcessor)
+ */
+ public void enableCasProcessor(String aCasProcessorName) {
+
+ cpEngine.enableCasProcessor(aCasProcessorName);
+ }
+
+ /*
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#isSerialProcessingRequired()
+ */
+ public boolean isSerialProcessingRequired() {
+ return false;
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#setSerialProcessingRequired(boolean)
+ */
+ public void setSerialProcessingRequired(boolean aRequired) {
+ }
+
+ /*
+ * Returns true if this cpEngine pauses on exception. False otherwise.
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#isPauseOnException()
+ */
+ public boolean isPauseOnException() {
+ return cpEngine.isPauseOnException();
+ }
+
+ /*
+ * Defines if cpEngine should pause on exception
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#setPauseOnException(boolean)
+ */
+ public void setPauseOnException(boolean aPause) {
+ cpEngine.setPauseOnException(aPause);
+ }
+
+ /*
+ * Adds Event Listener. Important events like, end of entity processing, exceptions, etc will be
+ * sent to the registered listeners.
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#addStatusCallbackListener(org.apache.uima.collection.base_cpm.BaseStatusCallbackListener)
+ */
+ public void addStatusCallbackListener(BaseStatusCallbackListener aListener) {
+ cpEngine.addStatusCallbackListener(aListener);
+ }
+
+ /*
+ * Remoces named listener from the listener list.
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#removeStatusCallbackListener(org.apache.uima.collection.base_cpm.BaseStatusCallbackListener)
+ */
+ public void removeStatusCallbackListener(BaseStatusCallbackListener aListener) {
+ cpEngine.removeStatusCallbackListener(aListener);
+ }
+
+ /*
+ * Starting point for the CPE. Before starting processing, the CPE must deploy all CasProcessors.
+ * Once all are deployed the processing begins in Worker Thread called CPEngine. This thread
+ * blocks until the CPEngine thread completes. CPMWorker Thread finishes.
+ *
+ * @see java.lang.Runnable#run()
+ */
+ public void run() {
+ long start, end;
+ // name this thread
+ Thread.currentThread().setName("BaseCPMImpl-Thread");
+
+ start = System.currentTimeMillis();
+ if (!useJediiReport) {
+ procTr.startEvent("CPM", "CPM PROCESSING TIME", "");
+ }
+
+ // Specify how docs to process
+ cpEngine.setNumToProcess(num2Process);
+ try {
+ // Deploy all CAS processors
+ cpEngine.deployCasProcessors();
+ cpEngine.setCollectionReader(collectionReader);
+
+ // Start the checkpoint thread
+ if (checkpoint != null) {
+ new Thread(checkpoint).start();
+ }
+ cpEngine.start();
+ // Joing the CPMWorker Thread and wait until it finishes
+ cpEngine.join();
+
+ completed = true;
+ // If the entire collection has been processed there is no need for a checkpoint.
+ // Delete it, there is no need to recover anything. Otherwise, the CPM has been killed
+ // and may need to be recovered. Checkpoint file contains the status of the CPM, including
+ // last document processed, status of all CasProcessors along with all counts and totals.
+ if (!killed && checkpoint != null) {
+ checkpoint.stop();
+ checkpoint.delete();
+ checkpoint = null;
+ }
+ // Terminate all threads and running services
+ cpEngine.stopCasProcessors(false);
+ // Notify Listeners that the processing pipeline has finished
+ if (!useJediiReport) {
+ procTr.endEvent("CPM", "CPM PROCESSING TIME", "success");
+ }
+ end = 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_show_total_time_in_cpm__FINEST",
+ new Object[] { Thread.currentThread().getName(),
+ String.valueOf(end - start) });
+ }
+ } catch (AbortCPMException e) {
+ if (!useJediiReport) {
+ procTr.endEvent("CPM", "CPM PROCESSING TIME", "failed");
+ }
+ // Terminate all threads and running services
+ try {
+ cpEngine.stopCasProcessors(true);
+ } catch (Exception ex) {
+ UIMAFramework.getLogger(this.getClass()).log(Level.SEVERE, "" + ex);
+ }
+ killed = true;
+ } catch (Exception e) {
+ if (!useJediiReport) {
+ procTr.endEvent("CPM", "CPM PROCESSING TIME", "failed");
+ }
+ UIMAFramework.getLogger(this.getClass()).log(Level.SEVERE, "" + e);
+ killed = true;
+ ArrayList statusCbL = cpEngine.getCallbackListeners();
+ EntityProcessStatusImpl enProcSt = new EntityProcessStatusImpl(procTr, true);
+ // e is the actual exception.
+ enProcSt.addEventStatus("CPM", "Failed", e);
+
+ // Notify all listeners that the CPM has finished processing
+ for (int j = 0; j < statusCbL.size(); j++) {
+ BaseStatusCallbackListener st = (BaseStatusCallbackListener) statusCbL.get(j);
+ if (st instanceof StatusCallbackListener) {
+ ((StatusCallbackListener) st).entityProcessComplete(null, enProcSt);
+ }
+ }
+ }
+
+ if (cpEngine.isKilled()) {
+ killed = 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_cpm_stopped__FINEST",
+ new Object[] { Thread.currentThread().getName(), String.valueOf(killed) });
+ }
+ ArrayList statusCbL = cpEngine.getCallbackListeners();
+ // Notify all listeners that the CPM has finished processing
+ for (int j = 0; j < statusCbL.size(); j++) {
+ BaseStatusCallbackListener st = (BaseStatusCallbackListener) statusCbL.get(j);
+ if (!killed) {
+ st.collectionProcessComplete();
+ } else {
+ st.aborted();
+ }
+ }
+ }
+
+ /**
+ * Called to cleanup CPE on shutdown
+ *
+ */
+ public void finalizeIt() {
+ // Do cleanup before terminating self
+ cpEngine.cleanup();
+
+ }
+
+ /**
+ * This method is called by an applications to begin CPM processing with a given Collection. It
+ * just creates a new thread and starts it.
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#process(org.apache.uima.collection.base_cpm.BaseCollectionReader)
+ * @deprecated
+ */
+ public void process(BaseCollectionReader aCollectionReader)
+ throws ResourceInitializationException {
+ // Retrieve number of entities to process from the CPM configuration
+ try {
+ num2Process = cpeFactory.getCPEConfig().getNumToProcess();
+ } catch (InstantiationException e) {
+ throw new ResourceInitializationException(e);
+ }
+ collectionReader = aCollectionReader;
+ if (cpeFactory.isDefault()) {
+ cpeFactory.addCollectionReader(collectionReader);
+ }
+ cpmThreadGroup.setProcessTrace(procTr);
+ cpmThreadGroup.setListeners(cpEngine.getCallbackListeners());
+ new Thread(this).start();
+ }
+
+ /*
+ * This method is called by an application to begin processing given Collection. It creates a new
+ * thread, adds it to a ThreadGroup and starts it.
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#process(org.apache.uima.collection.base_cpm.BaseCollectionReader)
+ */
+ public void process() throws ResourceInitializationException {
+ // Retrieve number of entities to process from the CPM configuration
+ try {
+ num2Process = cpeFactory.getCPEConfig().getNumToProcess();
+ if (collectionReader == null) {
+ collectionReader = cpeFactory.getCollectionReader();
+ }
+ } catch (InstantiationException e) {
+ throw new ResourceInitializationException(e);
+ } catch (Exception e) {
+ throw new ResourceInitializationException(e);
+ }
+ if (cpeFactory.isDefault()) {
+ cpeFactory.addCollectionReader(collectionReader);
+ }
+ cpmThreadGroup.setProcessTrace(procTr);
+ cpmThreadGroup.setListeners(cpEngine.getCallbackListeners());
+
+ new Thread(this).start();
+ }
+
+ /**
+ *
+ * This method is called by an applications to begin CPM processing with a given Collection. It
+ * just creates a new thread and starts it.
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#process(org.apache.uima.collection.base_cpm.BaseCollectionReader,
+ * int)
+ * @deprecated
+ */
+ public void process(BaseCollectionReader aCollectionReader, int aBatchSize)
+ throws ResourceInitializationException {
+ // Let the application define the size of Collection.
+ num2Process = aBatchSize;
+ collectionReader = aCollectionReader;
+ if (cpeFactory.isDefault()) {
+ cpeFactory.addCollectionReader(collectionReader);
+ }
+ cpmThreadGroup.setProcessTrace(procTr);
+ cpmThreadGroup.setListeners(cpEngine.getCallbackListeners());
+
+ new Thread(cpmThreadGroup, this).start();
+ }
+
+ /**
+ * Sets the Collection Reader for this CPE.
+ *
+ * @param aCollectionReader
+ * the collection reader
+ */
+ public void setCollectionReader(BaseCollectionReader aCollectionReader) {
+ collectionReader = aCollectionReader;
+ if (cpeFactory.isDefault()) {
+ cpeFactory.addCollectionReader(collectionReader);
+ }
+ }
+
+ /**
+ * Returns a Collection Reader for this CPE.
+ *
+ * @return the collection reader
+ */
+ public BaseCollectionReader getCollectionReader() {
+ try {
+ if (this.collectionReader == null) {
+ this.collectionReader = this.cpeFactory.getCollectionReader();
+ }
+ return this.collectionReader;
+ } catch (ResourceConfigurationException e) {
+ throw new UIMARuntimeException(e);
+ }
+ }
+
+ /*
+ * Returns current state of the CPM.
+ *
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#isProcessing()
+ */
+ public boolean isProcessing() {
+ return cpEngine.isRunning();
+ }
+
+ /*
+ * Pauses the CPM
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#pause()
+ */
+ public void pause() {
+ cpEngine.pauseIt();
+ if (checkpoint != null) {
+ checkpoint.doCheckpoint();
+ checkpoint.pause();
+ }
+ }
+
+ /*
+ * Returns true if the CPM is in pause state
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#isPaused()
+ */
+ public boolean isPaused() {
+ return cpEngine.isPaused();
+ }
+
+ /*
+ * Resumes the CPM
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#resume(boolean)
+ */
+ public void resume(boolean aRetryFailed) {
+ resume();
+ }
+
+ /*
+ * Resumes the CPM
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#resume()
+ */
+ public void resume() {
+ cpEngine.resumeIt();
+ if (checkpoint != null) {
+ checkpoint.resume();
+ }
+ }
+
+ /**
+ * Kills the CPM hard. CASes in transit are not processed.
+ *
+ *
+ */
+ public void kill() {
+ if (UIMAFramework.getLogger().isLoggable(Level.WARNING)) {
+ UIMAFramework.getLogger(this.getClass()).logrb(Level.WARNING, this.getClass().getName(),
+ "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_killing_cpm__WARNING",
+ new Object[] { Thread.currentThread().getName() });
+ }
+ killed = true;
+ // Stop processing pipeline. The CPMWorker will finish processing of the current
+ // entity through the processing pipeline and than will terminate processing.
+ cpEngine.killIt();
+
+ // If valid checkpoint reference and not already deleted do the next checkpoint and stop
+ // the checkpoint thread. Checkpoint file may be deleted if the CPM successfully completes
+ // its run. See BaseCPM#run().
+ if (checkpoint != null && !completed) {
+ checkpoint.doCheckpoint();
+ checkpoint.stop();
+ }
+
+ }
+
+ /*
+ * Stops the CPM and all of its processing components. The CPM finishes processing of all CASes
+ * that are in its queues.
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#stop()
+ */
+ public void stop() {
+ if (UIMAFramework.getLogger().isLoggable(Level.WARNING)) {
+ UIMAFramework.getLogger(this.getClass()).logrb(Level.WARNING, this.getClass().getName(),
+ "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE, "UIMA_CPM_stop_cpm__WARNING",
+ new Object[] { Thread.currentThread().getName() });
+ }
+ killed = true;
+ // Stop processing pipeline. The CPMWorker will finish processing of the current
+ // entity through the processing pipeline and than will terminate processing.
+ cpEngine.stopIt();
+
+ // If valid checkpoint reference and not already deleted do the next checkpoint and stop
+ // the checkpoint thread. Checkpoint file may be deleted if the CPM successfully completes
+ // its run. See BaseCPM#run().
+ if (checkpoint != null && !completed) {
+ checkpoint.doCheckpoint();
+ checkpoint.stop();
+ }
+ }
+
+ /*
+ * Stops/kills the CPM and all of its processing components.
+ *
+ * @deprecated
+ *
+ * @see org.apache.uima.collection.base_cpm.BaseCPM#stop()
+ */
+ public void asynchStop() {
+ if (UIMAFramework.getLogger().isLoggable(Level.INFO)) {
+ UIMAFramework.getLogger(this.getClass()).logrb(Level.WARNING, this.getClass().getName(),
+ "process", CPMUtils.CPM_LOG_RESOURCE_BUNDLE,
+ "UIMA_CPM_asynch_stop_cpm__WARNING",
+ new Object[] { Thread.currentThread().getName() });
+ }
+ killed = true;
+ // Stop processing pipeline. The CPMWorker will finish processing of the current
+ // entity through the processing pipeline and than will terminate processing.
+ cpEngine.asynchStop();
+
+ // If valid checkpoint reference and not already deleted do the next checkpoint and stop
+ // the checkpoint thread. Checkpoint file may be deleted if the CPM successfully completes
+ // its run. See BaseCPM#run().
+ if (checkpoint != null && !completed) {
+ checkpoint.doCheckpoint();
+ checkpoint.stop();
+ }
+ }
+
+ /*
+ * Returns a String describing a given CASProcessor state.
+ *
+ * @param aStatus - status of the CasProcessor @return - String corresponding to a given state of
+ * the CasProcessor
+ */
+ private String decodeStatus(int aStatus) {
+ try {
+ switch (aStatus) {
+ case Constants.CAS_PROCESSOR_COMPLETED:
+ return Constants.COMPLETED;
+ case Constants.CAS_PROCESSOR_DISABLED:
+ return Constants.DISABLED;
+ case Constants.CAS_PROCESSOR_READY:
+ return Constants.READY;
+ case Constants.CAS_PROCESSOR_RUNNING:
+ return Constants.RUNNING;
+ case Constants.CAS_PROCESSOR_KILLED:
+ return Constants.KILLED;
+ }
+ } catch (NumberFormatException e) {
+ e.printStackTrace();
+ }
+ return Constants.UNKNOWN;
+ }
+
+ /*
+ * Copies events of a given type found in the list to a provided ProcessTrace instance
+ *
+ * @param - aEvType, event type to copy from the list @param - List, list of events @param
+ * ProcessTrace, where to copy events of a given type
+ *
+ */
+ private void copyComponentEvents(String aEvType, List aList, ProcessTrace aPTr)
+ throws IOException {
+ for (int i = 0; i < aList.size(); i++) {
+ ProcessTraceEvent prEvent = (ProcessTraceEvent) aList.get(i);
+ if (aEvType != null && aEvType.equals(prEvent.getType())) {
+ aPTr.addEvent(prEvent);
+ }
+ }
+
+ }
+
+ /**
+ * Helper method to display stats and totals
+ *
+ * @param aProcessTrace -
+ * trace containing stats
+ *
+ * @param aNumDocsProcessed -
+ * number of entities processed so far
+ */
+ public void displayStats(ProcessTrace aProcessTrace, int aNumDocsProcessed) {
+ if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
+ UIMAFramework.getLogger(this.getClass()).log(Level.FINEST,
+ "Documents Processed: " + aNumDocsProcessed);
+ }
+ // count total time
+ int totalTime = 0;
+ Iterator it = aProcessTrace.getEvents().iterator();
+ while (it.hasNext()) {
+ ProcessTraceEvent event = (ProcessTraceEvent) it.next();
+ // Dont add total time the CPM ran for. Just add all of the times of all components to
+ // get the time.
+ if ("CPM".equals(event.getComponentName())) {
+ if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
+ UIMAFramework.getLogger(this.getClass()).log(
+ Level.FINEST,
+ "Current Component::" + event.getComponentName() + " Time::"
+ + event.getDuration());
+ }
+ continue;
+ }
+ if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
+ UIMAFramework.getLogger(this.getClass()).log(Level.FINEST,
+ "Current Component::" + event.getComponentName());
+ }
+ totalTime += event.getDuration();
+ }
+ float totalTimeSeconds = (float) totalTime / 1000;
+ if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
+ UIMAFramework.getLogger(this.getClass()).log(Level.FINEST,
+ "Total Time: " + totalTimeSeconds + " seconds");
+ }
+
+ // create root tree node
+ if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
+ UIMAFramework.getLogger(this.getClass()).log(Level.FINEST,
+ "100% (" + totalTime + "ms) - Collection Processing Engine");
+ }
+ // build tree
+ it = aProcessTrace.getEvents().iterator();
+ while (it.hasNext()) {
+ ProcessTraceEvent event = (ProcessTraceEvent) it.next();
+ buildEventTree(event, totalTime);
+ }
+ }
+
+ /**
+ * Helper method to help build the CPM report
+ *
+ * @param aEvent
+ * @param aTotalTime
+ */
+ public void buildEventTree(ProcessTraceEvent aEvent, int aTotalTime) {
+ // Skip reporting the CPM time.This time has already been acquired by summing up
+ // times from all individual components
+ if ("CPM".equals(aEvent.getComponentName())) {
+ return;
+ }
+
+ int duration = aEvent.getDuration();
+ float pct = (float) ((duration * 100 * 10) / aTotalTime) / 10;
+
+ String type = (String) mEventTypeMap.get(aEvent.getType());
+ if (type == null) {
+ type = aEvent.getType();
+ }
+
+ if (System.getProperty("DEBUG") != null)
+ UIMAFramework.getLogger(this.getClass()).log(
+ Level.FINEST,
+ "" + pct + "% (" + duration + "ms) - " + aEvent.getComponentName() + " ("
+ + type + ")");
+ Iterator it = aEvent.getSubEvents().iterator();
+ while (it.hasNext()) {
+ ProcessTraceEvent event = (ProcessTraceEvent) it.next();
+ buildEventTree(event, aTotalTime);
+ }
+ }
+
+ /**
+ * Returns PerformanceReport for the CPM. This report contains a snapshot of the CPM state.
+ *
+ */
+ public ProcessTrace getPerformanceReport() {
+ Map perfReport = cpEngine.getStats();
+ Progress[] colReaderProgress = (Progress[]) perfReport.get("COLLECTION_READER_PROGRESS");
+
+ ProcessTrace processTrace = new ProcessTrace_impl(cpEngine.getPerformanceTuningSettings());
+ if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
+ UIMAFramework.getLogger(this.getClass()).log(Level.FINEST,
+ "-------------------------------------------");
+ }
+ if (useJediiReport) {
+ try {
+ synchronized (procTr) {
+ List eventList = procTr.getEvents();
+
+ for (int j = 0; eventList != null && j < eventList.size(); j++) {
+ ProcessTraceEvent prEvent = (ProcessTraceEvent) eventList.get(j);
+ processTrace.addEvent(prEvent);
+ }
+ }
+ return processTrace;
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ if (defaultProcessTrace) {
+ createDefaultProcessTrace(getCasProcessors(), procTr, processTrace);
+
+ return processTrace;
+ }
+ try {
+
+ // To facilitate recovery from CPM untimely shutdown ( due to external STOP), it must
+ // have access to last entity id. CollectionReader will use this marker to synch itself
+ // up to the last known entity before the CPM shut itself down. More complicated
+ // recovery mechanism may be supported as long as CAS METADATA contains information
+ // about the last known point. For example, in case WF Large Store the cas must
+ // hold the entire EDATA frame, containing restart information. Its up to the
+ // CollectionReader to know what part of the CAS Metadata should be used for recovery.
+ // The "last cas" CasMetaData is added after each successfull read from the CollectionReader
+ // by the cpEngine in its run() processing loop.
+ CasMetaData casMetaData = (CasMetaData) perfReport.get("CPM_LAST_CAS_METADATA");
+ if (casMetaData != null) {
+ NameValuePair[] nvp = casMetaData.getCasMetaData();
+ StringBuffer sb = new StringBuffer();
+ for (int i = 0; i < nvp.length && nvp[i] != null; i++) {
+ if (i != 0) {
+ // Add separator between name-value pairs. StringTokenizer will parse this string
+ // and will use the separator to extract nvp.
+ sb.append(",");
+ }
+ sb.append(nvp[i].getName() + "=" + (String) nvp[i].getValue());
+ }
+ processTrace.addEvent("CPM", "CPM_LAST_CAS_METADATA", sb.toString(), 0, null);
+ }
+ List eList = null;
+ synchronized (procTr) {
+ eList = procTr.getEventsByComponentName("CPM", true);
+ }
+ if (!useJediiReport) {
+ copyComponentEvents("CPM PROCESSING TIME", eList, processTrace);
+ }
+ eList.clear();
+ if (colReaderProgress != null) {
+ Long totalCollectionReaderTime = (Long) perfReport.get("COLLECTION_READER_TIME");
+ String readerName = collectionReader.getProcessingResourceMetaData().getName();
+ if (totalCollectionReaderTime != null) {
+ processTrace.addEvent(readerName, "COLLECTION_READER_TIME", String
+ .valueOf(totalCollectionReaderTime), 0, null);
+ }
+ for (int i = 0; i < colReaderProgress.length; i++) {
+ if (Progress.BYTES.equals(colReaderProgress[i].getUnit())) {
+ processTrace.addEvent(readerName, Constants.COLLECTION_READER_BYTES_PROCESSED, String
+ .valueOf(colReaderProgress[i].getCompleted()), 0, null);
+ } else if (Progress.ENTITIES.equals(colReaderProgress[i].getUnit())) {
+ processTrace.addEvent(readerName, Constants.COLLECTION_READER_DOCS_PROCESSED, String
+ .valueOf(colReaderProgress[i].getCompleted()), 0, null);
+ }
+ }
+
+ synchronized (procTr) {
+ eList = procTr.getEventsByComponentName(readerName, true);
+ }
+ copyComponentEvents("COLLECTION READER PROCESSING TIME", eList, processTrace);
+ eList.clear();
+ processTrace.addEvent(readerName, "Last Entity ID Read", cpEngine.getLastProcessedDocId(),
+ 0, null);
+ }
+
+ LinkedList processors = cpEngine.getAllProcessingContainers();
+ for (int i = 0; i < processors.size(); i++) {
+ ProcessingContainer container = (ProcessingContainer) processors.get(i);
+ synchronized (procTr) {
+ eList = procTr.getEventsByComponentName(container.getName(), true);
+ }
+ copyComponentEvents("Process", eList, processTrace);
+
+ processTrace.addEvent(container.getName(), "Documents Processed", String.valueOf(container
+ .getProcessed()), 0, null);
+ String status = decodeStatus(container.getStatus());
+ processTrace.addEvent(container.getName(), "Processor Status", status, 0, null);
+
+ long bytesIn = container.getBytesIn();
+ processTrace.addEvent(container.getName(), "Processor BYTESIN", String.valueOf(bytesIn), 0,
+ null);
+
+ long bytesOut = container.getBytesOut();
+ processTrace.addEvent(container.getName(), "Processor BYTESOUT", String.valueOf(bytesOut),
+ 0, null);
+
+ int restartCount = container.getRestartCount();
+ processTrace.addEvent(container.getName(), "Processor Restarts", String
+ .valueOf(restartCount), 0, null);
+
+ int retryCount = container.getRetryCount();
+ processTrace.addEvent(container.getName(), "Processor Retries", String.valueOf(retryCount),
+ 0, null);
+
+ int filteredCount = container.getFilteredCount();
+ processTrace.addEvent(container.getName(), "Filtered Entities", String
+ .valueOf(filteredCount), 0, null);
+
+ long remainingCount = container.getRemaining();
+ processTrace.addEvent(container.getName(), "Processor Remaining", String
+ .valueOf(remainingCount), 0, null);
+
+ HashMap aMap = container.getAllStats();
+
+ if (aMap.keySet() != null) {
+ if (System.getProperty("SHOW_CUSTOM_STATS") != null)
+ UIMAFramework.getLogger(this.getClass()).log(Level.FINEST, "Adding Custom Stats");
+ Iterator it = aMap.keySet().iterator();
+ while (it != null && it.hasNext()) {
+
+ String key = (String) it.next();
+ if (key != null) {
+ Object o = aMap.get(key);
+ if (o instanceof String) {
+ processTrace.addEvent(container.getName(), key, (String) o, 0, null);
+ if (System.getProperty("SHOW_CUSTOM_STATS") != null)
+ UIMAFramework.getLogger(this.getClass()).log(Level.FINEST,
+ "Custom String Stat-" + key + " Value=" + (String) o);
+ } else if (o instanceof Integer) {
+ processTrace.addEvent(container.getName(), key, String.valueOf(((Integer) o)
+ .intValue()), 0, null);
+ if (System.getProperty("SHOW_CUSTOM_STATS") != null)
+ UIMAFramework.getLogger(this.getClass()).log(
+ Level.FINEST,
+ "Custom Integer Stat-" + key + " Value="
+ + ((Integer) o).intValue());
+ } else {
+ if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
+ UIMAFramework
+ .getLogger(this.getClass())
+ .log(
+ Level.FINEST,
+ "Invalid Type Found When Generating Status For "
+ + key
+ + ". Type::"
+ + o.getClass().getName()
+ + " Not supported. Use Integer or String instead.");
+ }
+ }
+ }
+ }
+ }
+ try {
+ String lastDocId = container.getLastProcessedEntityId();
+ if (lastDocId != null) {
+ processTrace.addEvent(container.getName(), "Processor Last EntityId", lastDocId, 0,
+ null);
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ return processTrace;
+ }
+
+ /**
+ * @param processors
+ * @param procTr
+ * @param processTrace
+ */
+ private void createDefaultProcessTrace(CasProcessor[] aProcessors, ProcessTrace srcProcTr,
+ ProcessTrace aProcessTrace) {
+ for (int i = 0; aProcessors != null && i < aProcessors.length; i++) {
+ String name = aProcessors[i].getProcessingResourceMetaData().getName();
+ if (name == null) {
+ name = aProcessors[i].getClass().getName();
+ }
+ synchronized (srcProcTr) {
+
+ List eventList = srcProcTr.getEventsByComponentName(name, false);
+ for (int j = 0; j < eventList.size(); j++) {
+ ProcessTraceEvent prEvent = (ProcessTraceEvent) eventList.get(j);
+ aProcessTrace.addEvent(prEvent);
+ }
+ }
+ }
+ }
+
+ /**
+ * Returns current CPE progress. How many entities processed and bytes processed.
+ */
+ public Progress[] getProgress() {
+ return cpEngine.getProgress();
+ }
+
+ /**
+ * Returns a CPE descriptor as a String
+ *
+ * @param aList -
+ * list of components
+ *
+ * @return - descriptor populated with a given components
+ */
+ public String getDescriptor(List aList) throws ResourceConfigurationException {
+ return cpeFactory.getDescriptor(aList);
+ }
+
+ /**
+ * Returns a {@link SynchPoint} object initialized by the Collection Reader if the Collection
+ * Reader implements {@link RecoverableCollectionReader}. The synchpoint object contains the
+ * current snapshot that includes the last document processed.
+ *
+ * @return - instance of SynchPoint if the Collection Reader is recoverable, null otherwise
+ */
+ public SynchPoint getSynchPoint() {
+ SynchPoint synchPoint = null;
+ // Check if the CR is recoverable
+ if (collectionReader != null && collectionReader instanceof RecoverableCollectionReader) {
+ synchPoint = ((RecoverableCollectionReader) collectionReader).getSynchPoint();
+ }
+ return synchPoint;
+ }
}
Modified: incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/CPEConfig.java
URL: http://svn.apache.org/viewvc/incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/CPEConfig.java?view=diff&rev=478226&r1=478225&r2=478226
==============================================================================
--- incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/CPEConfig.java (original)
+++ incubator/uima/uimaj/trunk/uimaj-cpe/src/main/java/org/apache/uima/collection/impl/cpm/CPEConfig.java Wed Nov 22 08:43:54 2006
@@ -25,140 +25,140 @@
*
*/
-public class CPEConfig
-{
- private CheckpointConfig chConfig = null;
- private long numToProcess = 0;
- private String deployAs = "";
- private String timerClass = null;
- private String startWith = "";
- private long inputQueueMaxMemoryThreshold = 0;
- private long maxTimeToWait = -1;
- private String outputQueueClass = null;
-
- /**
- * Returns CPE checkpoint configuration
- *
- * @return CheckpointConfig object
- */
- public CheckpointConfig getChConfig()
- {
- return chConfig;
- }
-
- /**
- * Returns CPE statup mode as defined in the CPE descriptor
- *
- * @return - statup mode
- */
- public String getDeployAs()
- {
- return deployAs;
- }
-
- /**
- * Returns number of entities to process by the CPE.
- *
- * @return - number of entities to process
- */
- public long getNumToProcess()
- {
- return numToProcess;
- }
-
- /**
- * Returns an id of the first entity the {@link CollectionReader} will be told to read
- *
- * @return - id of the first entity to read
- */
- public String getStartWith()
- {
- return startWith;
- }
-
- /**
- * Returns the name of custom {@link UimaTimer} class.
- *
- * @return - class as String
- */
- public String getTimerClass()
- {
- return timerClass;
- }
-
- /**
- * Copies Checkpoint configuration
- *
- * @param config - checkpoint configuration
- */
- public void setChConfig(CheckpointConfig config)
- {
- chConfig = config;
- }
-
- /**
- * Copies CPE startup mode
- *
- * @param string - startup mode
- */
- public void setDeployAs(String aCpeDeployMode)
- {
- deployAs = aCpeDeployMode;
- }
-
- /**
- * Copies number of entities to process
- *
- * @param aTotalCount - total number of entities to process
- */
- public void setNumToProcess(long aTotalCount)
- {
- numToProcess = aTotalCount;
- }
-
- /**
- * Copies ind of the first entity to start reading
- *
- * @param string - id of entity
- */
- public void setStartWith(String aStartEntityId)
- {
- startWith = aStartEntityId;
- }
-
- /**
- * Copies a name of the custom {@link UimaTimer} class
- *
- * @param string - timer class
- */
- public void setTimerClass(String aTimerClass)
- {
- timerClass = aTimerClass;
- }
- public long getInputQueueMaxMemoryThreshold()
- {
- return inputQueueMaxMemoryThreshold;
- }
- public void setInputQueueMaxMemoryThreshold(long aInputQueueMaxMemoryThreshold)
- {
- inputQueueMaxMemoryThreshold = aInputQueueMaxMemoryThreshold;
- }
- public long getMaxTimeToWait()
- {
- return maxTimeToWait;
- }
- public void setMaxTimeToWait(long aMaxTimeToWait)
- {
- maxTimeToWait = aMaxTimeToWait;
- }
- public String getOutputQueueClass()
- {
- return outputQueueClass;
- }
- public void setOutputQueueClass(String aOutputQueueClass)
- {
- outputQueueClass = aOutputQueueClass;
- }
+public class CPEConfig {
+ private CheckpointConfig chConfig = null;
+ private long numToProcess = 0;
+
+ private String deployAs = "";
+
+ private String timerClass = null;
+
+ private String startWith = "";
+
+ private long inputQueueMaxMemoryThreshold = 0;
+
+ private long maxTimeToWait = -1;
+
+ private String outputQueueClass = null;
+
+ /**
+ * Returns CPE checkpoint configuration
+ *
+ * @return CheckpointConfig object
+ */
+ public CheckpointConfig getChConfig() {
+ return chConfig;
+ }
+
+ /**
+ * Returns CPE statup mode as defined in the CPE descriptor
+ *
+ * @return - statup mode
+ */
+ public String getDeployAs() {
+ return deployAs;
+ }
+
+ /**
+ * Returns number of entities to process by the CPE.
+ *
+ * @return - number of entities to process
+ */
+ public long getNumToProcess() {
+ return numToProcess;
+ }
+
+ /**
+ * Returns an id of the first entity the {@link CollectionReader} will be told to read
+ *
+ * @return - id of the first entity to read
+ */
+ public String getStartWith() {
+ return startWith;
+ }
+
+ /**
+ * Returns the name of custom {@link UimaTimer} class.
+ *
+ * @return - class as String
+ */
+ public String getTimerClass() {
+ return timerClass;
+ }
+
+ /**
+ * Copies Checkpoint configuration
+ *
+ * @param config -
+ * checkpoint configuration
+ */
+ public void setChConfig(CheckpointConfig config) {
+ chConfig = config;
+ }
+
+ /**
+ * Copies CPE startup mode
+ *
+ * @param string -
+ * startup mode
+ */
+ public void setDeployAs(String aCpeDeployMode) {
+ deployAs = aCpeDeployMode;
+ }
+
+ /**
+ * Copies number of entities to process
+ *
+ * @param aTotalCount -
+ * total number of entities to process
+ */
+ public void setNumToProcess(long aTotalCount) {
+ numToProcess = aTotalCount;
+ }
+
+ /**
+ * Copies ind of the first entity to start reading
+ *
+ * @param string -
+ * id of entity
+ */
+ public void setStartWith(String aStartEntityId) {
+ startWith = aStartEntityId;
+ }
+
+ /**
+ * Copies a name of the custom {@link UimaTimer} class
+ *
+ * @param string -
+ * timer class
+ */
+ public void setTimerClass(String aTimerClass) {
+ timerClass = aTimerClass;
+ }
+
+ public long getInputQueueMaxMemoryThreshold() {
+ return inputQueueMaxMemoryThreshold;
+ }
+
+ public void setInputQueueMaxMemoryThreshold(long aInputQueueMaxMemoryThreshold) {
+ inputQueueMaxMemoryThreshold = aInputQueueMaxMemoryThreshold;
+ }
+
+ public long getMaxTimeToWait() {
+ return maxTimeToWait;
+ }
+
+ public void setMaxTimeToWait(long aMaxTimeToWait) {
+ maxTimeToWait = aMaxTimeToWait;
+ }
+
+ public String getOutputQueueClass() {
+ return outputQueueClass;
+ }
+
+ public void setOutputQueueClass(String aOutputQueueClass) {
+ outputQueueClass = aOutputQueueClass;
+ }
}