tags/uima-as-2.3.1-rc2/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/LocalCache.java - uima-async-scaleout - Git at Google

 /*
  * 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.util.ArrayList;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;

 import org.apache.uima.UIMAFramework;
 import org.apache.uima.aae.UIMAEE_Constants;
 import org.apache.uima.aae.InProcessCache.CacheEntry;
 import org.apache.uima.aae.delegate.Delegate;
 import org.apache.uima.flow.FinalStep;
 import org.apache.uima.util.Level;

 public class LocalCache extends ConcurrentHashMap<String, LocalCache.CasStateEntry> {
   private static final long serialVersionUID = 1L;

   private static final Class CLASS_NAME = LocalCache.class;

   private AnalysisEngineController controller;

   public LocalCache(AnalysisEngineController aController) {
     controller = aController;
   }

   public CasStateEntry createCasStateEntry(String aCasReferenceId) {
     CasStateEntry entry = new CasStateEntry(aCasReferenceId);
     super.put(aCasReferenceId, entry);
     return entry;
   }

   public CasStateEntry lookupEntry(String aCasReferenceId) {
     if (super.containsKey(aCasReferenceId)) {
       return super.get(aCasReferenceId);
     }
     return null;
   }

   public String lookupInputCasReferenceId(String aCasReferenceId) {
     String parentCasReferenceId = null;
     if (this.containsKey(aCasReferenceId)) {
       CasStateEntry entry = (CasStateEntry) get(aCasReferenceId);
       if (entry != null && entry.isSubordinate()) {
         // recursively call each parent until we get to the top of the
         // Cas hierarchy
         parentCasReferenceId = lookupInputCasReferenceId(entry.getInputCasReferenceId());
       } else {
         return aCasReferenceId;
       }
     }
     return parentCasReferenceId;
   }

   public String lookupInputCasReferenceId(CasStateEntry entry) {
     String parentCasReferenceId = null;
     if (entry.isSubordinate()) {
       // recursively call each parent until we get to the top of the
       // Cas hierarchy
       parentCasReferenceId = lookupInputCasReferenceId((CasStateEntry) get(entry
               .getInputCasReferenceId()));
     } else {
       return entry.getCasReferenceId();
     }
     return parentCasReferenceId;
   }
   public void dumpContents() {
     dumpContents(false);
   }
   public synchronized void dumpContents(boolean dump2Stdout) {
     int count = 0;
     if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
       StringBuffer sb = new StringBuffer("\n");

       for (Iterator it = entrySet().iterator(); it.hasNext();) {
         Map.Entry entry = (Map.Entry) it.next();
         CasStateEntry casStateEntry = (CasStateEntry) entry.getValue();
         if (casStateEntry == null) {
           continue;
         }
         count++;
         if (casStateEntry.isSubordinate()) {
           sb.append(entry.getKey() + " Number Of Child CASes In Play:"
                   + casStateEntry.getSubordinateCasInPlayCount() + " Parent CAS id:"
                   + casStateEntry.getInputCasReferenceId());
         } else {
           sb.append(entry.getKey() + " *** Input CAS. Number Of Child CASes In Play:"
                   + casStateEntry.getSubordinateCasInPlayCount());
         }
         if (casStateEntry.isWaitingForRelease()) {
           sb.append(" <<< Reached Final State in Controller:" + controller.getComponentName());
         }
         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[] { controller.getComponentName(), count, sb.toString() });
       if ( dump2Stdout ) {
         System.out.println(sb.toString());
       }
       sb.setLength(0);
     } else if (UIMAFramework.getLogger().isLoggable(Level.FINE)) {
       int inFinalState = 0;

       for (Iterator it = entrySet().iterator(); it.hasNext();) {
         Map.Entry entry = (Map.Entry) it.next();
         CasStateEntry casStateEntry = (CasStateEntry) entry.getValue();

         count++;
         if (casStateEntry != null && casStateEntry.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[] { controller.getComponentName(), count, inFinalState });

     }

   }

   public synchronized void remove(String aCasReferenceId) {
     if (aCasReferenceId != null && 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 });
       }
       super.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 });
       }
     }
   }

   public CasStateEntry getTopCasAncestor(String casReferenceId) throws Exception {
     if (!containsKey(casReferenceId)) {
       return null;
     }
     CasStateEntry casStateEntry = lookupEntry(casReferenceId);
     if (casStateEntry.isSubordinate()) {
       // Recurse until the top CAS reference Id is found
       return getTopCasAncestor(casStateEntry.getInputCasReferenceId());
     }
     // Return the top ancestor CAS id
     return casStateEntry;
   }

   public static class CasStateEntry {

     private String casReferenceId;

     private volatile boolean waitingForRealease;

     private volatile boolean pendingReply;

     private volatile boolean subordinateCAS;

     private volatile boolean replyReceived;

     private volatile boolean failed;

     private volatile boolean dropped;

     private List<Throwable> exceptionList = new ArrayList<Throwable>();

     private FinalStep step;

     private int state;

     private int subordinateCasInPlayCount;

     private Object childCountMux = new Object();

     private String inputCasReferenceId;

     private int numberOfParallelDelegates = 1;

     private Delegate lastDelegate = null;

     private int howManyDelegatesResponded = 0;

     private Endpoint freeCasNotificationEndpoint;

     private volatile boolean deliveryToClientFailed;

     public boolean deliveryToClientFailed() {
 		return deliveryToClientFailed;
 	}

 	public void setDeliveryToClientFailed() {
 		this.deliveryToClientFailed = true;
 	}

 	public boolean isDropped() {
       return dropped;
     }

     public void setDropped(boolean dropped) {
       this.dropped = dropped;
     }

     public Endpoint getFreeCasNotificationEndpoint() {
       return freeCasNotificationEndpoint;
     }

     public void setFreeCasNotificationEndpoint(Endpoint freeCasNotificationEndpoint) {
       this.freeCasNotificationEndpoint = freeCasNotificationEndpoint;
     }

     public CasStateEntry(String aCasReferenceId) {
       casReferenceId = aCasReferenceId;
     }

     public void setLastDelegate(Delegate aDelegate) {
       lastDelegate = aDelegate;
     }

     public Delegate getLastDelegate() {
       return lastDelegate;
     }

     public String getCasReferenceId() {
       return casReferenceId;
     }

     public String getInputCasReferenceId() {
       return inputCasReferenceId;
     }

     public void setInputCasReferenceId(String anInputCasReferenceId) {
       inputCasReferenceId = anInputCasReferenceId;
       subordinateCAS = true;
     }

     public void setWaitingForRelease(boolean flag) {
       waitingForRealease = flag;
     }

     public boolean isWaitingForRelease() {
       return waitingForRealease;
     }

     public void setFinalStep(FinalStep step) {
       this.step = step;
     }

     public FinalStep getFinalStep() {
       return step;
     }

     public int getState() {
       return state;
     }

     public void setState(int aState) {
       state = aState;
     }

     public boolean isSubordinate() {
       return subordinateCAS;
     }

     public int getSubordinateCasInPlayCount() {
       synchronized (childCountMux) {
         return subordinateCasInPlayCount;
       }
     }

     public void incrementSubordinateCasInPlayCount() {
       synchronized (childCountMux) {
         subordinateCasInPlayCount++;
       }
     }

     public int decrementSubordinateCasInPlayCount() {
       synchronized (childCountMux) {
         if (subordinateCasInPlayCount > 0) {
           subordinateCasInPlayCount--;
         }
         return subordinateCasInPlayCount;
       }
     }

     public boolean isPendingReply() {
       return pendingReply;
     }

     public void setPendingReply(boolean pendingReply) {
       this.pendingReply = pendingReply;
     }

     public void setReplyReceived() {
       replyReceived = true;
     }

     public boolean isReplyReceived() {
       return replyReceived;
     }
     public void resetReplyReceived() {
       replyReceived = false;
     }
     public synchronized void incrementHowManyDelegatesResponded() {
       if (howManyDelegatesResponded < numberOfParallelDelegates) {
         howManyDelegatesResponded++;
       }
     }

     public synchronized int howManyDelegatesResponded() {
       return howManyDelegatesResponded;
     }

     public synchronized void resetDelegateResponded() {
       howManyDelegatesResponded = 0;
     }

     public void setNumberOfParallelDelegates(int aNumberOfParallelDelegates) {
       numberOfParallelDelegates = aNumberOfParallelDelegates;
     }

     public int getNumberOfParallelDelegates() {
       return numberOfParallelDelegates;
     }

     public boolean isFailed() {
       return failed;
     }

     public void setFailed() {
       this.failed = true;
     }

     public void addThrowable(Throwable t) {
       exceptionList.add(t);
     }

     public List<Throwable> getErrors() {
       return exceptionList;
     }
   }
 }
	/*
	* 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.util.ArrayList;
	import java.util.Iterator;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;
	import java.util.concurrent.ConcurrentHashMap;

	import org.apache.uima.UIMAFramework;
	import org.apache.uima.aae.UIMAEE_Constants;
	import org.apache.uima.aae.InProcessCache.CacheEntry;
	import org.apache.uima.aae.delegate.Delegate;
	import org.apache.uima.flow.FinalStep;
	import org.apache.uima.util.Level;

	public class LocalCache extends ConcurrentHashMap<String, LocalCache.CasStateEntry> {
	private static final long serialVersionUID = 1L;

	private static final Class CLASS_NAME = LocalCache.class;

	private AnalysisEngineController controller;

	public LocalCache(AnalysisEngineController aController) {
	controller = aController;
	}

	public CasStateEntry createCasStateEntry(String aCasReferenceId) {
	CasStateEntry entry = new CasStateEntry(aCasReferenceId);
	super.put(aCasReferenceId, entry);
	return entry;
	}

	public CasStateEntry lookupEntry(String aCasReferenceId) {
	if (super.containsKey(aCasReferenceId)) {
	return super.get(aCasReferenceId);
	}
	return null;
	}

	public String lookupInputCasReferenceId(String aCasReferenceId) {
	String parentCasReferenceId = null;
	if (this.containsKey(aCasReferenceId)) {
	CasStateEntry entry = (CasStateEntry) get(aCasReferenceId);
	if (entry != null && entry.isSubordinate()) {
	// recursively call each parent until we get to the top of the
	// Cas hierarchy
	parentCasReferenceId = lookupInputCasReferenceId(entry.getInputCasReferenceId());
	} else {
	return aCasReferenceId;
	}
	}
	return parentCasReferenceId;
	}

	public String lookupInputCasReferenceId(CasStateEntry entry) {
	String parentCasReferenceId = null;
	if (entry.isSubordinate()) {
	// recursively call each parent until we get to the top of the
	// Cas hierarchy
	parentCasReferenceId = lookupInputCasReferenceId((CasStateEntry) get(entry
	.getInputCasReferenceId()));
	} else {
	return entry.getCasReferenceId();
	}
	return parentCasReferenceId;
	}
	public void dumpContents() {
	dumpContents(false);
	}
	public synchronized void dumpContents(boolean dump2Stdout) {
	int count = 0;
	if (UIMAFramework.getLogger().isLoggable(Level.FINEST)) {
	StringBuffer sb = new StringBuffer("\n");

	for (Iterator it = entrySet().iterator(); it.hasNext();) {
	Map.Entry entry = (Map.Entry) it.next();
	CasStateEntry casStateEntry = (CasStateEntry) entry.getValue();
	if (casStateEntry == null) {
	continue;
	}
	count++;
	if (casStateEntry.isSubordinate()) {
	sb.append(entry.getKey() + " Number Of Child CASes In Play:"
	+ casStateEntry.getSubordinateCasInPlayCount() + " Parent CAS id:"
	+ casStateEntry.getInputCasReferenceId());
	} else {
	sb.append(entry.getKey() + " *** Input CAS. Number Of Child CASes In Play:"
	+ casStateEntry.getSubordinateCasInPlayCount());
	}
	if (casStateEntry.isWaitingForRelease()) {
	sb.append(" <<< Reached Final State in Controller:" + controller.getComponentName());
	}
	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[] { controller.getComponentName(), count, sb.toString() });
	if ( dump2Stdout ) {
	System.out.println(sb.toString());
	}
	sb.setLength(0);
	} else if (UIMAFramework.getLogger().isLoggable(Level.FINE)) {
	int inFinalState = 0;

	for (Iterator it = entrySet().iterator(); it.hasNext();) {
	Map.Entry entry = (Map.Entry) it.next();
	CasStateEntry casStateEntry = (CasStateEntry) entry.getValue();

	count++;
	if (casStateEntry != null && casStateEntry.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[] { controller.getComponentName(), count, inFinalState });

	}

	}

	public synchronized void remove(String aCasReferenceId) {
	if (aCasReferenceId != null && 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 });
	}
	super.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 });
	}
	}
	}

	public CasStateEntry getTopCasAncestor(String casReferenceId) throws Exception {
	if (!containsKey(casReferenceId)) {
	return null;
	}
	CasStateEntry casStateEntry = lookupEntry(casReferenceId);
	if (casStateEntry.isSubordinate()) {
	// Recurse until the top CAS reference Id is found
	return getTopCasAncestor(casStateEntry.getInputCasReferenceId());
	}
	// Return the top ancestor CAS id
	return casStateEntry;
	}

	public static class CasStateEntry {

	private String casReferenceId;

	private volatile boolean waitingForRealease;

	private volatile boolean pendingReply;

	private volatile boolean subordinateCAS;

	private volatile boolean replyReceived;

	private volatile boolean failed;

	private volatile boolean dropped;

	private List<Throwable> exceptionList = new ArrayList<Throwable>();

	private FinalStep step;

	private int state;

	private int subordinateCasInPlayCount;

	private Object childCountMux = new Object();

	private String inputCasReferenceId;

	private int numberOfParallelDelegates = 1;

	private Delegate lastDelegate = null;

	private int howManyDelegatesResponded = 0;

	private Endpoint freeCasNotificationEndpoint;

	private volatile boolean deliveryToClientFailed;

	public boolean deliveryToClientFailed() {
	return deliveryToClientFailed;
	}

	public void setDeliveryToClientFailed() {
	this.deliveryToClientFailed = true;
	}

	public boolean isDropped() {
	return dropped;
	}

	public void setDropped(boolean dropped) {
	this.dropped = dropped;
	}

	public Endpoint getFreeCasNotificationEndpoint() {
	return freeCasNotificationEndpoint;
	}

	public void setFreeCasNotificationEndpoint(Endpoint freeCasNotificationEndpoint) {
	this.freeCasNotificationEndpoint = freeCasNotificationEndpoint;
	}

	public CasStateEntry(String aCasReferenceId) {
	casReferenceId = aCasReferenceId;
	}

	public void setLastDelegate(Delegate aDelegate) {
	lastDelegate = aDelegate;
	}

	public Delegate getLastDelegate() {
	return lastDelegate;
	}

	public String getCasReferenceId() {
	return casReferenceId;
	}

	public String getInputCasReferenceId() {
	return inputCasReferenceId;
	}

	public void setInputCasReferenceId(String anInputCasReferenceId) {
	inputCasReferenceId = anInputCasReferenceId;
	subordinateCAS = true;
	}

	public void setWaitingForRelease(boolean flag) {
	waitingForRealease = flag;
	}

	public boolean isWaitingForRelease() {
	return waitingForRealease;
	}

	public void setFinalStep(FinalStep step) {
	this.step = step;
	}

	public FinalStep getFinalStep() {
	return step;
	}

	public int getState() {
	return state;
	}

	public void setState(int aState) {
	state = aState;
	}

	public boolean isSubordinate() {
	return subordinateCAS;
	}

	public int getSubordinateCasInPlayCount() {
	synchronized (childCountMux) {
	return subordinateCasInPlayCount;
	}
	}

	public void incrementSubordinateCasInPlayCount() {
	synchronized (childCountMux) {
	subordinateCasInPlayCount++;
	}
	}

	public int decrementSubordinateCasInPlayCount() {
	synchronized (childCountMux) {
	if (subordinateCasInPlayCount > 0) {
	subordinateCasInPlayCount--;
	}
	return subordinateCasInPlayCount;
	}
	}

	public boolean isPendingReply() {
	return pendingReply;
	}

	public void setPendingReply(boolean pendingReply) {
	this.pendingReply = pendingReply;
	}

	public void setReplyReceived() {
	replyReceived = true;
	}

	public boolean isReplyReceived() {
	return replyReceived;
	}
	public void resetReplyReceived() {
	replyReceived = false;
	}
	public synchronized void incrementHowManyDelegatesResponded() {
	if (howManyDelegatesResponded < numberOfParallelDelegates) {
	howManyDelegatesResponded++;
	}
	}

	public synchronized int howManyDelegatesResponded() {
	return howManyDelegatesResponded;
	}

	public synchronized void resetDelegateResponded() {
	howManyDelegatesResponded = 0;
	}

	public void setNumberOfParallelDelegates(int aNumberOfParallelDelegates) {
	numberOfParallelDelegates = aNumberOfParallelDelegates;
	}

	public int getNumberOfParallelDelegates() {
	return numberOfParallelDelegates;
	}

	public boolean isFailed() {
	return failed;
	}

	public void setFailed() {
	this.failed = true;
	}

	public void addThrowable(Throwable t) {
	exceptionList.add(t);
	}

	public List<Throwable> getErrors() {
	return exceptionList;
	}
	}
	}