AfflictionNetworks/narya
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Bureaucracy

Browse Source 
* Fixed signature of Receiver methods
* Implemented create/destroy mechanism and bootstrap request in BureauDirector
* Added BureauContext class to expose bureau information on top of usual 
  presents stuff
* Added BureauContext implementation in BureauClient
* Added code to kick off the director when the client logs on
* Added credentials for a bureau logon, which is just a token passed to 
  command line launcher
* Added concept of bureau types to distinguish between executable and command 
  line arguments; includes interface BureauRegistry.Launcher
* Implemented backend for the registry


git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@5027 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
Jamie Doornbos
2008-04-28 18:34:57 +00:00
parent b10cc3643d
commit 74dac11f97
11 changed files with 724 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/java/com/threerings/bureau/server/BureauRegistry.java
+377 -33
View File
@@ -23,89 +23,433 @@ package com.threerings.bureau.server;
import java.util.Map;
import java.util.Set;
import java.util.HashMap;
import java.util.HashSet;
import com.threerings.bureau.data.AgentObject;
import com.threerings.bureau.data.BureauCodes;
import com.threerings.presents.data.ClientObject;
import com.threerings.presents.dobj.RootDObjectManager;
import com.threerings.presents.server.InvocationManager;
import static com.samskivert.util.StringUtil.safeToString;
import com.threerings.bureau.Log;
/**
* Abstracts the launching and termination of external processes (bureaus) that host instances of
* server-side code (agents).
* TODO: this class needs lots of fleshing out and as such is not 100% documented
*/
public class BureauRegistry
implements BureauProvider
{
/**
* The binary executable for the external vm being used.
* TODO: what args are needed?
* TODO: how to format the args?
* TODO: where does the binary path really come from - a bureau specification?
* Defines a way to launch a bureau. Launchers are set by the server on startup and are
* invoked whenever an agent is requested whose bureauType matches the registered type.
* @see registerBureau
*/
public String vmBinary = "java %{jar} %{host}:%{port} %{agentclassname}";
public static interface Launcher
{
/**
* Launches a new bureau using the given server connect-back url and other information.
* Called by the registry when it decides a new bureau is needed.
* @param serverUrl the url the bureau should use to connect back to the server
* @param bureauId the id of the bureau being launched
* @param token the token string to use for the credentials when logging in
*/
Process launch (String serverUrl, String bureauId, String token);
}
/**
* Creates a new registry, prepared to provide bureau services.
*/
public BureauRegistry (InvocationManager invmgr, RootDObjectManager omgr)
public BureauRegistry (
String serverUrl,
InvocationManager invmgr,
RootDObjectManager omgr)
{
_serverUrl = serverUrl;
_invmgr = invmgr;
_omgr = omgr;
invmgr.registerDispatcher(
new BureauDispatcher(this),
BureauProvider provider = new BureauProvider () {
public void bureauInitialized (ClientObject client, String bureauId) {
BureauRegistry.this.bureauInitialized(client, bureauId);
}
public void agentCreated (ClientObject client, int agentId) {
BureauRegistry.this.agentCreated(client, agentId);
}
public void agentDestroyed (ClientObject client, int agentId) {
BureauRegistry.this.agentDestroyed(client, agentId);
}
};
_invmgr.registerDispatcher(
new BureauDispatcher(provider),
BureauCodes.BUREAU_GROUP);
}
/**
* Registers a launcher of a given type. When an agent is started and no bureaus are currently
* running, the <code>bureauType</code> is used to determine the <code>Launcher</code> to call.
* @param bureauType the type of bureau that will be launched
* @param launcher the launcher to be used when the <code>bureauType</code> is requested
*/
public void setLauncher (String bureauType, Launcher launcher)
{
if (_launchers.get(bureauType) != null) {
Log.warning("Launcher for type already exists [type=" +
bureauType + "]");
return;
}
_launchers.put(bureauType, launcher);
}
/**
* Starts a new agent using the data in the given object, creating a new bureau if necessary.
*/
public void startAgent (AgentObject agent)
public synchronized void startAgent (AgentObject agent)
{
// TODO: create new bureau if needed
// TODO: otherwise call BureauSender.createAgent and pend
Bureau bureau = _bureaus.get(agent.bureauId);
if (bureau != null && bureau.ready()) {
Log.info("Bureau ready, sending createAgent " +
safeToString(agent));
BureauSender.createAgent(bureau.clientObj, agent.getOid());
// !TODO: is this the right place to register the object?
_omgr.registerObject(agent);
bureau.agentStates.put(agent, Bureau.STARTED);
bureau.summarize();
return;
}
if (bureau == null) {
Launcher launcher = _launchers.get(agent.bureauType);
if (launcher == null) {
Log.warning("Launcher not found for agent's " +
"bureau type " + safeToString(agent));
return;
}
Log.info("Creating new bureau " +
safeToString(agent) + " " +
safeToString(launcher));
bureau = new Bureau();
try {
bureau.process = launcher.launch(_serverUrl, agent.bureauId, "");
}
catch (Exception e) {
Log.warning("Could not launch process for bureau " +
safeToString(agent));
Log.logStackTrace(e);
// !TODO: how to hook into caller and keep the other clients from hanging
return;
}
_bureaus.put(agent.bureauId, bureau);
}
_omgr.registerObject(agent);
bureau.agentStates.put(agent, Bureau.PENDING);
bureau.summarize();
}
/**
* Destorys a previously started agent using the data in the given object.
* Destroys a previously started agent using the data in the given object.
*/
public void destroyAgent (AgentObject agent)
public synchronized void destroyAgent (AgentObject agent)
{
// TODO: call BureauSender.destroyAgent and pend
FoundAgent found = resolve(null, agent.getOid(), "destroyAgent");
if (found == null) {
return;
}
Log.warning("Destroying agent " + safeToString(agent));
// transition the agent to a new state and perform the effect of the transition
switch (found.state) {
case Bureau.PENDING:
found.bureau.agentStates.remove(found.agent);
// !TODO: is the the right place to destroy it?
_omgr.destroyObject(found.agent.getOid());
break;
case Bureau.STARTED:
found.bureau.agentStates.put(found.agent, Bureau.STILL_BORN);
break;
case Bureau.RUNNING:
BureauSender.destroyAgent(found.bureau.clientObj, agent.getOid());
found.bureau.agentStates.put(found.agent, Bureau.DESTROYED);
break;
case Bureau.DESTROYED:
case Bureau.STILL_BORN:
Log.warning("Acknowledging a request to destory an agent, but agent " +
"is in state " + found.state + ", ignoring request " +
safeToString(found.agent));
break;
}
found.bureau.summarize();
}
/**
* Callback for when the bureau client acknowledges starting up. Starts all pending agents and
* causes subsequent agent start requests to be sent directly to the bureau.
*/
protected synchronized void bureauInitialized (ClientObject client, String bureauId)
{
Bureau bureau = _bureaus.get(bureauId);
if (bureau == null) {
Log.warning("Acknowledging initialization of non-existent bureau " +
safeToString(bureauId));
return;
}
bureau.clientObj = client;
Log.info("Bureau created " + safeToString(bureau) +
", launching pending agents");
// find all pending agents
Set<AgentObject> pending = new HashSet<AgentObject>();
for (Map.Entry<AgentObject, Integer> entry :
bureau.agentStates.entrySet()) {
if (entry.getValue() == Bureau.PENDING) {
pending.add(entry.getKey());
}
}
// create them
for (AgentObject agent : pending) {
Log.info("Creating agent " + safeToString(agent));
BureauSender.createAgent(bureau.clientObj, agent.getOid());
bureau.agentStates.put(agent, Bureau.STARTED);
}
bureau.summarize();
}
/**
* Callback for when the bureau client acknowledges the creation of an agent.
*/
protected synchronized void agentCreated (ClientObject client, int agentId)
{
FoundAgent found = resolve(client, agentId, "agentCreated");
if (found == null) {
return;
}
switch (found.state) {
case Bureau.STARTED:
found.bureau.agentStates.put(found.agent, Bureau.RUNNING);
break;
case Bureau.STILL_BORN:
BureauSender.destroyAgent(found.bureau.clientObj, agentId);
found.bureau.agentStates.put(found.agent, Bureau.DESTROYED);
break;
case Bureau.PENDING:
case Bureau.RUNNING:
case Bureau.DESTROYED:
Log.warning("Received acknowledgement of the creation of an " +
"agent in state " + found.state + ", ignoring request " +
safeToString(found.agent));
break;
}
found.bureau.summarize();
}
/**
* Callback for when the bureau client acknowledges the destruction of an agent.
*/
protected synchronized void agentDestroyed (ClientObject client, int agentId)
{
FoundAgent found = resolve(client, agentId, "agentDestroyed");
if (found == null) {
return;
}
switch (found.state) {
case Bureau.DESTROYED:
found.bureau.agentStates.remove(found.agent);
break;
case Bureau.PENDING:
case Bureau.STARTED:
case Bureau.RUNNING:
case Bureau.STILL_BORN:
Log.warning("Acknowledging agent destruction, but state is " +
found.state + ", ignoring request " +
safeToString(found.agent));
break;
}
found.bureau.summarize();
// TODO: schedule a shutdown event for the bureau if this is the last agent
}
// from BureauProvider
public void bureauInitialized (ClientObject client, String bureauId)
/**
* Does lots of null checks and lookups and resolves the given information into FoundAgent.
*/
protected FoundAgent resolve (ClientObject client, int agentId, String resolver)
{
// TODO: create pending agents
// TODO: synchronization
com.threerings.presents.dobj.DObject dobj = _omgr.getObject(agentId);
if (dobj == null) {
Log.warning("Non-existent agent in " + resolver +
" [agentId=" + agentId + "]");
return null;
}
if (!(dobj instanceof AgentObject)) {
Log.warning("Object not an agent in " + resolver +
" " + safeToString(dobj));
return null;
}
AgentObject agent = (AgentObject)dobj;
Bureau bureau = _bureaus.get(agent.bureauId);
if (bureau == null) {
Log.warning("Bureau not found for agent in " + resolver +
" " + safeToString(agent));
return null;
}
if (!bureau.agentStates.containsKey(agent)) {
Log.warning("Bureau does not have agent in " + resolver +
" " + safeToString(agent));
return null;
}
if (bureau.clientObj == null) {
Log.warning("Bureau not yet connected in " + resolver +
" " + safeToString(agent));
return null;
}
if (client != null && bureau.clientObj != client) {
Log.warning("Masquerading request in " + resolver +
" " + safeToString(agent) +
" " + safeToString(bureau.clientObj) +
" " + safeToString(client));
return null;
}
return new FoundAgent(bureau, agent, bureau.agentStates.get(agent));
}
// from BureauProvider
public void agentCreated (ClientObject client, int agentId)
// Models the results of searching for an agent
protected static class FoundAgent
{
// TODO: remove from pending
FoundAgent (
Bureau bureau,
AgentObject agent,
int state)
{
this.bureau = bureau;
this.agent = agent;
this.state = state;
}
// Bureau containing the agent
Bureau bureau;
// The object
AgentObject agent;
// The state of the agent
int state;
}
// from BureauProvider
public void agentDestroyed (ClientObject caller, int arg1)
{
// TODO: remove from pending destroyed
}
// first stab at the data structures for holding bureaus and agents
// Models a bureau, including the process handle, all running agents and their states
protected static class Bureau
{
// Agent states {
// Not yet stated, waiting for bureau to ack
static final int PENDING = 0;
// Bureau acked, agent told to start
static final int STARTED = 1;
// Agent ack'ed, now live and hosting, ready to tell other clients
static final int RUNNING = 2;
// Agent destruction requested, waiting for acknowledge (after which the agent is removed
// from the Bureau, so has no state)
static final int DESTROYED = 3;
// Edge case: destroy request prior to RUNNING
static final int STILL_BORN = 4;
// }
// Should be non-null once the bureau is kicked off
Process process;
// The bureau's key in the map of bureaus. All requests for this bureau
// with this id should be associated with one instance
String bureauId;
// The client object of the bureau that has opened a dobj connection to
// the registry
ClientObject clientObj;
Set<AgentObject> agents;
// The states of the various agents allocated to this bureau
Map<AgentObject, Integer> agentStates =
new HashMap<AgentObject, Integer>();
boolean ready ()
{
return clientObj != null;
}
int countInState (int state)
{
Integer state1 = state;
if (!agentStates.containsValue(state1)) {
return 0;
}
int count = 0;
for (Map.Entry<AgentObject, Integer> me : agentStates.entrySet()) {
if (me.getValue() == state1) {
++count;
}
}
return count;
}
void summarize ()
{
Log.info("Bureau " + bureauId + " [" +
countInState(PENDING) + " pending, " +
countInState(STARTED) + " started, " +
countInState(RUNNING) + " running, " +
countInState(DESTROYED) + " destroyed, " +
countInState(STILL_BORN) + " still born]");
}
}
protected String _serverUrl;
protected InvocationManager _invmgr;
protected RootDObjectManager _omgr;
protected Map<String, Bureau> _bureaus;
protected Map<String, Set<AgentObject>> _pending;
protected Map<String, Launcher> _launchers = new HashMap<String, Launcher>();
protected Map<String, Bureau> _bureaus = new HashMap<String, Bureau>();
}