| /* |
| * 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.aae.controller; |
| |
| import java.io.ByteArrayInputStream; |
| import java.io.ByteArrayOutputStream; |
| import java.io.IOException; |
| import java.io.ObjectInputStream; |
| import java.io.ObjectOutputStream; |
| import java.lang.reflect.InvocationTargetException; |
| import java.lang.reflect.Method; |
| import java.util.ArrayList; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Timer; |
| import java.util.TimerTask; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.Semaphore; |
| |
| import org.apache.uima.UIMAFramework; |
| import org.apache.uima.UimaContext; |
| import org.apache.uima.UimaContextAdmin; |
| import org.apache.uima.aae.AsynchAECasManager; |
| import org.apache.uima.aae.InProcessCache; |
| import org.apache.uima.aae.InProcessCache.CacheEntry; |
| import org.apache.uima.aae.UIMAEE_Constants; |
| import org.apache.uima.aae.UimaClassFactory; |
| import org.apache.uima.aae.controller.LocalCache.CasStateEntry; |
| import org.apache.uima.aae.delegate.Delegate; |
| import org.apache.uima.aae.error.AsynchAEException; |
| import org.apache.uima.aae.error.ErrorContext; |
| import org.apache.uima.aae.error.ErrorHandler; |
| import org.apache.uima.aae.jmx.JmxManagement; |
| import org.apache.uima.aae.jmx.PrimitiveServiceInfo; |
| import org.apache.uima.aae.jmx.ServicePerformance; |
| import org.apache.uima.aae.message.AsynchAEMessage; |
| import org.apache.uima.aae.message.MessageContext; |
| import org.apache.uima.aae.monitor.Monitor; |
| import org.apache.uima.aae.monitor.statistics.AnalysisEnginePerformanceMetrics; |
| import org.apache.uima.aae.spi.transport.UimaMessage; |
| import org.apache.uima.aae.spi.transport.UimaTransport; |
| import org.apache.uima.analysis_engine.AnalysisEngine; |
| import org.apache.uima.analysis_engine.AnalysisEngineDescription; |
| import org.apache.uima.analysis_engine.AnalysisEngineManagement; |
| import org.apache.uima.analysis_engine.CasIterator; |
| import org.apache.uima.analysis_engine.metadata.AnalysisEngineMetaData; |
| import org.apache.uima.cas.CAS; |
| import org.apache.uima.cas.impl.CASImpl; |
| import org.apache.uima.collection.CollectionReaderDescription; |
| import org.apache.uima.impl.RootUimaContext_impl; |
| import org.apache.uima.resource.Resource; |
| import org.apache.uima.resource.ResourceInitializationException; |
| import org.apache.uima.resource.ResourceSpecifier; |
| import org.apache.uima.resource.metadata.ConfigurationParameter; |
| import org.apache.uima.resource.metadata.ProcessingResourceMetaData; |
| import org.apache.uima.resource.metadata.impl.ConfigurationParameter_impl; |
| import org.apache.uima.util.Level; |
| |
| public class PrimitiveAnalysisEngineController_impl extends BaseAnalysisEngineController implements |
| PrimitiveAnalysisEngineController { |
| private static final Class CLASS_NAME = PrimitiveAnalysisEngineController_impl.class; |
| private static final String DUMP_HEAP_THRESHOLD = "dumpHeapThreshold"; |
| |
| private volatile boolean casPoolInited = false; |
| // Stores AE metadata |
| private AnalysisEngineMetaData analysisEngineMetadata; |
| |
| // Number of AE instances |
| private int analysisEnginePoolSize; |
| |
| // Mutex |
| protected Object notifyObj = new Object(); |
| |
| // Temp list holding instances of AE |
| private List aeList = new ArrayList(); |
| |
| // Stores service info for JMX |
| private PrimitiveServiceInfo serviceInfo = null; |
| |
| // Pool containing instances of AE. The default implementation provides Thread affinity |
| // meaning each thread executes the same AE instance. |
| protected AnalysisEngineInstancePool aeInstancePool = null; |
| |
| // Create a shared semaphore to serialize creation of AE instances. |
| // There is a single instance of this semaphore per JVM and it |
| // guards uima core code that is not thread safe. |
| private static Semaphore sharedInitSemaphore = new Semaphore(1); |
| |
| static private Object threadDumpMonitor = new Object(); |
| static private Long lastDump = Long.valueOf(0); |
| |
| // 6 args |
| public PrimitiveAnalysisEngineController_impl(String anEndpointName, |
| String anAnalysisEngineDescriptor, AsynchAECasManager aCasManager, |
| InProcessCache anInProcessCache, int aWorkQueueSize, int anAnalysisEnginePoolSize) |
| throws Exception { |
| this(null, anEndpointName, anAnalysisEngineDescriptor, aCasManager, anInProcessCache, |
| aWorkQueueSize, anAnalysisEnginePoolSize, 0); |
| } |
| |
| // 7 args - adds parentController at beginning |
| public PrimitiveAnalysisEngineController_impl(AnalysisEngineController aParentController, |
| String anEndpointName, String anAnalysisEngineDescriptor, AsynchAECasManager aCasManager, |
| InProcessCache anInProcessCache, int aWorkQueueSize, int anAnalysisEnginePoolSize) |
| throws Exception { |
| this(aParentController, anEndpointName, anAnalysisEngineDescriptor, aCasManager, |
| anInProcessCache, aWorkQueueSize, anAnalysisEnginePoolSize, 0); |
| } |
| |
| // 8 args - adds componentCasPoolSize at end |
| public PrimitiveAnalysisEngineController_impl(AnalysisEngineController aParentController, |
| String anEndpointName, String anAnalysisEngineDescriptor, AsynchAECasManager aCasManager, |
| InProcessCache anInProcessCache, int aWorkQueueSize, int anAnalysisEnginePoolSize, |
| int aComponentCasPoolSize) throws Exception { |
| this(aParentController, anEndpointName, anAnalysisEngineDescriptor, aCasManager, |
| anInProcessCache, aWorkQueueSize, anAnalysisEnginePoolSize, aComponentCasPoolSize, null, false); |
| } |
| |
| // 9 args - adds initialCasHeapSize |
| // ************* USED BY DD2SPRING ************* |
| public PrimitiveAnalysisEngineController_impl(AnalysisEngineController aParentController, |
| String anEndpointName, String anAnalysisEngineDescriptor, AsynchAECasManager aCasManager, |
| InProcessCache anInProcessCache, int aWorkQueueSize, int anAnalysisEnginePoolSize, |
| int aComponentCasPoolSize, long anInitialCasHeapSize, boolean disableJCasCache) throws Exception { |
| this(aParentController, anEndpointName, anAnalysisEngineDescriptor, aCasManager, |
| anInProcessCache, aWorkQueueSize, anAnalysisEnginePoolSize, aComponentCasPoolSize, |
| anInitialCasHeapSize, null, disableJCasCache); |
| } |
| public PrimitiveAnalysisEngineController_impl(AnalysisEngineController aParentController, |
| String anEndpointName, String anAnalysisEngineDescriptor, AsynchAECasManager aCasManager, |
| InProcessCache anInProcessCache, int aWorkQueueSize, int anAnalysisEnginePoolSize, |
| int aComponentCasPoolSize, long anInitialCasHeapSize) throws Exception { |
| this(aParentController, anEndpointName, anAnalysisEngineDescriptor, aCasManager, |
| anInProcessCache, aWorkQueueSize, anAnalysisEnginePoolSize, aComponentCasPoolSize, |
| anInitialCasHeapSize, null, false); |
| } |
| // 9 args - repl initialCasHeapSize with jmxManagement |
| public PrimitiveAnalysisEngineController_impl(AnalysisEngineController aParentController, |
| String anEndpointName, String anAnalysisEngineDescriptor, AsynchAECasManager aCasManager, |
| InProcessCache anInProcessCache, int aWorkQueueSize, int anAnalysisEnginePoolSize, |
| int aComponentCasPoolSize, JmxManagement aJmxManagement, boolean disableJCasCache) throws Exception { |
| this(aParentController, anEndpointName, anAnalysisEngineDescriptor, aCasManager, |
| anInProcessCache, aWorkQueueSize, anAnalysisEnginePoolSize, aComponentCasPoolSize, 0, |
| aJmxManagement, disableJCasCache); |
| } |
| |
| // 10 args - adds initialCasHeapSize back |
| public PrimitiveAnalysisEngineController_impl(AnalysisEngineController aParentController, |
| String anEndpointName, String anAnalysisEngineDescriptor, AsynchAECasManager aCasManager, |
| InProcessCache anInProcessCache, int aWorkQueueSize, int anAnalysisEnginePoolSize, |
| int aComponentCasPoolSize, long anInitialCasHeapSize, JmxManagement aJmxManagement, |
| boolean disableJCasCache) |
| throws Exception { |
| super(aParentController, aComponentCasPoolSize, anInitialCasHeapSize, anEndpointName, |
| anAnalysisEngineDescriptor, aCasManager, anInProcessCache, null, aJmxManagement, disableJCasCache); |
| analysisEnginePoolSize = anAnalysisEnginePoolSize; |
| } |
| |
| // 8 args - drops componentCasPoolSize and initialCasHeapSize |
| public PrimitiveAnalysisEngineController_impl(AnalysisEngineController aParentController, |
| String anEndpointName, String anAnalysisEngineDescriptor, AsynchAECasManager aCasManager, |
| InProcessCache anInProcessCache, int aWorkQueueSize, int anAnalysisEnginePoolSize, |
| JmxManagement aJmxManagement) throws Exception { |
| this(aParentController, anEndpointName, anAnalysisEngineDescriptor, aCasManager, |
| anInProcessCache, aWorkQueueSize, anAnalysisEnginePoolSize, 0, aJmxManagement, false ); |
| } |
| |
| public int getAEInstanceCount() { |
| return analysisEnginePoolSize; |
| } |
| public static Object copy(Object orig) { |
| Object obj = null; |
| try { |
| // Write the object out to a byte array |
| ByteArrayOutputStream bos = new ByteArrayOutputStream(); |
| ObjectOutputStream out = new ObjectOutputStream(bos); |
| out.writeObject(orig); |
| out.flush(); |
| out.close(); |
| |
| // Make an input stream from the byte array and read |
| // a copy of the object back in. |
| ObjectInputStream in = new ObjectInputStream( |
| new ByteArrayInputStream(bos.toByteArray())); |
| obj = in.readObject(); |
| } |
| catch(IOException e) { |
| e.printStackTrace(); |
| } |
| catch(ClassNotFoundException cnfe) { |
| cnfe.printStackTrace(); |
| } |
| return obj; |
| } |
| public void dumpContext(UimaContextAdmin ctx) { |
| if ( !(ctx instanceof RootUimaContext_impl) ) { |
| dumpContext(ctx.getRootContext()); |
| } else { |
| Map<String, AnalysisEngineManagement> m = ctx.getManagementInterface().getComponents(); |
| for( Entry<String, AnalysisEngineManagement> e : m.entrySet()) { |
| System.out.println(">>>>>>>> "+e.getKey()+" "+e.getValue().getUniqueMBeanName()); |
| } |
| } |
| } |
| public void initializeAnalysisEngine() throws ResourceInitializationException { |
| ResourceSpecifier rSpecifier = null; |
| |
| try { |
| // Acquire single-permit semaphore to serialize instantiation of AEs. |
| // This is done to control access to non-thread safe structures in the |
| // core. The sharedInitSemaphore is a static and is shared by all instances |
| // of this class. |
| sharedInitSemaphore.acquire(); |
| // Parse the descriptor in the calling thread. |
| rSpecifier = UimaClassFactory.produceResourceSpecifier(super.aeDescriptor); |
| UimaContextAdmin uctx = null; |
| if ( parentController != null ) { |
| int scaleout = 1; |
| if (parentController instanceof AggregateAnalysisEngineController) { |
| String key = ((AggregateAnalysisEngineController) parentController) |
| .lookUpDelegateKey(endpointName); |
| if (key == null) { |
| if (((AggregateAnalysisEngineController) parentController).isDelegateKeyValid(endpointName)) { |
| key = endpointName; |
| } |
| } |
| if (key == null) { |
| throw new AsynchAEException(getName() + "-Unable to look up delegate " |
| + endpointName + " in internal map"); |
| } |
| Delegate d = ((AggregateAnalysisEngineController) parentController) |
| .lookupDelegate(key); |
| scaleout = d.getEndpoint().getConcurrentRequestConsumers(); |
| } |
| if ( scaleout > 1) { |
| uctx = (UimaContextAdmin)parentController.getChildUimaContext(endpointName); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| dumpContext(uctx); |
| } |
| paramsMap.remove(Resource.PARAM_UIMA_CONTEXT); |
| paramsMap.put(Resource.PARAM_UIMA_CONTEXT, uctx); |
| } else { |
| uctx = (UimaContextAdmin)paramsMap.get(Resource.PARAM_UIMA_CONTEXT); |
| } |
| } |
| AnalysisEngine ae = UIMAFramework.produceAnalysisEngine(rSpecifier, paramsMap); |
| |
| super.addUimaObject(ae.getManagementInterface().getUniqueMBeanName()); |
| // Call to produceAnalysisEngine() may take a long time to complete. While this |
| // method was executing, the service may have been stopped. Before continuing |
| // check if the service has been stopped. If so, destroy AE instance and return. |
| if ( isStopped() ) { |
| ae.destroy(); |
| return; |
| } |
| if (aeInstancePool == null) { |
| aeInstancePool = new AnalysisEngineInstancePoolWithThreadAffinity();//analysisEnginePoolSize); |
| } |
| if (analysisEngineMetadata == null) { |
| analysisEngineMetadata = ae.getAnalysisEngineMetaData(); |
| } |
| // Check if OperationalProperties allow replication of this AE. Throw exception if |
| // the deployment descriptor says to scale the service *but* the AE descriptor's |
| // OperationalProperties disallow it. |
| if ( !analysisEngineMetadata.getOperationalProperties().isMultipleDeploymentAllowed() && |
| aeInstancePool.size() >= 1 ) { |
| throw new ResourceInitializationException( UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_multiple_deployment_not_allowed__WARNING", new Object[] {this.getComponentName(), ae.getMetaData().getName()}); |
| } |
| aeInstancePool.checkin(ae); |
| |
| if (!isStopped() && !casPoolInited) { |
| casPoolInited = true; |
| |
| if (errorHandlerChain == null) { |
| super.plugInDefaultErrorHandlerChain(); |
| } |
| |
| getMonitor().setThresholds(getErrorHandlerChain().getThresholds()); |
| // Initialize Cas Manager |
| if (getCasManagerWrapper() != null) { |
| try { |
| // Below should always be true. In spring context file AsynchAECasManager_impl |
| // is instantiated and setCasPoolSize() method is called which sets the |
| // initialized state = true. isInitialized() returning true just means that |
| // setCasPoolSize() was called. |
| if (getCasManagerWrapper().isInitialized()) { |
| getCasManagerWrapper().addMetadata(getAnalysisEngineMetadata()); |
| if (isTopLevelComponent()) { |
| getCasManagerWrapper().initialize("PrimitiveAEService"); |
| CAS cas = getCasManagerWrapper().getNewCas("PrimitiveAEService"); |
| cas.release(); |
| } |
| } |
| } catch( Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "postInitialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_service_exception_WARNING", getComponentName()); |
| |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "postInitialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", e); |
| } |
| throw new AsynchAEException(e); |
| } |
| } |
| } |
| |
| |
| String warmUpDataPath=null; |
| // start warm up engine which will prime pipeline analytics |
| if ( isTopLevelComponent() && (warmUpDataPath = System.getProperty("WarmUpDataPath")) != null ) { |
| CountDownLatch warmUpLatch = new CountDownLatch(1); |
| warmUp(warmUpDataPath, warmUpLatch); |
| warmUpLatch.await(); |
| } |
| |
| if (aeInstancePool.size() == analysisEnginePoolSize) { |
| try { |
| postInitialize(); |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "initializeAnalysisEngine", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_service_exception_WARNING", getComponentName()); |
| |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "initializeAnalysisEngine", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", e); |
| } |
| throw new ResourceInitializationException(e); |
| } |
| } |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "initializeAnalysisEngine", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_service_exception_WARNING", getComponentName()); |
| |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "initializeAnalysisEngine", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", e); |
| } |
| super.notifyListenersWithInitializationStatus(e); |
| if (isTopLevelComponent()) { |
| super.notifyListenersWithInitializationStatus(e); |
| } else { |
| // get the top level controller to notify |
| AnalysisEngineController controller = this.getParentController(); |
| while (!controller.isTopLevelComponent()) { |
| controller = controller.getParentController(); |
| } |
| getParentController().notifyListenersWithInitializationStatus(e); |
| } |
| |
| throw new ResourceInitializationException(e); |
| } finally { |
| // Release the shared semaphore so that another instance of this class can instantiate |
| // an Analysis Engine. |
| sharedInitSemaphore.release(); |
| } |
| |
| } |
| |
| public boolean threadAssignedToAE() { |
| if (aeInstancePool == null) { |
| return false; |
| } |
| |
| return aeInstancePool.exists(); |
| } |
| |
| public void initialize() throws AsynchAEException { |
| } |
| |
| /** |
| * This method is called after all AE instances initialize. It is called once. It initializes |
| * service Cas Pool, notifies the deployer that initialization completed and finally loweres a |
| * semaphore allowing messages to be processed. |
| * |
| * @throws AsynchAEException |
| */ |
| private void postInitialize() throws AsynchAEException { |
| try { |
| if (errorHandlerChain == null) { |
| super.plugInDefaultErrorHandlerChain(); |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.CONFIG)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.CONFIG, getClass().getName(), "initialize", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_primitive_ctrl_init_info__CONFIG", new Object[] { analysisEnginePoolSize }); |
| } |
| |
| if (serviceInfo == null) { |
| serviceInfo = new PrimitiveServiceInfo(isCasMultiplier(), this); |
| } |
| serviceInfo.setServiceKey(delegateKey); |
| serviceInfo.setAnalysisEngineInstanceCount(analysisEnginePoolSize); |
| |
| if (!isStopped()) { |
| /* |
| getMonitor().setThresholds(getErrorHandlerChain().getThresholds()); |
| // Initialize Cas Manager |
| if (getCasManagerWrapper() != null) { |
| try { |
| // Below should always be true. In spring context file AsynchAECasManager_impl |
| // is instantiated and setCasPoolSize() method is called which sets the |
| // initialized state = true. isInitialized() returning true just means that |
| // setCasPoolSize() was called. |
| if (getCasManagerWrapper().isInitialized()) { |
| getCasManagerWrapper().addMetadata(getAnalysisEngineMetadata()); |
| if (isTopLevelComponent()) { |
| getCasManagerWrapper().initialize("PrimitiveAEService"); |
| CAS cas = getCasManagerWrapper().getNewCas("PrimitiveAEService"); |
| cas.release(); |
| } |
| } |
| */ |
| if (getCasManagerWrapper() != null) { |
| try { |
| if (isTopLevelComponent()) { |
| // add delay to allow controller listener to plug itself in |
| synchronized(this) { |
| try { |
| this.wait(100); |
| } catch(Exception exx) {} |
| } |
| |
| super.notifyListenersWithInitializationStatus(null); |
| } |
| |
| // All internal components of this Primitive have been initialized. Open the latch |
| // so that this service can start processing requests. |
| latch.openLatch(getName(), isTopLevelComponent(), true); |
| |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "postInitialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_service_exception_WARNING", getComponentName()); |
| |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "postInitialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", e); |
| } |
| throw new AsynchAEException(e); |
| } |
| } else { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.CONFIG)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.CONFIG, getClass().getName(), |
| "postInitialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_cas_manager_wrapper_notdefined__CONFIG", new Object[] {}); |
| } |
| } |
| if (!isStopped()){ |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "postInitialize", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_initialized_controller__INFO", |
| new Object[] { getComponentName() }); |
| } |
| super.serviceInitialized = true; |
| } |
| } |
| } catch (AsynchAEException e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "postInitialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_service_exception_WARNING", getComponentName()); |
| |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "postInitialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", e); |
| } |
| throw e; |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "postInitialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_service_exception_WARNING", getComponentName()); |
| |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "postInitialize", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", e); |
| } |
| throw new AsynchAEException(e); |
| } |
| |
| } |
| |
| /** |
| * Forces initialization of a Cas Pool if this is a Cas Multiplier delegate collocated with an |
| * aggregate. The parent aggregate calls this method when all type systems have been merged. |
| */ |
| public synchronized void onInitialize() { |
| // Component's Cas Pool is registered lazily, when the process() is called for |
| // the first time. For monitoring purposes, we need the comoponent's Cas Pool |
| // MBeans to register during initialization of the service. For a Cas Multiplier |
| // force creation of the Cas Pool and registration of a Cas Pool with the JMX Server. |
| // Just get the CAS and release it back to the component's Cas Pool. |
| if (isCasMultiplier() && !isTopLevelComponent() ) { |
| boolean isUimaAggregate = false; |
| if ( !(resourceSpecifier instanceof CollectionReaderDescription) ) { |
| // determine if this AE is a UIMA aggregate |
| isUimaAggregate = ((AnalysisEngineDescription) resourceSpecifier).isPrimitive() == false ? true : false; |
| } |
| if ( !isUimaAggregate ) { // !uima core aggregate CM |
| CAS cas = (CAS) getUimaContext().getEmptyCas(CAS.class); |
| cas.release(); |
| } |
| } |
| } |
| |
| /** |
| * |
| */ |
| public void collectionProcessComplete(Endpoint anEndpoint)// throws AsynchAEException |
| { |
| AnalysisEngine ae = null; |
| long start = super.getCpuTime(); |
| localCache.dumpContents(); |
| try { |
| ae = aeInstancePool.checkout(); |
| if (ae != null) { |
| ae.collectionProcessComplete(); |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, getClass().getName(), |
| "collectionProcessComplete", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_cpc_all_cases_processed__FINEST", new Object[] { getComponentName() }); |
| } |
| getServicePerformance().incrementAnalysisTime(super.getCpuTime() - start); |
| if (!anEndpoint.isRemote()) { |
| UimaTransport transport = getTransport(anEndpoint.getEndpoint()); |
| UimaMessage message = transport.produceMessage(AsynchAEMessage.CollectionProcessComplete, |
| AsynchAEMessage.Response, getName()); |
| // Send CPC completion reply back to the client. Use internal (non-jms) transport |
| transport.getUimaMessageDispatcher(anEndpoint.getEndpoint()).dispatch(message); |
| } else { |
| getOutputChannel().sendReply(AsynchAEMessage.CollectionProcessComplete, anEndpoint, null, false); |
| } |
| |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, getClass().getName(), |
| "collectionProcessComplete", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_cpc_completed__FINE", new Object[] { getComponentName() }); |
| } |
| |
| } catch (Exception e) { |
| ErrorContext errorContext = new ErrorContext(); |
| errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.CollectionProcessComplete); |
| errorContext.add(AsynchAEMessage.Endpoint, anEndpoint); |
| getErrorHandlerChain().handle(e, errorContext, this); |
| } finally { |
| clearStats(); |
| if (ae != null) { |
| try { |
| aeInstancePool.checkin(ae); |
| } catch (Exception ex) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "collectionProcessComplete", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_service_exception_WARNING", getComponentName()); |
| |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "collectionProcessComplete", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_unable_to_check_ae_back_to_pool__WARNING", |
| new Object[] { getComponentName(), ex }); |
| } |
| } |
| } |
| } |
| } |
| /** |
| * Conditionally creates and starts a timer thread that will time a task/method. |
| * The timer is only created if the System property -DheapDumpThreshold=<x> is set. |
| * The x is the max number of seconds the task/method is allowed before the timer pops. |
| * |
| * @return instance of HeapDumpTimer or null |
| */ |
| private StackDumpTimer ifEnabledStartHeapDumpTimer() { |
| try { |
| if ( System.getProperty(DUMP_HEAP_THRESHOLD) != null) { |
| int thresholdInSeconds = Integer.parseInt(System.getProperty(DUMP_HEAP_THRESHOLD)); |
| // Creates and returns an instance of a timer. The timer is auto started in a |
| // constructor and will expire in a given number of seconds |
| return new StackDumpTimer(thresholdInSeconds); |
| } |
| } catch( NumberFormatException e) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "ifEnabledStartHeapDumpTimer", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", e); |
| } |
| return null; |
| } |
| |
| private void getLeafManagementObjects(AnalysisEngineManagement aem, List<AnalysisEnginePerformanceMetrics> result) { |
| getLeafManagementObjects(aem, result, ""); |
| } |
| /** |
| * Recursively |
| * |
| * @param aem |
| * @param result |
| * @param uimaFullyQualifiedAEContext |
| */ |
| private void getLeafManagementObjects(AnalysisEngineManagement aem, List<AnalysisEnginePerformanceMetrics> result, String uimaFullyQualifiedAEContext) { |
| if (aem.getComponents().isEmpty() && Thread.currentThread().getId() == aem.getThreadId()) { |
| // is this primitive AE delegate in an aggregate. If so the mbean unique name will have "p0=" string. An examples mbean |
| // name looks like this: |
| // org.apache.uima:type=ee.jms.services,s=Top Level Aggregate TAE Uima EE Service,p0=Top Level Aggregate TAE Components,p1=SecondLevelAggregateCM Components,p2=ThirdLevelAggregateCM Components,name=Multiplier1 |
| if ( aem.getUniqueMBeanName().indexOf("p0=") > -1 ) { |
| // check id the parent aggregate has been scaled up by looking at the last char in its name. If it is a number strip it from the name |
| if ( Character.isDigit(uimaFullyQualifiedAEContext.charAt(uimaFullyQualifiedAEContext.length()-1) ) && |
| uimaFullyQualifiedAEContext.lastIndexOf(" ") > -1 ) { |
| String indx = uimaFullyQualifiedAEContext.substring(uimaFullyQualifiedAEContext.lastIndexOf(" ")); |
| if ( indx != null ) { |
| int value = -1; |
| try { |
| value = Integer.parseInt(indx.trim()); |
| // Prepend "X Components" to the unique name with X stripped. |
| uimaFullyQualifiedAEContext = value+" Components "+ uimaFullyQualifiedAEContext.substring(0,uimaFullyQualifiedAEContext.lastIndexOf(" ")); |
| } catch( NumberFormatException ex) { |
| |
| } |
| } |
| } |
| } |
| AnalysisEnginePerformanceMetrics m = deepCopyMetrics(aem, uimaFullyQualifiedAEContext); |
| result.add(m); |
| } else { |
| for (AnalysisEngineManagement child : (Iterable<AnalysisEngineManagement>) aem.getComponents().values()) { |
| getLeafManagementObjects(child, result, produceUniqueName(aem)); |
| } |
| } |
| } |
| |
| |
| public void destroyAE() { |
| try { |
| if ( aeInstancePool != null ) { |
| AnalysisEngine ae = aeInstancePool.checkout(); |
| if ( ae != null ) { |
| ae.destroy(); |
| } |
| } |
| } catch( AsynchAEException e) { |
| // No-op. AE instance not found. Most likely, the UIMA AS service is in shutdown state |
| } catch( Exception e) { |
| e.printStackTrace(); |
| } |
| } |
| |
| private String produceUniqueName(AnalysisEngineManagement aem) { |
| String[] parts = aem.getUniqueMBeanName().split(","); |
| StringBuffer sb = new StringBuffer(); |
| for( String part : parts) { |
| int pos; |
| if ( (pos = part.indexOf("=") )> -1 && part.startsWith("p")) { |
| String n = part.substring(pos+1, part.indexOf(" Components")); |
| if ( part.startsWith("p0=") && n.indexOf(" ") > -1) { |
| String indx = n.substring(n.lastIndexOf(" ")); |
| if ( indx != null ) { |
| int instanceNumber=-1; |
| try { |
| instanceNumber = Integer.parseInt(indx.trim()); |
| sb.append(instanceNumber).append(" Components "); |
| n = n.substring(0,n.lastIndexOf(" ")); |
| } catch(NumberFormatException nfe) { |
| } |
| } |
| } |
| sb.append("/").append(n.trim()); |
| } else if ( part.trim().startsWith("name=")) { |
| sb.append("/").append(part.substring(part.trim().indexOf("=")+1)); |
| } |
| } |
| return sb.toString(); |
| } |
| |
| private AnalysisEnginePerformanceMetrics deepCopyMetrics(AnalysisEngineManagement aem, String uimaFullyQualifiedAEContext) { |
| String index = ""; |
| // Create a unique name with each AE name is separated with "/". Prepend "X Components" where |
| // X is a instance number of a scaled AE. Also, strip the X from the AE name. The instance number |
| // is added to each scaled up component during initialization of the uima-as. We need to prepend |
| // "X Components" to allow DUCC JD to parse the unique name correctly ( basically for backwards |
| // compatibility. |
| int pos = aem.getUniqueMBeanName().lastIndexOf("name="); |
| if ( pos > -1 ) { |
| // get the name of the component. In case of nested component this will be the KEY from AE descriptor |
| String tmp = aem.getUniqueMBeanName().substring(pos+5); |
| // in case this is the top level AE, check if it has been scaled up by extracting its instance number.For example, |
| // NoOpAnnotator 2. |
| int last = tmp.lastIndexOf(" "); |
| if ( last > -1 ) { |
| // extract instance number |
| index = tmp.substring(last); |
| |
| try { |
| // check if the instance number is a number. If not silently handle the exception. |
| Integer.parseInt(index.trim()); |
| // strip the instance number from the AE name |
| uimaFullyQualifiedAEContext = uimaFullyQualifiedAEContext.substring(0, last+1); |
| } catch( NumberFormatException nfe) { |
| |
| } |
| } else { |
| |
| if ( !uimaFullyQualifiedAEContext.endsWith(tmp)) { |
| uimaFullyQualifiedAEContext += "/"+tmp; |
| } |
| } |
| } |
| // Primitive AE will not have "X Components" prefix, but it is required |
| // by the DUCC JD to be there. Prepend it to the unique name. |
| if ( uimaFullyQualifiedAEContext.indexOf(" Components ") == -1) { |
| uimaFullyQualifiedAEContext = index + " Components "+uimaFullyQualifiedAEContext; |
| } |
| return new AnalysisEnginePerformanceMetrics(aem.getName(), |
| uimaFullyQualifiedAEContext, |
| aem.getAnalysisTime(), |
| aem.getNumberOfCASesProcessed()); |
| |
| } |
| |
| /** |
| * This is called when a Stop request is received from a client. Add the provided Cas id to the |
| * list of aborted CASes. The process() method checks this list to determine if it should continue |
| * generating children. |
| * |
| * @param aCasReferenceId |
| * - Id of an input CAS. The client wants to stop generation of child CASes from this |
| * CAS. |
| * @param anEndpoint endpoint |
| */ |
| |
| public void process(CAS aCAS, String aCasReferenceId, Endpoint anEndpoint) { |
| |
| if (stopped) { |
| return; |
| } |
| |
| List<AnalysisEnginePerformanceMetrics> beforeAnalysisManagementObjects = new ArrayList<AnalysisEnginePerformanceMetrics>(); |
| List<AnalysisEnginePerformanceMetrics> afterAnalysisManagementObjects = new ArrayList<AnalysisEnginePerformanceMetrics>(); |
| CasStateEntry parentCasStateEntry = null; |
| // If enabled, keep a reference to a timer which |
| // when it expires, will cause a JVM to dump a stack |
| StackDumpTimer stackDumpTimer = null; |
| try { |
| parentCasStateEntry = getLocalCache().lookupEntry(aCasReferenceId); |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "process", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_service_exception_WARNING", getComponentName()); |
| |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "process", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", e); |
| } |
| return; |
| } |
| |
| long totalProcessTime = 0; // stored total time spent producing ALL CASes |
| |
| boolean inputCASReturned = false; |
| boolean processingFailed = false; |
| // This is a primitive controller. No more processing is to be done on the Cas. Mark the |
| // destination as final and return CAS in reply. |
| anEndpoint.setFinal(true); |
| AnalysisEngine ae = null; |
| boolean clientUnreachable = false; |
| try { |
| // Checkout an instance of AE from the pool |
| ae = aeInstancePool.checkout(); |
| // Get input CAS entry from the InProcess cache |
| long time = super.getCpuTime(); |
| // Start the heap dump timer. This timer is only started if explicitly enabled |
| // via System property: -DheapDumpThreshold=<x> where is number of seconds the |
| // method is allowed to complete. If the method is not complete in allowed window |
| // the heap and stack trace dump of all threads will be produced. |
| stackDumpTimer = ifEnabledStartHeapDumpTimer(); |
| |
| |
| AnalysisEngineManagement rootAem = ae.getManagementInterface(); |
| if ( rootAem.getComponents().size() > 0 ) { |
| getLeafManagementObjects(rootAem, beforeAnalysisManagementObjects); |
| } else { |
| String path=produceUniqueName(rootAem); |
| beforeAnalysisManagementObjects.add(deepCopyMetrics(rootAem, path)); |
| } |
| |
| CasIterator casIterator = ae.processAndOutputNewCASes(aCAS); |
| if ( stackDumpTimer != null ) { |
| stackDumpTimer.cancel(); |
| stackDumpTimer = null; // nullify timer instance so that we dont have to worry about |
| // it in case an exception happens below |
| } |
| |
| // Store how long it took to call processAndOutputNewCASes() |
| totalProcessTime = (super.getCpuTime() - time); |
| long sequence = 1; |
| long hasNextTime = 0; // stores time in hasNext() |
| long getNextTime = 0; // stores time in next(); |
| boolean moreCASesToProcess = true; |
| boolean casAbortedDueToExternalRequest = false; |
| while (moreCASesToProcess) { |
| long timeToProcessCAS = 0; // stores time in hasNext() and next() for each CAS |
| hasNextTime = super.getCpuTime(); |
| // Start the heap dump timer. This timer is only started if explicitly enabled |
| // via System property: -DheapDumpThreshold=<x> where x is a number of seconds the |
| // method is allowed to complete. If the method is not complete in allowed window |
| // the heap and stack trace dump of all threads will be produced. |
| stackDumpTimer = ifEnabledStartHeapDumpTimer(); |
| if (!casIterator.hasNext()) { |
| moreCASesToProcess = false; |
| // Measure how long it took to call hasNext() |
| timeToProcessCAS = (super.getCpuTime() - hasNextTime); |
| totalProcessTime += timeToProcessCAS; |
| if ( stackDumpTimer != null ) { |
| stackDumpTimer.cancel(); |
| stackDumpTimer = null; // nullify timer instance so that we dont have to worry about |
| // it in case an exception happens below |
| } |
| break; // from while |
| } |
| if ( stackDumpTimer != null ) { |
| stackDumpTimer.cancel(); |
| stackDumpTimer = null; // nullify timer instance so that we dont have to worry about |
| // it in case an exception happens below |
| } |
| // Measure how long it took to call hasNext() |
| timeToProcessCAS = (super.getCpuTime() - hasNextTime); |
| getNextTime = super.getCpuTime(); |
| // Start the heap dump timer. This timer is only started if explicitly enabled |
| // via System property: -DheapDumpThreshold=<x> where is number of seconds the |
| // method is allowed to complete. If the method is not complete in allowed window |
| // the heap and stack trace dump of all threads will be produced. |
| stackDumpTimer = ifEnabledStartHeapDumpTimer(); |
| CAS casProduced = casIterator.next(); |
| if ( stackDumpTimer != null ) { |
| stackDumpTimer.cancel(); |
| stackDumpTimer = null; // nullify timer instance so that we dont have to worry about |
| // it in case an exception happens below |
| } |
| // Add how long it took to call next() |
| timeToProcessCAS += (super.getCpuTime() - getNextTime); |
| // Add time to call hasNext() and next() to the running total |
| totalProcessTime += timeToProcessCAS; |
| casAbortedDueToExternalRequest = abortGeneratingCASes(aCasReferenceId); |
| // If the service is stopped or aborted, stop generating new CASes and just return the input |
| // CAS |
| if (stopped || casAbortedDueToExternalRequest) { |
| if (getInProcessCache() != null && getInProcessCache().getSize() > 0 |
| && getInProcessCache().entryExists(aCasReferenceId)) { |
| try { |
| // Set a flag on the input CAS to indicate that the processing was aborted |
| getInProcessCache().getCacheEntryForCAS(aCasReferenceId).setAborted(true); |
| } catch (Exception e) { |
| // An exception be be thrown here if the service is being stopped. |
| // The top level controller may have already cleaned up the cache |
| // and the getCacheEntryForCAS() will throw an exception. Ignore it |
| // here, we are shutting down. |
| } finally { |
| // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! |
| // We are terminating the iterator here, release the internal CAS lock |
| // so that we can release the CAS. This approach may need to be changed |
| // as there may potentially be a problem with a Class Loader. |
| // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! |
| ((CASImpl) aCAS).enableReset(true); |
| try { |
| // We are either stopping the service or aborting input CAS due to explicit STOP |
| // request |
| // from a client. If a new CAS was produced, release it back to the pool. |
| if (casProduced != null) { |
| casProduced.release(); |
| } |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.INFO, |
| getClass().getName(), |
| "process", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_cas_release_failed__INFO", |
| new Object[] { getComponentName(), |
| aCasReferenceId }); |
| } |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.INFO, |
| getClass().getName(), |
| "process", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_stopped_producing_new_cases__INFO", |
| new Object[] { Thread.currentThread().getId(), getComponentName(), |
| aCasReferenceId }); |
| } |
| } |
| } |
| if (casAbortedDueToExternalRequest) { |
| |
| // The controller was told to stop generating new CASes. Just return the input CAS to |
| // the |
| // client |
| // throw new ResourceProcessException(new |
| // InterruptedException("Cas Multiplier:"+getComponentName()+" Aborted CAS:"+aCasReferenceId)); |
| |
| break; // break out of the cas producing loop and return an input CAS to the client |
| } else { |
| // The controller is stopping |
| return; |
| } |
| } |
| // OutOfTypeSystemData otsd = getInProcessCache().getOutOfTypeSystemData(aCasReferenceId); |
| MessageContext mContext = getInProcessCache() |
| .getMessageAccessorByReference(aCasReferenceId); |
| CacheEntry newEntry = getInProcessCache().register(casProduced, mContext /*, otsd*/); |
| // if this Cas Multiplier is not Top Level service, add new Cas Id to the private |
| // cache of the parent aggregate controller. The Aggregate needs to know about |
| // all CASes it has in play that were generated from the input CAS. |
| CasStateEntry childCasStateEntry = null; |
| if (!isTopLevelComponent()) { |
| newEntry.setNewCas(true, parentController.getComponentName()); |
| // Create CAS state entry in the aggregate's local cache |
| childCasStateEntry = parentController.getLocalCache().createCasStateEntry( |
| newEntry.getCasReferenceId()); |
| // Fetch the parent CAS state entry from the aggregate's local cache. We need to increment |
| // number of child CASes associated with it. |
| parentCasStateEntry = parentController.getLocalCache().lookupEntry(aCasReferenceId); |
| } else { |
| childCasStateEntry = getLocalCache().createCasStateEntry(newEntry.getCasReferenceId()); |
| } |
| // Associate parent CAS (input CAS) with the new CAS. |
| childCasStateEntry.setInputCasReferenceId(aCasReferenceId); |
| // Increment number of child CASes generated from the input CAS |
| parentCasStateEntry.incrementSubordinateCasInPlayCount(); |
| parentCasStateEntry.incrementOutstandingFlowCounter(); |
| |
| // Associate input CAS with the new CAS |
| newEntry.setInputCasReferenceId(aCasReferenceId); |
| newEntry.setCasSequence(sequence); |
| // Add to the cache how long it took to process the generated (subordinate) CAS |
| getCasStatistics(newEntry.getCasReferenceId()).incrementAnalysisTime(timeToProcessCAS); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINE, |
| getClass().getName(), |
| "process", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_produced_new_cas__FINE", |
| new Object[] { Thread.currentThread().getName(), |
| getUimaContextAdmin().getQualifiedContextName(), |
| newEntry.getCasReferenceId(), aCasReferenceId }); |
| } |
| // Add the generated CAS to the outstanding CAS Map. Client notification will release |
| // this CAS back to its pool |
| synchronized (syncObject) { |
| if (isTopLevelComponent()) { |
| // Add the id of the generated CAS to the map holding outstanding CASes. This |
| // map will be referenced when a client sends Free CAS Notification. The map |
| // stores the id of the CAS both as a key and a value. Map is used to facilitate |
| // quick lookup |
| cmOutstandingCASes.put(newEntry.getCasReferenceId(), newEntry.getCasReferenceId()); |
| } |
| // Increment number of CASes processed by this service |
| sequence++; |
| } |
| try { |
| CacheEntry cacheEntry = |
| getInProcessCache().getCacheEntryForCAS(aCasReferenceId); |
| |
| if ( isTopLevelComponent() && cacheEntry.isWarmUp() && "WarmupDelegate".equals(anEndpoint.getDelegateKey())) { |
| localCache.lookupEntry(newEntry.getCasReferenceId()).setDropped(true); |
| localCache.remove(newEntry.getCasReferenceId()); |
| // Remove Stats from the global Map associated with the new CAS |
| // These stats for this CAS were added to the response message |
| // and are no longer needed |
| dropCasStatistics(newEntry.getCasReferenceId()); |
| |
| return; |
| } |
| |
| } catch( Exception exx ) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "process", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", exx); |
| } |
| |
| if (!anEndpoint.isRemote()) { |
| UimaTransport transport = getTransport(anEndpoint.getEndpoint()); |
| UimaMessage message = transport.produceMessage(AsynchAEMessage.Process, |
| AsynchAEMessage.Request, getName()); |
| message.addStringProperty(AsynchAEMessage.CasReference, newEntry.getCasReferenceId()); |
| message.addStringProperty(AsynchAEMessage.InputCasReference, aCasReferenceId); |
| message.addLongProperty(AsynchAEMessage.CasSequence, sequence); |
| ServicePerformance casStats = getCasStatistics(aCasReferenceId); |
| |
| message.addLongProperty(AsynchAEMessage.TimeToSerializeCAS, casStats |
| .getRawCasSerializationTime()); |
| message.addLongProperty(AsynchAEMessage.TimeToDeserializeCAS, casStats |
| .getRawCasDeserializationTime()); |
| message.addLongProperty(AsynchAEMessage.TimeInProcessCAS, casStats.getRawAnalysisTime()); |
| long iT = getIdleTimeBetweenProcessCalls(AsynchAEMessage.Process); |
| message.addLongProperty(AsynchAEMessage.IdleTime, iT); |
| if (!stopped) { |
| transport.getUimaMessageDispatcher(anEndpoint.getEndpoint()).dispatch(message); |
| dropStats(newEntry.getCasReferenceId(), getName()); |
| } |
| } else { |
| // Send generated CAS to the remote client |
| if (!stopped) { |
| getOutputChannel().sendReply(newEntry, anEndpoint); |
| |
| // Check for delivery failure. The client may have terminated while an input CAS was being processed |
| if ( childCasStateEntry.deliveryToClientFailed() ) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "process", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_delivery_to_client_failed_INFO", |
| new Object[] { getComponentName(), aCasReferenceId }); |
| } |
| clientUnreachable = true; |
| if ( cmOutstandingCASes.containsKey(childCasStateEntry.getCasReferenceId())) { |
| cmOutstandingCASes.remove(childCasStateEntry.getCasReferenceId()); |
| } |
| // Stop generating new CASes. We failed to send a CAS to a client. Most likely |
| // the client has terminated. |
| moreCASesToProcess = false; // exit the while loop |
| |
| dropCAS(childCasStateEntry.getCasReferenceId(), true); |
| } |
| } |
| } |
| // Remove new CAS state entry from the local cache if this is a top level primitive. |
| // If not top level, the client (an Aggregate) will remove this entry when this new |
| // generated CAS reaches Final State. |
| if (isTopLevelComponent()) { |
| try { |
| localCache.lookupEntry(newEntry.getCasReferenceId()).setDropped(true); |
| } catch (Exception e) { |
| } |
| localCache.remove(newEntry.getCasReferenceId()); |
| } |
| |
| // Remove Stats from the global Map associated with the new CAS |
| // These stats for this CAS were added to the response message |
| // and are no longer needed |
| dropCasStatistics(newEntry.getCasReferenceId()); |
| } // while |
| |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINEST)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb( |
| Level.FINEST, |
| getClass().getName(), |
| "process", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_completed_analysis__FINEST", |
| new Object[] { Thread.currentThread().getName(), getComponentName(), |
| aCasReferenceId, (double) (super.getCpuTime() - time) / (double) 1000000 }); |
| } |
| |
| |
| // check if this is a warm up CAS. Such CAS is internally created with |
| // a main purpose of warming up analytics. |
| CacheEntry cacheEntry = |
| getInProcessCache().getCacheEntryForCAS(aCasReferenceId); |
| |
| if ( isTopLevelComponent() && cacheEntry.isWarmUp()) { |
| // if ( cacheEntry.getThreadCompletionSemaphore() != null ) { |
| // cacheEntry.getThreadCompletionSemaphore().release(); |
| // } |
| // we are in the warm state which means the pipelines have been initialized |
| // and we are sending CASes to warm up/prime the analytics. Since we |
| // reached the end of the flow here, just return now. There is nothing |
| // else to do with the CAS. |
| return; |
| } |
| |
| getMonitor().resetCountingStatistic("", Monitor.ProcessErrorCount); |
| // Set total number of children generated from this CAS |
| // Store total time spent processing this input CAS |
| getCasStatistics(aCasReferenceId).incrementAnalysisTime(totalProcessTime); |
| // Fetch AE's management information that includes per component performance stats |
| // These stats are internally maintained in a Map. If the AE is an aggregate |
| // the Map will contain AnalysisEngineManagement instance for each AE. |
| AnalysisEngineManagement aem = ae.getManagementInterface(); |
| if ( aem.getComponents().size() > 0) { |
| // Flatten the hierarchy by recursively (if this AE is an aggregate) extracting |
| // primitive AE's AnalysisEngineManagement instance and placing it in |
| // afterAnalysisManagementObjects List. |
| getLeafManagementObjects(aem, afterAnalysisManagementObjects); |
| } else { |
| String path=produceUniqueName(aem); |
| afterAnalysisManagementObjects.add(deepCopyMetrics(aem, path)); |
| |
| } |
| // Create a List to hold per CAS analysisTime and total number of CASes processed |
| // by each AE. This list will be serialized and sent to the client |
| List<AnalysisEnginePerformanceMetrics> performanceList = |
| new ArrayList<AnalysisEnginePerformanceMetrics>(); |
| // Diff the before process() performance metrics with post process performance |
| // metrics |
| for (AnalysisEnginePerformanceMetrics after : afterAnalysisManagementObjects) { |
| for( AnalysisEnginePerformanceMetrics before: beforeAnalysisManagementObjects) { |
| if ( before.getUniqueName().equals(after.getUniqueName())) { |
| boolean found = false; |
| AnalysisEnginePerformanceMetrics metrics = null; |
| for( AnalysisEnginePerformanceMetrics met : parentCasStateEntry.getAEPerformanceList() ) { |
| String un = after.getUniqueName(); |
| if ( un.indexOf("Components") >= -1 ) { |
| un = un.substring(un.indexOf("/")); |
| } |
| if ( met.getUniqueName().equals(un)) { |
| long at = after.getAnalysisTime()- before.getAnalysisTime(); |
| metrics = new AnalysisEnginePerformanceMetrics(after.getName(), |
| un,//after.getUniqueName(), |
| met.getAnalysisTime()+at, |
| after.getNumProcessed()); |
| found = true; |
| parentCasStateEntry.getAEPerformanceList().remove(met); |
| break; |
| } |
| } |
| if ( !found ) { |
| String un = after.getUniqueName(); |
| |
| if ( un.indexOf("Components") >= -1 ) { |
| un = un.substring(un.indexOf("/")); |
| } |
| metrics = new AnalysisEnginePerformanceMetrics(after.getName(), |
| un,//after.getUniqueName(), |
| after.getAnalysisTime()- before.getAnalysisTime(), |
| after.getNumProcessed()); |
| |
| } |
| performanceList.add(metrics); |
| break; |
| } |
| } |
| } |
| parentCasStateEntry.getAEPerformanceList().addAll(performanceList); |
| if (!anEndpoint.isRemote()) { |
| inputCASReturned = true; |
| UimaTransport transport = getTransport(anEndpoint.getEndpoint()); |
| |
| if (getInProcessCache() != null && getInProcessCache().getSize() > 0 |
| && getInProcessCache().entryExists(aCasReferenceId)) { |
| try { |
| CacheEntry ancestor = |
| getInProcessCache(). |
| getTopAncestorCasEntry(getInProcessCache().getCacheEntryForCAS(aCasReferenceId)); |
| if ( ancestor != null ) { |
| ancestor.addDelegateMetrics(getKey(), performanceList); |
| } |
| } catch (Exception e) { |
| // An exception be be thrown here if the service is being stopped. |
| // The top level controller may have already cleaned up the cache |
| // and the getCacheEntryForCAS() will throw an exception. Ignore it |
| // here, we are shutting down. |
| } |
| } |
| |
| UimaMessage message = transport.produceMessage(AsynchAEMessage.Process, |
| AsynchAEMessage.Response, getName()); |
| message.addStringProperty(AsynchAEMessage.CasReference, aCasReferenceId); |
| ServicePerformance casStats = getCasStatistics(aCasReferenceId); |
| |
| message.addLongProperty(AsynchAEMessage.TimeToSerializeCAS, casStats |
| .getRawCasSerializationTime()); |
| message.addLongProperty(AsynchAEMessage.TimeToDeserializeCAS, casStats |
| .getRawCasDeserializationTime()); |
| message.addLongProperty(AsynchAEMessage.TimeInProcessCAS, casStats.getRawAnalysisTime()); |
| long iT = getIdleTimeBetweenProcessCalls(AsynchAEMessage.Process); |
| message.addLongProperty(AsynchAEMessage.IdleTime, iT); |
| // Send reply back to the client. Use internal (non-jms) transport |
| if (!stopped) { |
| transport.getUimaMessageDispatcher(anEndpoint.getEndpoint()).dispatch(message); |
| dropStats(aCasReferenceId, getName()); |
| } |
| } else { |
| try { |
| List<AnalysisEnginePerformanceMetrics> perfMetrics = |
| new ArrayList<AnalysisEnginePerformanceMetrics>(); |
| String aeName = getMetaData().getName(); |
| |
| CacheEntry entry = |
| getInProcessCache().getCacheEntryForCAS(aCasReferenceId); |
| for( AnalysisEnginePerformanceMetrics m : performanceList ) { |
| // System.out.println("...............BEFORE: Name:"+m.getName()+" UniqueName:"+m.getUniqueName()+" How Many="+m.getNumProcessed()); |
| boolean aggregate = m.getUniqueName().startsWith("/"+aeName); |
| int pos = m.getUniqueName().indexOf("/",1); |
| String uName = m.getUniqueName(); |
| if ( pos > -1 && aeInstancePool.size() > 1 && aeName != null && aggregate) { |
| String st = m.getUniqueName().substring(pos); |
| uName = "/"+aeName+st; |
| } |
| AnalysisEnginePerformanceMetrics newMetrics = |
| new AnalysisEnginePerformanceMetrics(m.getName(),uName,m.getAnalysisTime(), m.getNumProcessed()); |
| // System.out.println("... Metrics - AE:"+metrics.getUniqueName()+" AE Analysis Time:"+metrics.getAnalysisTime()); |
| perfMetrics.add(newMetrics); |
| // System.out.println("...............AFTER: Name:"+newMetrics.getName()+" UniqueName:"+newMetrics.getUniqueName()+" How Many="+newMetrics.getNumProcessed()); |
| |
| } |
| entry.addDelegateMetrics(getKey(), perfMetrics); //performanceList); |
| } catch (Exception e) { |
| // An exception be be thrown here if the service is being stopped. |
| // The top level controller may have already cleaned up the cache |
| // and the getCacheEntryForCAS() will throw an exception. Ignore it |
| // here, we are shutting down. |
| } |
| |
| if (!stopped && !clientUnreachable ) { |
| getOutputChannel().sendReply(getInProcessCache().getCacheEntryForCAS(aCasReferenceId), anEndpoint); |
| } |
| |
| inputCASReturned = true; |
| } |
| |
| // Remove input CAS state entry from the local cache |
| if (!isTopLevelComponent()) { |
| localCache.lookupEntry(aCasReferenceId).setDropped(true); |
| localCache.remove(aCasReferenceId); |
| } |
| } catch (Throwable e) { |
| |
| if ( e instanceof OutOfMemoryError ) { |
| e.printStackTrace(); |
| System.err.println("\n\n\n\t!!!!! UIMA AS Service Caught Java Error While in process() method. Exiting via System.exit(2)\n\n\n"); |
| System.err.flush(); |
| System.exit(2); |
| } |
| if ( stackDumpTimer != null ) { |
| stackDumpTimer.cancel(); |
| } |
| processingFailed = true; |
| ErrorContext errorContext = new ErrorContext(); |
| errorContext.add(AsynchAEMessage.CasReference, aCasReferenceId); |
| errorContext.add(AsynchAEMessage.Command, AsynchAEMessage.Process); |
| errorContext.add(AsynchAEMessage.MessageType, AsynchAEMessage.Request); |
| errorContext.add(AsynchAEMessage.Endpoint, anEndpoint); |
| // Handle the exception. Pass reference to the PrimitiveController instance |
| getErrorHandlerChain().handle(e, errorContext, this); |
| } finally { |
| dropCasStatistics(aCasReferenceId); |
| |
| if (ae != null) { |
| try { |
| aeInstancePool.checkin(ae); |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "process", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_service_exception_WARNING", getComponentName()); |
| |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "process", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", e); |
| } |
| } |
| } |
| // drop the CAS if it has been successfully processed. If there was a failure, the Error |
| // Handler |
| // will drop the CAS |
| if (isTopLevelComponent() && !processingFailed) { |
| // Release CASes produced from the input CAS if the input CAS has been aborted |
| if (abortGeneratingCASes(aCasReferenceId) || clientUnreachable ) { |
| |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "process", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_remove_cache_entry__INFO", |
| new Object[] { getComponentName(), aCasReferenceId }); |
| } |
| getInProcessCache().releaseCASesProducedFromInputCAS(aCasReferenceId); |
| } else if (inputCASReturned && isTopLevelComponent()) { |
| |
| if ( clientUnreachable ) { |
| ((CASImpl) aCAS).enableReset(true); |
| } |
| // Remove input CAS cache entry if the CAS has been sent to the client |
| dropCAS(aCasReferenceId, true); |
| localCache.dumpContents(); |
| } |
| } |
| } |
| } |
| |
| private void addConfigIntParameter(String aParamName, int aParamValue) { |
| ConfigurationParameter cp = new ConfigurationParameter_impl(); |
| cp.setMandatory(false); |
| cp.setMultiValued(false); |
| cp.setName(aParamName); |
| cp.setType("Integer"); |
| getAnalysisEngineMetadata().getConfigurationParameterDeclarations().addConfigurationParameter( |
| cp); |
| getAnalysisEngineMetadata().getConfigurationParameterSettings().setParameterValue(aParamName, |
| aParamValue); |
| |
| } |
| |
| // Return metadata |
| public void sendMetadata(Endpoint anEndpoint) throws AsynchAEException { |
| |
| if ( ((ProcessingResourceMetaData) getAnalysisEngineMetadata()) |
| .getConfigurationParameterSettings().getParameterValue( |
| AnalysisEngineController.AEInstanceCount) == null ) { |
| addConfigIntParameter(AnalysisEngineController.AEInstanceCount, analysisEnginePoolSize); |
| } |
| |
| if (getAnalysisEngineMetadata().getOperationalProperties().getOutputsNewCASes()) { |
| if ( ((ProcessingResourceMetaData) getAnalysisEngineMetadata()) |
| .getConfigurationParameterSettings().getParameterValue( |
| AnalysisEngineController.CasPoolSize) == null ) { |
| addConfigIntParameter(AnalysisEngineController.CasPoolSize, super.componentCasPoolSize); |
| } |
| |
| } |
| super.sendMetadata(anEndpoint, getAnalysisEngineMetadata()); |
| } |
| |
| private AnalysisEngineMetaData getAnalysisEngineMetadata() { |
| return analysisEngineMetadata; |
| } |
| |
| /** |
| * Executes action on error. Primitive Controller allows two types of actions TERMINATE and |
| * DROPCAS. |
| */ |
| public void takeAction(String anAction, String anEndpointName, ErrorContext anErrorContext) { |
| try { |
| if (ErrorHandler.TERMINATE.equalsIgnoreCase(anAction) |
| || ErrorHandler.DROPCAS.equalsIgnoreCase(anAction)) { |
| super.handleAction(anAction, anEndpointName, anErrorContext); |
| } |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "takeAction", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_service_exception_WARNING", getComponentName()); |
| |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "takeAction", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", e); |
| } |
| } |
| } |
| |
| public String getServiceEndpointName() { |
| return getName(); |
| } |
| |
| public synchronized ControllerLatch getControllerLatch() { |
| return latch; |
| } |
| |
| public boolean isPrimitive() { |
| return true; |
| } |
| |
| public Monitor getMonitor() { |
| return super.monitor; |
| } |
| |
| public void setMonitor(Monitor monitor) { |
| this.monitor = monitor; |
| } |
| |
| public void handleDelegateLifeCycleEvent(String anEndpoint, int aDelegateCount) { |
| if (aDelegateCount == 0) { |
| // tbi |
| } |
| } |
| |
| protected String getNameFromMetadata() { |
| return super.getMetaData().getName(); |
| } |
| |
| public void setAnalysisEngineInstancePool(AnalysisEngineInstancePool aPool) { |
| aeInstancePool = aPool; |
| } |
| |
| public PrimitiveServiceInfo getServiceInfo() { |
| if (serviceInfo == null) { |
| serviceInfo = new PrimitiveServiceInfo(isCasMultiplier(), this); |
| serviceInfo.setServiceKey(delegateKey); |
| } |
| if (isTopLevelComponent() && getInputChannel() != null) { |
| serviceInfo.setInputQueueName(getInputChannel().getServiceInfo().getInputQueueName()); |
| serviceInfo.setBrokerURL(super.getBrokerURL()); |
| serviceInfo.setDeploymentDescriptorPath(super.aeDescriptor); |
| } |
| serviceInfo.setAnalysisEngineInstanceCount(analysisEnginePoolSize); |
| return serviceInfo; |
| } |
| |
| public void stop() { |
| super.stop(true); // shutdown now |
| if (aeInstancePool != null) { |
| try { |
| aeInstancePool.destroy(); |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "stop", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_service_exception_WARNING", getComponentName()); |
| |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "stop", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", e); |
| } |
| } |
| } |
| try { |
| for (Entry<String, UimaTransport> transport : transports.entrySet()) { |
| transport.getValue().stopIt(); |
| } |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "stop", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_service_exception_WARNING", getComponentName()); |
| |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "stop", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", e); |
| } |
| } |
| |
| if (cmOutstandingCASes != null) { |
| if (!cmOutstandingCASes.isEmpty()) { |
| // If there are outstanding CASes, force them to be released |
| // If the CM is blocking on getCAS() this will unblock it and |
| // enable termination. Otherwise, a hang may occur |
| Iterator<String> it = cmOutstandingCASes.keySet().iterator(); |
| while (it.hasNext()) { |
| String casId = it.next(); |
| try { |
| CacheEntry entry = getInProcessCache().getCacheEntryForCAS(casId); |
| if (entry != null && entry.getCas() != null) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "stop", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_force_cas_release___INFO", |
| new Object[] { getComponentName(), casId }); |
| } |
| // Force CAS release to unblock CM thread |
| entry.getCas().release(); |
| } |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "stop", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_parent_cas_notin_cache__INFO", |
| new Object[] { getComponentName(), casId }); |
| } |
| } |
| } |
| |
| } |
| cmOutstandingCASes.clear(); |
| } |
| if (aeList != null) { |
| aeList.clear(); |
| aeList = null; |
| } |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.INFO)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.INFO, CLASS_NAME.getName(), "stop", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_service_stopped__INFO", |
| new Object[] { getComponentName()}); |
| } |
| // Only top-level service may call exit. Dont exit if running jUnit tests. |
| if (isTopLevelComponent() && System.getProperty("dontKill") == null) { |
| System.exit(0); |
| } |
| } |
| /** |
| This method forces a heap and java dump. It only works with IBM jvm. |
| **/ |
| protected final synchronized void forceStackDump() { |
| Class<?> params[] = {}; |
| Object paramsObj[] = {}; |
| |
| |
| try { |
| Class<?> dumpClass = Class.forName("com.ibm.jvm.Dump"); |
| // Get a handle to the static method |
| Method javaDumpMethod = dumpClass.getDeclaredMethod("JavaDump", params); |
| // dump the stack of all threads |
| javaDumpMethod.invoke(null, paramsObj); |
| } catch( ClassNotFoundException e) { |
| // ignore ClassNotFoundException in case of sun jvm |
| } catch( InvocationTargetException e) { |
| |
| } catch (NoSuchMethodException e) { |
| |
| } catch( IllegalAccessException e) { |
| |
| } |
| } |
| |
| /** |
| * The HeapDumpTimer is optionally used to dump the heap if a task takes too much time to finish. |
| * It is enabled from the System property -DheapDumpThreshold=x where x is a number of seconds |
| * the task is allowed to complete. If the task is not completed, the heap dump will be created. |
| * |
| * |
| */ |
| public class StackDumpTimer { |
| Timer timer; |
| |
| public StackDumpTimer ( int seconds ) { |
| timer = new Timer ( ) ; |
| timer.schedule ( new dumpTheStackTask ( ) , seconds*1000 ) ; |
| } |
| public void cancel() { |
| timer.cancel(); |
| timer.purge(); |
| } |
| class dumpTheStackTask extends TimerTask { |
| public void run ( ) { |
| timer.cancel ( ) ; //Terminate the thread |
| // create a heap dump. NOTE: this only works with IBM jvm |
| synchronized (threadDumpMonitor) { |
| long now = System.currentTimeMillis(); |
| if (30000 < (now-lastDump)) { |
| // create a stack dump. NOTE: this only works with IBM jvm |
| forceStackDump(); |
| lastDump = now; |
| } |
| } |
| } |
| } |
| |
| } |
| |
| public void dumpState(StringBuffer buffer, String lbl1) { |
| buffer.append(getComponentName()+" State:"+getState()); |
| } |
| |
| protected void doWarmUp(CAS cas, String casReferenceId) throws Exception { |
| long processTime = 0; |
| try { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, getClass().getName(), "doWarmUp", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_warmup_start_cas__FINE", |
| new Object[] { casReferenceId}); |
| } |
| |
| long t1 = System.currentTimeMillis(); |
| |
| Endpoint endpoint = new Endpoint_impl(); |
| // set fake delegate key. This will be checked in process() method |
| // to make sure we are not attempting sending the CAS to a reply queue. |
| // Warm up simulates incoming CASes but there is no client to reply to. |
| endpoint.setDelegateKey("WarmupDelegate"); |
| |
| process(cas, casReferenceId, endpoint); |
| processTime = System.currentTimeMillis() - t1; |
| |
| // CacheEntry entry = getInProcessCache().getCacheEntryForCAS(casReferenceId); |
| // if ( entry != null && entry.isFailed()) { |
| // if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| // UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| // "doWarmUp", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| // "UIMAEE_service_warmup_failed_WARNING", getComponentName()); |
| // } |
| // throw new ResourceInitializationException(); |
| // } |
| } catch( Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, CLASS_NAME.getName(), |
| "doWarmUp", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_service_warmup_failed_WARNING", getComponentName()); |
| } |
| throw e; |
| } finally { |
| dropCAS(casReferenceId, true); |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, getClass().getName(), "doWarmUp", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_warmup_dropping_cas__FINE", |
| new Object[] { casReferenceId, processTime}); |
| } |
| } |
| } |
| } |
