| /* |
| * 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; |
| |
| import java.util.ArrayList; |
| import java.util.HashMap; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Map.Entry; |
| import java.util.Set; |
| import java.util.Stack; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.Semaphore; |
| |
| import org.apache.uima.UIMAFramework; |
| import org.apache.uima.aae.controller.BaseAnalysisEngineController; |
| import org.apache.uima.aae.controller.Endpoint; |
| import org.apache.uima.aae.controller.EventSubscriber; |
| import org.apache.uima.aae.error.AsynchAEException; |
| import org.apache.uima.aae.message.AsynchAEMessage; |
| import org.apache.uima.aae.message.MessageContext; |
| import org.apache.uima.aae.monitor.statistics.AnalysisEnginePerformanceMetrics; |
| import org.apache.uima.aae.monitor.statistics.DelegateStats; |
| import org.apache.uima.cas.CAS; |
| import org.apache.uima.cas.Marker; |
| import org.apache.uima.cas.impl.BinaryCasSerDes6.ReuseInfo; |
| import org.apache.uima.cas.impl.XmiSerializationSharedData; |
| import org.apache.uima.util.Level; |
| |
| public class InProcessCache implements InProcessCacheMBean { |
| /** |
| * |
| */ |
| private static final long serialVersionUID = 1L; |
| |
| private static final Class<InProcessCache> CLASS_NAME = InProcessCache.class; |
| |
| public static final int NotifyWhenRegistering = 1; |
| |
| private transient UIDGenerator idGenerator = new UIDGenerator(); |
| |
| private ConcurrentHashMap<String, CacheEntry> cache = new ConcurrentHashMap<String, CacheEntry>(); |
| |
| private String name = "InProcessCache"; |
| |
| private List<EventSubscriber> callbackListeners = new ArrayList<EventSubscriber>(); |
| |
| int size = 0; |
| |
| private BaseAnalysisEngineController controller; |
| |
| /** |
| Register controller to call when the cache becomes empty. |
| This call is made when the controller enters quiesce |
| state. In this state the controller waits for |
| the cache to send notification when all CASes have been |
| processed. |
| **/ |
| public void registerController(BaseAnalysisEngineController ctrl) { |
| controller = ctrl; |
| } |
| public void registerCallbackWhenCacheEmpty(EventSubscriber aController) { |
| registerCallbackWhenCacheEmpty(aController, 0); |
| } |
| |
| public void registerCallbackWhenCacheEmpty(EventSubscriber aController, int notification) { |
| if (!callbackListeners.isEmpty()) { |
| Iterator<EventSubscriber> it = callbackListeners.iterator(); |
| while (it.hasNext()) { |
| EventSubscriber es = (EventSubscriber) it.next(); |
| if (es == aController) { |
| return; |
| } |
| } |
| } else if (notification == NotifyWhenRegistering) { |
| if (isEmpty()) { |
| aController.onCacheEmpty(); |
| } |
| } |
| // If not registered already add the event subscriber to the list of |
| // of objects to be called when the cache becomes empty |
| if (!callbackListeners.contains(aController)) { |
| callbackListeners.add(aController); |
| } |
| } |
| |
| public void destroy() { |
| callbackListeners.clear(); |
| Set<Entry<String, CacheEntry>> set = cache.entrySet(); |
| for (Iterator<Entry<String, CacheEntry>> it = set.iterator(); it.hasNext();) { |
| Map.Entry<String, CacheEntry> entry = (Map.Entry<String, CacheEntry>) it.next(); |
| CacheEntry cacheEntry = (CacheEntry) entry.getValue(); |
| if (cacheEntry != null && cacheEntry.getCas() != null) { |
| try { |
| cacheEntry.getCas().release(); |
| } catch (Exception e) { |
| } |
| } |
| } |
| |
| cache.clear(); |
| } |
| |
| /** |
| * Checks if a given input CAS is in pending state. CAS is in pending state if it has been fully |
| * processed, *but* its subordinate CASes are still in play. Input CAS is only returned back to |
| * the client if all if its subordinate CASes are fully processed. |
| * |
| * @param anInputCASReferenceId |
| * @return |
| * @throws Exception |
| */ |
| public boolean isInputCASPendingReply(String anInputCASReferenceId) throws Exception { |
| if (anInputCASReferenceId == null) { |
| return false; |
| } |
| CacheEntry inputCASEntry = getCacheEntryForCAS(anInputCASReferenceId); |
| |
| return inputCASEntry.isPendingReply(); |
| } |
| |
| public boolean producedCASesStillInPlay(String anInputCASReferenceId, |
| String aSubordinateCASReferenceId) throws Exception { |
| Iterator<String> it = cache.keySet().iterator(); |
| while (it.hasNext()) { |
| String key = (String) it.next(); |
| CacheEntry entry = (CacheEntry) cache.get(key); |
| if (entry != null && aSubordinateCASReferenceId != null |
| && aSubordinateCASReferenceId.equals(key)) { |
| continue; // dont count the current subordinate |
| } |
| if (entry != null && anInputCASReferenceId.equals(entry.getInputCasReferenceId())) { |
| return true; |
| } |
| } |
| return false; |
| } |
| |
| public void releaseCASesProducedFromInputCAS(String anInputCASReferenceId) { |
| if (anInputCASReferenceId == null) { |
| return; |
| } |
| Iterator<String> it = cache.keySet().iterator(); |
| while (it.hasNext()) { |
| String key = (String) it.next(); |
| CacheEntry entry = (CacheEntry) cache.get(key); |
| if (entry != null |
| && (anInputCASReferenceId.equals(key) || anInputCASReferenceId.equals(entry |
| .getInputCasReferenceId()))) { |
| if (entry.getCas() != null) { |
| entry.getCas().release(); |
| } |
| remove(key); |
| } |
| } |
| |
| } |
| |
| public void releaseAllCASes() { |
| Iterator<String> it = cache.keySet().iterator(); |
| while (it.hasNext()) { |
| String key = (String) it.next(); |
| CacheEntry entry = (CacheEntry) cache.get(key); |
| if (entry != null && entry.getCas() != null) { |
| try { |
| entry.getCas().release(); |
| } catch (Exception e) { |
| } |
| } |
| cache.remove(key); |
| } |
| |
| } |
| |
| public void setName(String aName) { |
| name = aName; |
| } |
| |
| public String getName() { |
| return name; |
| } |
| |
| public void cancelTimers() { |
| } |
| |
| public boolean isEmpty() { |
| return (cache.size() == 0); |
| } |
| |
| public void reset() { |
| size = 0; |
| } |
| |
| public int getSize() { |
| return cache.size(); |
| } |
| |
| public void setSize(int i) { |
| size = i; |
| } |
| |
| public synchronized void dumpContents(String aControllerName) { |
| // int count = 0; |
| /* |
| * if ( UIMAFramework.getLogger().isLoggable(Level.FINEST) ) { Iterator it = |
| * cache.keySet().iterator(); StringBuffer sb = new StringBuffer("\n"); |
| * |
| * while( it.hasNext() ) { String key = (String) it.next(); CacheEntry entry = |
| * (CacheEntry)cache.get(key); count++; if ( entry.isSubordinate()) { sb.append(key+ |
| * " Number Of Child CASes In Play:" |
| * +entry.getSubordinateCasInPlayCount()+" Parent CAS id:"+entry.getInputCasReferenceId()); } |
| * else { sb.append(key+ |
| * " *** Input CAS. Number Of Child CASes In Play:"+entry.getSubordinateCasInPlayCount()); } |
| * |
| * // if ( entry.isWaitingForRelease() ) // { // |
| * sb.append(" <<< Reached Final State in Controller:"+aControllerName); // } |
| * |
| * sb.append("\n"); } UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINEST, |
| * CLASS_NAME.getName(), "dumpContents", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| * "UIMAEE_show_cache_entry_key__FINEST", new Object[] { aControllerName, count, sb.toString() |
| * }); |
| * |
| * sb.setLength(0); } else if ( UIMAFramework.getLogger().isLoggable(Level.FINE) ) { Iterator it |
| * = cache.keySet().iterator(); StringBuffer sb = new StringBuffer("\n"); int inFinalState=0; |
| * |
| * while( it.hasNext() ) { String key = (String) it.next(); CacheEntry entry = |
| * (CacheEntry)cache.get(key); count++; |
| * |
| * //if ( entry.isWaitingForRelease() ) //{ //inFinalState++; //} |
| * |
| * } UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), "dumpContents", |
| * UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_show_abbrev_cache_stats___FINE", new |
| * Object[] { aControllerName, count, inFinalState }); |
| * |
| * |
| * } |
| */ |
| } |
| |
| public synchronized void remove(String aCasReferenceId) { |
| if (aCasReferenceId != null && cache.containsKey(aCasReferenceId)) { |
| |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, getClass().getName(), "remove", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_remove_cache_entry_for_cas__FINE", new Object[] { aCasReferenceId }); |
| } |
| cache.remove(aCasReferenceId); |
| this.notifyAll(); |
| } else if (aCasReferenceId == null) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, getClass().getName(), "remove", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_cas_is_null_remove_from_cache_failed__FINE"); |
| } |
| } else { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, getClass().getName(), "remove", |
| UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_cas_is_invalid_remove_from_cache_failed__FINE", |
| new Object[] { aCasReferenceId }); |
| } |
| } |
| if (cache.size() == 0 && callbackListeners.size() > 0) { |
| for (int i = 0; i < callbackListeners.size(); i++) { |
| ((EventSubscriber) callbackListeners.get(i)).onCacheEmpty(); |
| } |
| } |
| if (cache.size() == 0 && controller != null ) { |
| // unblock the controller waiting in quiesceAndStop |
| controller.notifyOnCacheEmpty(); |
| } |
| } |
| |
| // never called 5/2013 |
| // public void removeCas(String aCasReferenceId) { |
| // if (!cache.containsKey(aCasReferenceId)) { |
| // return; |
| // } |
| // CacheEntry casRefEntry = getEntry(aCasReferenceId); |
| // casRefEntry.deleteCAS(); |
| // } |
| |
| public synchronized CacheEntry[] getCacheEntriesForEndpoint(String anEndpointName) { |
| CacheEntry[] entries; |
| ArrayList<CacheEntry> list = new ArrayList<CacheEntry>(); |
| Iterator<String> it = cache.keySet().iterator(); |
| while (it.hasNext()) { |
| String key = (String) it.next(); |
| CacheEntry entry = (CacheEntry) cache.get(key); |
| if (entry != null && entry.getEndpoint(anEndpointName) != null) { |
| list.add(entry); |
| } |
| } |
| if (list.size() > 0) { |
| entries = new CacheEntry[list.size()]; |
| list.toArray(entries); |
| return entries; |
| } |
| return null; |
| } |
| |
| public void saveSerializedCAS(String aCasReferenceId, String anXCAS) { |
| if (!cache.containsKey(aCasReferenceId)) { |
| return; |
| } |
| CacheEntry casRefEntry = getEntry(aCasReferenceId); |
| casRefEntry.setSerializedCas(anXCAS); |
| } |
| |
| public String getSerializedCAS(String aCasReferenceId) { |
| if (cache.containsKey(aCasReferenceId)) { |
| CacheEntry casRefEntry = getEntry(aCasReferenceId); |
| return casRefEntry.getSerializedCas(); |
| } |
| return null; |
| } |
| |
| public void setCasProducer(String aCasReferenceId, String aCasProducerKey) { |
| if (cache.containsKey(aCasReferenceId)) { |
| CacheEntry casRefEntry = getEntry(aCasReferenceId); |
| casRefEntry.setCasProducerKey(aCasProducerKey); |
| } |
| |
| } |
| |
| public String getCasProducer(String aCasReferenceId) { |
| if (cache.containsKey(aCasReferenceId)) { |
| CacheEntry casRefEntry = getEntry(aCasReferenceId); |
| return casRefEntry.getCasProducerKey(); |
| } |
| return null; |
| } |
| |
| public synchronized CAS getCasByReference(String aCasReferenceId) { |
| if (!cache.containsKey(aCasReferenceId)) { |
| return null; |
| } |
| CacheEntry casRefEntry = getEntry(aCasReferenceId); |
| return casRefEntry.getCas(); |
| } |
| |
| public MessageContext getMessageAccessorByReference(String aCasReferenceId) { |
| if (!cache.containsKey(aCasReferenceId)) { |
| return null; |
| } |
| CacheEntry casRefEntry = getEntry(aCasReferenceId); |
| return casRefEntry.getMessageAccessor(); |
| } |
| |
| |
| // not used 5/2013 |
| // public OutOfTypeSystemData getOutOfTypeSystemData(String aCasReferenceId) { |
| // if (!cache.containsKey(aCasReferenceId)) { |
| // return null; |
| // } |
| // CacheEntry casRefEntry = getEntry(aCasReferenceId); |
| // return casRefEntry.getOtsd(); |
| // } |
| |
| private synchronized CacheEntry getEntry(String aCasReferenceId) { |
| if (!cache.containsKey(aCasReferenceId)) { |
| return null; |
| } |
| return (CacheEntry) cache.get(aCasReferenceId); |
| } |
| |
| public void addEndpoint(Endpoint anEndpoint, String aCasReferenceId) { |
| CacheEntry casRefEntry = getEntry(aCasReferenceId); |
| if (!casRefEntry.getEndpointMap().containsKey(anEndpoint.getEndpoint())) { |
| casRefEntry.addEndpoint(anEndpoint); |
| } |
| } |
| |
| public Endpoint getEndpoint(String anEndpointName, String aCasReferenceId) { |
| CacheEntry casRefEntry = getEntry(aCasReferenceId); |
| if (anEndpointName == null && casRefEntry != null) { |
| return casRefEntry.getMessageOrigin(); |
| } |
| if (casRefEntry == null) { |
| return null; |
| } else { |
| return casRefEntry.getEndpoint(anEndpointName); |
| } |
| } |
| |
| public void removeEndpoint(String anEndpointName, String aCasReferenceId) { |
| CacheEntry casRefEntry = getEntry(aCasReferenceId); |
| casRefEntry.deleteEndpoint(anEndpointName); |
| } |
| |
| public long getStartTime(String aCasReferenceId) { |
| CacheEntry casRefEntry = getEntry(aCasReferenceId); |
| return casRefEntry.getStartTime(); |
| } |
| |
| public synchronized boolean entryExists(String aCasReferenceId) { |
| try { |
| CacheEntry casRefEntry = getEntry(aCasReferenceId); |
| if (casRefEntry == null) { |
| return false; |
| } |
| } catch (Exception e) { |
| // ignore |
| } |
| return true; |
| } |
| |
| public CacheEntry register(CAS aCAS, MessageContext aMessageContext /*, OutOfTypeSystemData otsd*/) |
| throws AsynchAEException { |
| return register(aCAS, aMessageContext, /*otsd, */ idGenerator.nextId()); |
| } |
| |
| // not used 5/2013 |
| // public CacheEntry register(CAS aCAS, MessageContext aMessageContext, |
| // XmiSerializationSharedData sharedData) throws AsynchAEException { |
| // return register(aCAS, aMessageContext, sharedData, idGenerator.nextId()); |
| // } |
| |
| public CacheEntry register(CAS aCAS, MessageContext aMessageContext, /* OutOfTypeSystemData otsd,*/ |
| String aCasReferenceId) throws AsynchAEException { |
| return registerCacheEntry(aCasReferenceId, new CacheEntry(aCAS, aCasReferenceId, |
| aMessageContext /*, otsd*/)); |
| } |
| |
| // not used 5/2013 |
| // public CacheEntry register(CAS aCAS, MessageContext aMessageContext, |
| // XmiSerializationSharedData sharedData, String aCasReferenceId) throws AsynchAEException { |
| // return registerCacheEntry(aCasReferenceId, new CacheEntry(aCAS, aCasReferenceId, |
| // aMessageContext, sharedData)); |
| // } |
| |
| public CacheEntry register(CAS aCAS, MessageContext aMessageContext, |
| XmiSerializationSharedData sharedData, |
| ReuseInfo compress6ReuseInfo, |
| String aCasReferenceId, Marker marker, |
| boolean acceptsDeltaCas) throws AsynchAEException { |
| return registerCacheEntry(aCasReferenceId, new CacheEntry(aCAS, aCasReferenceId, |
| aMessageContext, sharedData, compress6ReuseInfo, marker, acceptsDeltaCas)); |
| } |
| |
| public CacheEntry register(CAS aCAS, MessageContext aMessageContext, |
| ReuseInfo compress6ReuseInfo, String aCasReferenceId, Marker marker, boolean acceptsDeltaCas) |
| throws AsynchAEException { |
| return registerCacheEntry(aCasReferenceId, new CacheEntry(aCAS, aCasReferenceId, |
| aMessageContext, compress6ReuseInfo, marker, acceptsDeltaCas)); |
| } |
| |
| private synchronized CacheEntry registerCacheEntry(String aCasReferenceId, CacheEntry entry) { |
| cache.put(aCasReferenceId, entry); |
| return entry; |
| } |
| |
| public boolean hasNoSubordinates(String aCasReferenceId) { |
| Iterator<String> it = cache.keySet().iterator(); |
| while (it.hasNext()) { |
| String key = (String) it.next(); |
| CacheEntry entry = (CacheEntry) cache.get(key); |
| if (entry != null && entry.getInputCasReferenceId() != null |
| && entry.getInputCasReferenceId().equals(aCasReferenceId)) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.FINE)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.FINE, CLASS_NAME.getName(), |
| "hasNoSubordinates", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_cache_entry_exists___FINE", |
| new Object[] { entry.getCasReferenceId(), entry.getInputCasReferenceId() }); |
| } |
| return false; |
| } |
| } |
| return true; |
| } |
| |
| public Endpoint getTopAncestorEndpoint(CacheEntry anEntry) throws Exception { |
| if (anEntry == null) { |
| return null; |
| } |
| |
| if (anEntry.getInputCasReferenceId() == null) { |
| return anEntry.getMessageOrigin(); |
| } |
| CacheEntry parentEntry = getCacheEntryForCAS(anEntry.getInputCasReferenceId()); |
| return getTopAncestorEndpoint(parentEntry); |
| } |
| |
| public CacheEntry getTopAncestorCasEntry(CacheEntry anEntry) throws Exception { |
| if (anEntry == null) { |
| return null; |
| } |
| |
| if (anEntry.getInputCasReferenceId() == null) { |
| return anEntry; |
| } |
| CacheEntry parentEntry = getCacheEntryForCAS(anEntry.getInputCasReferenceId()); |
| return getTopAncestorCasEntry(parentEntry); |
| } |
| |
| public synchronized CacheEntry getCacheEntryForCAS(String aCasReferenceId) |
| throws AsynchAEException { |
| CacheEntry casRefEntry = getEntry(aCasReferenceId); |
| if (casRefEntry == null) { |
| throw new AsynchAEException("Cas Not Found In CasManager Cache. CasReferenceId::" |
| + aCasReferenceId + " is Invalid"); |
| } |
| return casRefEntry; |
| } |
| |
| public static class CacheEntry { |
| public static final int FINAL_STATE = 1; |
| |
| private CAS cas; |
| |
| // the following is set to true if the CAS has been created by CAS Multiplier |
| // This flag is used to determine if the CAS should be output to client. |
| private boolean newCas; |
| |
| private String casReferenceId; |
| |
| // This is set if the CAS was produced by a Cas Multiplier |
| private String inputCasReferenceId; |
| |
| // This is Cas Reference Id of the CAS produced by Remote CM |
| private String remoteCMCasReferenceId; |
| |
| private DelegateStats stats; |
| |
| private MessageContext messageAccessor; |
| |
| // private OutOfTypeSystemData otsd = null; |
| |
| private String serializedCas; |
| |
| private String casProducerKey; |
| |
| private Map<String, Endpoint> endpointMap = new HashMap<String, Endpoint>(); |
| |
| private final long timeIn = System.nanoTime(); |
| |
| private Endpoint messageOrigin; |
| |
| private Stack<Endpoint> originStack = new Stack<Endpoint>(); |
| |
| private int highWaterMark; |
| |
| private XmiSerializationSharedData deserSharedData; |
| |
| private ReuseInfo compress6ReuseInfo; |
| |
| private String aggregateProducingTheCas; |
| |
| private long timeWaitingForCAS = 0; |
| |
| private long timeToDeserializeCAS = 0; |
| |
| private long timeToSerializeCAS = 0; |
| |
| private long timeToProcessCAS = 0; |
| |
| private long totalTimeToProcessCAS = 0; |
| |
| private String casMultiplierKey; |
| |
| private boolean sendRequestToFreeCas = true; |
| |
| private boolean aborted = false; |
| |
| private boolean pendingReply = false; |
| |
| private long sequence = 0; |
| |
| private Endpoint freeCasEndpoint; |
| |
| private Marker marker = null; |
| |
| private boolean acceptsDeltaCas = false; |
| |
| private boolean sentDeltaCas = false; |
| |
| // list containing delegates that must be called sequentially. This list |
| // is added to the cache if there are collocated delegates in a parallel |
| // step. Only remote delegates can be part of the parallel step. Any |
| // collocated delegates are removed from the parallel step added to the |
| // list. The delegates in this list will be called sequentially when |
| // all delegates in parallel step respond. |
| private List delayedSingleStepList = null; |
| // shared semaphore that blocks UIMA AS aggregate receiving |
| // thread after a CAS is handed off to the first delegate. |
| // This stops the aggregate's receiving thread from taking |
| // another CAS off the input queue while the current CAS is |
| // still being processed. This semaphore is shared with a |
| // receiving thread which has a reference to the semaphore |
| // via ThreadLocal var |
| private Semaphore threadCompletionSemaphore; |
| |
| private Map<String,List<AnalysisEnginePerformanceMetrics>> delegateMetrics = |
| new ConcurrentHashMap<String, List<AnalysisEnginePerformanceMetrics>>(); |
| |
| public Semaphore getThreadCompletionSemaphore() { |
| return threadCompletionSemaphore; |
| } |
| |
| public void setThreadCompletionSemaphore(Semaphore threadCompletionSemaphore) { |
| this.threadCompletionSemaphore = threadCompletionSemaphore; |
| } |
| // never called 5/2013 was for XCAS |
| // protected CacheEntry(CAS aCas, String aCasReferenceId, MessageContext aMessageAccessor, |
| // OutOfTypeSystemData aotsd) { |
| // this(aCas, aCasReferenceId, aMessageAccessor); |
| // messageAccessor = aMessageAccessor; |
| // } |
| |
| // never called 5/2013 |
| // protected CacheEntry(CAS aCas, String aCasReferenceId, MessageContext aMessageAccessor, |
| // XmiSerializationSharedData sdata) { |
| // this(aCas, aCasReferenceId, aMessageAccessor); |
| // deserSharedData = sdata; |
| // } |
| |
| protected CacheEntry(CAS aCas, String aCasReferenceId, MessageContext aMessageAccessor, |
| XmiSerializationSharedData sdata, Marker aMarker, boolean acceptsDeltaCas) { |
| this(aCas, aCasReferenceId, aMessageAccessor, sdata, null, aMarker, acceptsDeltaCas); |
| } |
| |
| protected CacheEntry(CAS aCas, String aCasReferenceId, MessageContext aMessageAccessor, |
| ReuseInfo compress6ReuseInfo, Marker aMarker, boolean acceptsDeltaCas) { |
| this(aCas, aCasReferenceId, aMessageAccessor, null, compress6ReuseInfo, aMarker, acceptsDeltaCas); |
| } |
| |
| protected CacheEntry(CAS aCas, String aCasReferenceId, MessageContext aMessageAccessor, |
| XmiSerializationSharedData sdata, |
| ReuseInfo compress6ReuseInfo, |
| Marker aMarker, boolean acceptsDeltaCas) { |
| this(aCas, aCasReferenceId, aMessageAccessor); |
| this.deserSharedData = sdata; |
| this.compress6ReuseInfo = compress6ReuseInfo; |
| this.marker = aMarker; |
| this.acceptsDeltaCas = acceptsDeltaCas; |
| } |
| |
| private CacheEntry(CAS aCas, String aCasReferenceId, MessageContext aMessageAccessor) { |
| cas = aCas; |
| messageAccessor = aMessageAccessor; |
| if (aMessageAccessor != null) { |
| messageOrigin = aMessageAccessor.getEndpoint(); |
| } |
| casReferenceId = aCasReferenceId; |
| try { |
| if (aMessageAccessor != null |
| && aMessageAccessor.propertyExists(AsynchAEMessage.CasSequence)) { |
| sequence = aMessageAccessor.getMessageLongProperty(AsynchAEMessage.CasSequence); |
| } |
| } catch (Exception e) { |
| if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) { |
| UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING, getClass().getName(), |
| "CacheEntry", UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, |
| "UIMAEE_exception__WARNING", e); |
| } |
| } |
| } |
| |
| public String getCasReferenceId() { |
| return casReferenceId; |
| } |
| |
| public Map<String, Endpoint> getEndpointMap() { |
| return endpointMap; |
| } |
| |
| public String getInputCasReferenceId() { |
| return inputCasReferenceId; |
| } |
| |
| public void setInputCasReferenceId(String anInputCasReferenceId) { |
| inputCasReferenceId = anInputCasReferenceId; |
| } |
| |
| public void setStat(DelegateStats aStat) { |
| stats = aStat; |
| } |
| |
| public DelegateStats getStat() { |
| return stats; |
| } |
| |
| public void incrementTimeWaitingForCAS(long aTimeWaitingForCAS) { |
| timeWaitingForCAS += aTimeWaitingForCAS; |
| } |
| |
| public void incrementTimeToDeserializeCAS(long aTimeToDeserializeCAS) { |
| timeToDeserializeCAS += aTimeToDeserializeCAS; |
| } |
| |
| public void incrementTimeToProcessCAS(long aTimeToProcessCAS) { |
| timeToProcessCAS += aTimeToProcessCAS; |
| } |
| |
| public void setCasMultiplierKey(String aKey) { |
| casMultiplierKey = aKey; |
| } |
| |
| public String getCasMultiplierKey() { |
| return casMultiplierKey; |
| } |
| |
| public void incrementTimeToSerializeCAS(long aTimeToSerializeCAS) { |
| timeToSerializeCAS += aTimeToSerializeCAS; |
| } |
| |
| public long getTimeWaitingForCAS() { |
| return timeWaitingForCAS; |
| } |
| |
| public long getTimeToDeserializeCAS() { |
| return timeToDeserializeCAS; |
| } |
| |
| public long getTimeToSerializeCAS() { |
| return timeToSerializeCAS; |
| } |
| |
| public Endpoint getMessageOrigin() { |
| // Endpoint ep = (Endpoint)originStack.pop(); |
| return messageOrigin; |
| } |
| |
| public void addOrigin(Endpoint anEndpoint) { |
| originStack.push(anEndpoint); |
| } |
| |
| protected long getStartTime() { |
| return timeIn; |
| } |
| |
| protected void addEndpoint(Endpoint anEndpoint) { |
| endpointMap.put(anEndpoint.getEndpoint(), anEndpoint); |
| } |
| |
| protected Endpoint getEndpoint(String anEndpointName) { |
| return (Endpoint) endpointMap.get(anEndpointName); |
| } |
| |
| protected void deleteEndpoint(String anEndpointName) { |
| if (endpointMap.containsKey(anEndpointName)) { |
| endpointMap.remove(anEndpointName); |
| } |
| } |
| |
| // never called 5/2013 |
| // protected void deleteCAS() { |
| // cas = null; |
| //// otsd = null; |
| // compress6ReuseInfo = null; |
| // } |
| |
| public CAS getCas() { |
| return cas; |
| } |
| |
| // protected void setCas(CAS aCAS, OutOfTypeSystemData aotsd) { |
| // cas = aCAS; |
| // otsd = aotsd; |
| // } |
| |
| // protected void setCas(CAS aCAS) { |
| // cas = aCAS; |
| // } |
| |
| protected void setSerializedCas(String aSerializedCas) { |
| serializedCas = aSerializedCas; |
| } |
| |
| protected String getSerializedCas() { |
| return serializedCas; |
| } |
| |
| protected MessageContext getMessageAccessor() { |
| return messageAccessor; |
| } |
| |
| // not used 5/2013 |
| // public OutOfTypeSystemData getOtsd() { |
| // return otsd; |
| // } |
| |
| public int getHighWaterMark() { |
| return highWaterMark; |
| } |
| |
| public void setHighWaterMark(int aHighWaterMark) { |
| highWaterMark = aHighWaterMark; |
| } |
| |
| public boolean isNewCas() { |
| return newCas; |
| } |
| |
| public void setNewCas(boolean newCas, String producedIn) { |
| this.newCas = newCas; |
| aggregateProducingTheCas = producedIn; |
| } |
| |
| public XmiSerializationSharedData getDeserSharedData() { |
| return deserSharedData; |
| } |
| |
| public void setXmiSerializationData(XmiSerializationSharedData anXmiSerializationData) { |
| deserSharedData = anXmiSerializationData; |
| } |
| |
| public ReuseInfo getCompress6ReuseInfo() { |
| return compress6ReuseInfo; |
| } |
| |
| public void setCompress6ReuseInfo(ReuseInfo compress6ReuseInfo) { |
| this.compress6ReuseInfo = compress6ReuseInfo; |
| } |
| |
| public String getCasProducerAggregateName() { |
| return aggregateProducingTheCas; |
| } |
| |
| public void setCasProducerKey(String aCasProducerKey) { |
| casProducerKey = aCasProducerKey; |
| } |
| |
| public String getCasProducerKey() { |
| return casProducerKey; |
| } |
| |
| public String getRemoteCMCasReferenceId() { |
| return remoteCMCasReferenceId; |
| } |
| |
| public void setRemoteCMCasReferenceId(String remoteCMCasReferenceId) { |
| this.remoteCMCasReferenceId = remoteCMCasReferenceId; |
| } |
| |
| public boolean shouldSendRequestToFreeCas() { |
| return sendRequestToFreeCas; |
| } |
| |
| public void setSendRequestToFreeCas(boolean sendRequestToFreeCas) { |
| this.sendRequestToFreeCas = sendRequestToFreeCas; |
| } |
| |
| public boolean isAborted() { |
| return aborted; |
| } |
| |
| public void setAborted(boolean aborted) { |
| this.aborted = aborted; |
| } |
| |
| public long getTimeToProcessCAS() { |
| return timeToProcessCAS; |
| } |
| |
| public void setTimeToProcessCAS(long aTimeToProcessCAS) { |
| timeToProcessCAS = aTimeToProcessCAS; |
| totalTimeToProcessCAS += aTimeToProcessCAS; |
| } |
| |
| public long getTotalTimeToProcessCAS() { |
| return totalTimeToProcessCAS; |
| } |
| |
| public boolean isPendingReply() { |
| return pendingReply; |
| } |
| |
| public void setPendingReply(boolean pendingReply) { |
| this.pendingReply = pendingReply; |
| } |
| |
| public long getCasSequence() { |
| return sequence; |
| } |
| |
| public void setCasSequence(long sequence) { |
| this.sequence = sequence; |
| } |
| |
| public void setFreeCasEndpoint(Endpoint aFreeCasEndpoint) { |
| freeCasEndpoint = aFreeCasEndpoint; |
| } |
| |
| public Endpoint getFreeCasEndpoint() { |
| return freeCasEndpoint; |
| } |
| |
| public boolean acceptsDeltaCas() { |
| return this.acceptsDeltaCas; |
| } |
| |
| public void setAcceptsDeltaCas(boolean accepts) { |
| this.acceptsDeltaCas = accepts; |
| } |
| |
| public void setSentDeltaCas(boolean sendingDeltaCas) { |
| this.sentDeltaCas = sendingDeltaCas; |
| } |
| |
| public boolean sentDeltaCas() { |
| return this.sentDeltaCas; |
| } |
| |
| public Marker getMarker() { |
| return this.marker; |
| } |
| |
| public void setDelayedSingleStepList(List aList) { |
| delayedSingleStepList = aList; |
| } |
| |
| public List getDelayedSingleStepList() { |
| return delayedSingleStepList; |
| } |
| |
| public void setMarker(Marker mark) { |
| this.marker = mark; |
| } |
| public void addDelegateMetrics(String delegateKey, List<AnalysisEnginePerformanceMetrics> metrics) { |
| addDelegateMetrics(delegateKey, metrics, false); |
| } |
| |
| public void addDelegateMetrics(String delegateKey, List<AnalysisEnginePerformanceMetrics> metrics, boolean remote) { |
| /* |
| System.out.println("................ Adding metrics for delegate:"+delegateKey+" Metrics Size:"+metrics.size()+" CAS:"+getCasReferenceId()); |
| if ( remote && delegateMetrics.containsKey(delegateKey)) { |
| // List<AnalysisEnginePerformanceMetrics> delegateMetrics = |
| // delegateMetrics.get(delegateKey); |
| List<AnalysisEnginePerformanceMetrics> currentMetrics = |
| delegateMetrics.get(delegateKey); |
| for( AnalysisEnginePerformanceMetrics rm : metrics) { |
| for( AnalysisEnginePerformanceMetrics cm : currentMetrics ) { |
| if ( cm.getUniqueName().equals(rm.getUniqueName())) { |
| AnalysisEnginePerformanceMetrics apm = |
| new AnalysisEnginePerformanceMetrics(rm.getName(),rm.getUniqueName(),rm.getAnalysisTime(),cm.getNumProcessed()+rm.getNumProcessed()); |
| currentMetrics.remove(cm); |
| currentMetrics.add(apm); |
| break; |
| } |
| } |
| } |
| } else { |
| delegateMetrics.put(delegateKey, metrics); |
| } |
| */ |
| delegateMetrics.put(delegateKey, metrics); |
| } |
| public List<AnalysisEnginePerformanceMetrics> getDelegateMetrics() { |
| List<AnalysisEnginePerformanceMetrics> metrics = new ArrayList<AnalysisEnginePerformanceMetrics>(); |
| for( Entry<String,List<AnalysisEnginePerformanceMetrics>> dm : delegateMetrics.entrySet()) { |
| for(AnalysisEnginePerformanceMetrics metric : dm.getValue()) { |
| metrics.add(metric); |
| } |
| } |
| return metrics; |
| } |
| } |
| |
| } |
