You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@manifoldcf.apache.org by kw...@apache.org on 2013/11/22 14:48:48 UTC
svn commit: r1544538 - in
/manifoldcf/branches/CONNECTORS-781/framework/agents/src/main/java/org/apache/manifoldcf/agents:
AgentRun.java AgentStop.java
Author: kwright
Date: Fri Nov 22 13:48:48 2013
New Revision: 1544538
URL: http://svn.apache.org/r1544538
Log:
Allow multiple agents processes to run, and make AgentStop be synchronous (and stop all of the agents processes)
Modified:
manifoldcf/branches/CONNECTORS-781/framework/agents/src/main/java/org/apache/manifoldcf/agents/AgentRun.java
manifoldcf/branches/CONNECTORS-781/framework/agents/src/main/java/org/apache/manifoldcf/agents/AgentStop.java
Modified: manifoldcf/branches/CONNECTORS-781/framework/agents/src/main/java/org/apache/manifoldcf/agents/AgentRun.java
URL: http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-781/framework/agents/src/main/java/org/apache/manifoldcf/agents/AgentRun.java?rev=1544538&r1=1544537&r2=1544538&view=diff
==============================================================================
--- manifoldcf/branches/CONNECTORS-781/framework/agents/src/main/java/org/apache/manifoldcf/agents/AgentRun.java (original)
+++ manifoldcf/branches/CONNECTORS-781/framework/agents/src/main/java/org/apache/manifoldcf/agents/AgentRun.java Fri Nov 22 13:48:48 2013
@@ -28,7 +28,7 @@ public class AgentRun extends BaseAgents
{
public static final String _rcsid = "@(#)$Id: AgentRun.java 988245 2010-08-23 18:39:35Z kwright $";
- public static final String agentInUseSignal = "_AGENTINUSE_";
+ public static final String agentServiceType = "AGENT";
public static final String agentShutdownSignal = "_AGENTRUN_";
public AgentRun()
@@ -37,22 +37,29 @@ public class AgentRun extends BaseAgents
protected void doExecute(IThreadContext tc) throws ManifoldCFException
{
- ILockManager lockManager = LockManagerFactory.make(tc);
- // Agent already in use?
- if (lockManager.checkGlobalFlag(agentInUseSignal))
- {
- System.err.println("Agent already in use");
- System.exit(1);
- }
-
- ManifoldCF.addShutdownHook(new AgentRunShutdownRunner());
+ // Note well:
+ // As part of CONNECTORS-781, multiple agents processes are now permitted, provided
+ // that a truly global lock manager implementation is used. This implementation thus does the
+ // following:
+ // (1) Register the agent, and begin its execution
+ // (2) Periodically check for any new registered IAgent implementations
+ // (3) Await a shutdown signal
+ // (4) If exit signal seen, exit active block
+ // (5) Trap JVM exit to be sure we exit active block no matter what
+ // (This latter option requires the ability to exit active blocks from different ILockManager instances)
+ //
+ // Note well that the agents shutdown signal is NEVER modified by this code; it will be set/cleared by
+ // AgentStop only, and AgentStop will wait until all services become inactive before exiting.
- // Set the agents in use signal.
- lockManager.setGlobalFlag(agentInUseSignal);
+ ILockManager lockManager = LockManagerFactory.make(tc);
+ // Don't come up at all if shutdown signal in force
+ if (lockManager.checkGlobalFlag(agentShutdownSignal))
+ return;
+ lockManager.registerServiceBeginServiceActivity(agentServiceType, ManifoldCF.getProcessID());
try
{
- // Clear the agents shutdown signal.
- lockManager.clearGlobalFlag(agentShutdownSignal);
+ ManifoldCF.addShutdownHook(new AgentRunShutdownRunner());
+
Logging.root.info("Running...");
while (true)
{
@@ -65,7 +72,7 @@ public class AgentRun extends BaseAgents
try
{
- ManifoldCF.sleep(5000);
+ ManifoldCF.sleep(5000L);
}
catch (InterruptedException e)
{
@@ -81,7 +88,9 @@ public class AgentRun extends BaseAgents
}
finally
{
- lockManager.clearGlobalFlag(agentInUseSignal);
+ // Exit service
+ // This is a courtesy; some lock managers (i.e. ZooKeeper) manage to do this anyway
+ lockManager.endServiceActivity(agentServiceType, ManifoldCF.getProcessID());
}
}
@@ -120,7 +129,18 @@ public class AgentRun extends BaseAgents
{
IThreadContext tc = ThreadContextFactory.make();
ILockManager lockManager = LockManagerFactory.make(tc);
- lockManager.clearGlobalFlag(agentInUseSignal);
+ // We can blast the active flag off here; we may have already exited though and an exception will
+ // therefore be thrown.
+ try
+ {
+ lockManager.endServiceActivity(agentServiceType, ManifoldCF.getProcessID());
+ }
+ catch (ManifoldCFException e)
+ {
+ if (e.getErrorCode() == ManifoldCFException.INTERRUPTED)
+ throw e;
+ // Otherwise eat the exception
+ }
}
}
Modified: manifoldcf/branches/CONNECTORS-781/framework/agents/src/main/java/org/apache/manifoldcf/agents/AgentStop.java
URL: http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-781/framework/agents/src/main/java/org/apache/manifoldcf/agents/AgentStop.java?rev=1544538&r1=1544537&r2=1544538&view=diff
==============================================================================
--- manifoldcf/branches/CONNECTORS-781/framework/agents/src/main/java/org/apache/manifoldcf/agents/AgentStop.java (original)
+++ manifoldcf/branches/CONNECTORS-781/framework/agents/src/main/java/org/apache/manifoldcf/agents/AgentStop.java Fri Nov 22 13:48:48 2013
@@ -34,9 +34,44 @@ public class AgentStop extends BaseAgent
protected void doExecute(IThreadContext tc) throws ManifoldCFException
{
+ // As part of the work for CONNECTORS-781, this method is now synchronous.
+ // We assert the shutdown signal, and then wait until all active services have shut down.
ILockManager lockManager = LockManagerFactory.make(tc);
lockManager.setGlobalFlag(AgentRun.agentShutdownSignal);
- Logging.root.info("Shutdown signal sent");
+ try
+ {
+ Logging.root.info("Shutdown signal sent");
+ while (true)
+ {
+ // Check to see if services are down yet
+ String[] agents = lockManager.getRegisteredServices(AgentRun.agentServiceType);
+ boolean alive = false;
+ for (String agent : agents)
+ {
+ if (lockManager.checkServiceActive(AgentRun.agentServiceType, agent))
+ {
+ alive = true;
+ break;
+ }
+ }
+ if (!alive)
+ break;
+ try
+ {
+ ManifoldCF.sleep(1000L);
+ }
+ catch (InterruptedException e)
+ {
+ throw new ManifoldCFException(e.getMessage(),e,ManifoldCFException.INTERRUPTED);
+ }
+ }
+ Logging.root.info("All agents shut down");
+ }
+ finally
+ {
+ // Clear shutdown signal
+ lockManager.clearGlobalFlag(AgentRun.agentShutdownSignal);
+ }
}