Author: cwiklik
Date: Sat Nov 2 17:21:10 2013
New Revision: 1538234
URL: http://svn.apache.org/r1538234
Log:
UIMA-3383 Refactored code related to handling of Quiesce request. Both 'q' and
CTRL-C initiated quiesce now wait for notification from InProcessCache when it
becomes empty. Only when the cache is empty the process is allowed to exit.
Modified:
uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/service/UIMA_Service.java
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/InProcessCache.java
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AnalysisEngineInstancePoolWithThreadAffinity.java
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java
uima/uima-as/trunk/uimaj-as-jms/src/main/resources/jms_adapter_messages.properties
Modified:
uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/service/UIMA_Service.java
URL:
http://svn.apache.org/viewvc/uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/service/UIMA_Service.java?rev=1538234&r1=1538233&r2=1538234&view=diff
==============================================================================
---
uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/service/UIMA_Service.java
(original)
+++
uima/uima-as/trunk/uimaj-as-activemq/src/main/java/org/apache/uima/adapter/jms/service/UIMA_Service.java
Sat Nov 2 17:21:10 2013
@@ -23,12 +23,8 @@ import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InvalidClassException;
-import java.net.Socket;
import org.apache.uima.UIMAFramework;
-import org.apache.uima.UimaContext;
-import org.apache.uima.UimaContextAdmin;
-import org.apache.uima.aae.UimaASApplicationEvent;
import org.apache.uima.aae.UimaASApplicationExitEvent;
import org.apache.uima.aae.UimaAsVersion;
import org.apache.uima.aae.controller.AnalysisEngineController;
@@ -395,7 +391,7 @@ public class UIMA_Service implements App
} else if ( event instanceof UimaASApplicationExitEvent ) {
System.out.println("Service Wrapper Received UimaASApplicationEvent.
Message:"+event.getSource());
} else if ( event instanceof ContextStoppedEvent ){ // Spring has been
shutdown
- System.exit(0);
+
}
}
@@ -490,7 +486,6 @@ public class UIMA_Service implements App
}
static class ServiceShutdownHook extends Thread {
-
public SpringContainerDeployer serviceDeployer;
public ServiceShutdownHook(SpringContainerDeployer serviceDeployer) {
@@ -501,18 +496,19 @@ public class UIMA_Service implements App
try {
AnalysisEngineController topLevelController =
serviceDeployer.getTopLevelController();
if (topLevelController != null && !topLevelController.isStopped() ) {
- UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING,
CLASS_NAME.getName(),
+ UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING,
CLASS_NAME.getName(),
"run", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
"UIMAJMS_caught_signal__INFO", new Object[] {
topLevelController.getComponentName() });
- serviceDeployer.undeploy(SpringContainerDeployer.QUIESCE_AND_STOP);
- }
+ serviceDeployer.undeploy(SpringContainerDeployer.QUIESCE_AND_STOP);
+ }
} catch( Exception e) {
if (UIMAFramework.getLogger(CLASS_NAME).isLoggable(Level.WARNING)) {
UIMAFramework.getLogger(CLASS_NAME).logrb(Level.WARNING,
CLASS_NAME.getName(),
"run", JmsConstants.JMS_LOG_RESOURCE_BUNDLE,
"UIMAJMS_exception__WARNING", e);
}
- }
+ }
}
+
}
}
Modified:
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/InProcessCache.java
URL:
http://svn.apache.org/viewvc/uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/InProcessCache.java?rev=1538234&r1=1538233&r2=1538234&view=diff
==============================================================================
---
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/InProcessCache.java
(original)
+++
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/InProcessCache.java
Sat Nov 2 17:21:10 2013
@@ -31,6 +31,7 @@ import java.util.concurrent.ConcurrentHa
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;
@@ -64,6 +65,18 @@ public class InProcessCache implements I
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);
}
@@ -271,7 +284,10 @@ public class InProcessCache implements I
((EventSubscriber) callbackListeners.get(i)).onCacheEmpty();
}
}
-
+ if (cache.size() == 0 && controller != null ) {
+ // unblock the controller waiting in quiesceAndStop
+ controller.notifyOnCacheEmpty();
+ }
}
// never called 5/2013
Modified:
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AnalysisEngineInstancePoolWithThreadAffinity.java
URL:
http://svn.apache.org/viewvc/uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AnalysisEngineInstancePoolWithThreadAffinity.java?rev=1538234&r1=1538233&r2=1538234&view=diff
==============================================================================
---
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AnalysisEngineInstancePoolWithThreadAffinity.java
(original)
+++
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/AnalysisEngineInstancePoolWithThreadAffinity.java
Sat Nov 2 17:21:10 2013
@@ -50,7 +50,6 @@ public class AnalysisEngineInstancePoolW
lock.acquireUninterruptibly();
// Call destroy() on AE on checkin if the UIMA AS process is
in quiesce mode
if ( destroyAEInstanceIfFree ) {
- System.out.println("...........
AnalysisEngineInstancePool.checkin() -
Thread:"+Thread.currentThread().getId()+" calling destroy() on AE checkin");
anAnalysisEngine.destroy();
} else {
aeInstanceMap.put(Thread.currentThread().getId(),
anAnalysisEngine);
Modified:
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java
URL:
http://svn.apache.org/viewvc/uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java?rev=1538234&r1=1538233&r2=1538234&view=diff
==============================================================================
---
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java
(original)
+++
uima/uima-as/trunk/uimaj-as-core/src/main/java/org/apache/uima/aae/controller/BaseAnalysisEngineController.java
Sat Nov 2 17:21:10 2013
@@ -105,6 +105,9 @@ public abstract class BaseAnalysisEngine
private static final String JMS_PROVIDER_HOME = "ACTIVEMQ_HOME";
public static enum ServiceState { INITIALIZING, RUNNING, DISABLED, STOPPING,
FAILED };
public static final boolean NO_RECOVERY = true;
+ // Semaphore use only when quiesceAndStop is called
+ // When the cache becomes empty the semaphore is released.
+ private Semaphore quiesceSemaphore = new Semaphore(0);
protected ServiceState currentState = ServiceState.INITIALIZING;
@@ -1895,7 +1898,6 @@ public abstract class BaseAnalysisEngine
}
}
}
-
if (daemonServiceExecutor != null) {
daemonServiceExecutor.shutdown();
}
@@ -1907,8 +1909,8 @@ public abstract class BaseAnalysisEngine
if (getOutputChannel() != null) {
getOutputChannel().cancelTimers();
}
+
if (this instanceof PrimitiveAnalysisEngineController) {
-
getControllerLatch().release();
// Stops the input channel of this service
stopInputChannels(InputChannel.CloseAllChannels, shutdownNow);
@@ -1946,7 +1948,6 @@ public abstract class BaseAnalysisEngine
}
}
}
-
getInProcessCache().releaseAllCASes();
getLocalCache().clear();
releasedAllCASes = true;
@@ -2019,7 +2020,13 @@ public abstract class BaseAnalysisEngine
super.destroy();
}
-
+ /**
+ * This method is called by InProcessCache when the cache becomes empty
while the controller
+ * is in Quiesce mode.
+ */
+ public void notifyOnCacheEmpty() {
+ quiesceSemaphore.release();
+ }
/**
* Stops input channel(s) and waits for CASes still in play to complete
processing. When the
* InProcessCache becomes empty, initiate the service shutdown.
@@ -2030,6 +2037,10 @@ public abstract class BaseAnalysisEngine
UIMAEE_Constants.JMS_LOG_RESOURCE_BUNDLE, "UIMAEE_stop__INFO",
new Object[] { getComponentName() });
}
+ System.out.println("Quiescing UIMA-AS Service. Remaining Number of CASes
to Process:"+getInProcessCache().getSize());
+ // Register callback when the inProcessCache becomes empty
+ getInProcessCache().registerController(this);
+
if (!isStopped() && !callbackReceived) {
getControllerLatch().release();
// To support orderly shutdown, the service must first stop its input
channel and
@@ -2065,12 +2076,22 @@ public abstract class BaseAnalysisEngine
}
}
stopInputChannels(InputChannel.InputChannels, true);
- // close JMS connection
- stop(false); // wait for any remaining CASes in flight to finish
+
+ try {
+ if ( !getInProcessCache().isEmpty() ) {
+ // acquire semaphore and wait for the InProcessCache to call
notifyOnCacheEmpty()
+ // on this controller when the cache becomes empty.
+ quiesceSemaphore.acquire();
+ }
+ System.out.println("UIMA-AS Service is Stopping, All CASes Have Been
Processed");
+ } catch( InterruptedException e) {
+
+ }
+ stop(false);
}
}
}
-
+
protected void stopDelegateTimers() {
Iterator<Delegate> it = delegates.iterator();
while (it.hasNext()) {
Modified:
uima/uima-as/trunk/uimaj-as-jms/src/main/resources/jms_adapter_messages.properties
URL:
http://svn.apache.org/viewvc/uima/uima-as/trunk/uimaj-as-jms/src/main/resources/jms_adapter_messages.properties?rev=1538234&r1=1538233&r2=1538234&view=diff
==============================================================================
---
uima/uima-as/trunk/uimaj-as-jms/src/main/resources/jms_adapter_messages.properties
(original)
+++
uima/uima-as/trunk/uimaj-as-jms/src/main/resources/jms_adapter_messages.properties
Sat Nov 2 17:21:10 2013
@@ -238,3 +238,7 @@ UIMAJMS_start_inactivity_monitor__INFO=C
UIMAJMS_discard_msg__INFO=Component: {0} Message From: {1} IP: {2} Message
Type: {3} Command: {4} CasId: {5}
UIMAJMS_show_stats__INFO=Component: {0} Stats for CasId:{1}\n\t{2}
UIMAJMS_session_open__FINE=Service: {0} Opened Session To Queue:{1} Broker:{2}
+UIMAJMS_invalid_broker_url__WARNING=Unexpected State - the Broker URL is NULL
+UIMAJMS_shared_connections__INFO=Shared Connection: {0}
+UIMAJMS_retrying_jms_connection__WARNING=Connection Invalid - Retrying
Connection to Broker:{0} Every 5 Seconds
+
