Difference between revisions of "Executor"

Revision as of 09:55, 27 August 2009

The Executor acts as a container for gCube tasks, i.e. functionally unconstrained bodies of code that lack a network interface but can be dynamically deployed into the service and executed through its interface. In particular, gCube tasks are designed, packaged, and deployed as plugins of the Executor service.

An instance of the Executor publishes descriptive information about the co-deployed tasks, can execute them on demand on behalf of clients, and can inform clients about the state of their execution. Clients may interact with the Executor service through a library of high-level facilities that subsumes standard service stubs to simplify the discovery of service instances and the execution of tasks available in those instances.

Design

Like all services that can be dynamically extended with plugins, the Executor has a plugin manager that accepts requests to register or deregister plugins of gCube tasks. The requests are not issued by service clients, however. They are issued by the Deployer in response to the availability of tasks in the infrastructure. The manager persists plugin profiles to autonomically re-register them at each container restarts.

Clients interact with either one of two port-types, the Engine and the Task.

The Engine port-type is the point of contact for clients that wish to launch the execution of registered tasks. The port-type is stateful, in that it maintains descriptions of the available tasks in a single stateful resource, the engine. The engine is created at service startup, when it subscribes with the plugin manager to be notified of plugin registration and de-registration events. It is then bound to the port-type into a WS-Resource accessible to clients via the implied resource pattern of WSRF. The task descriptions are modelled as a single, multi-valued Resource Property (RP) of the WS-Resource and published at regular intervals in all the scopes of the service instance. Task descriptions include the name of the task, a textual description for it, a set of arbitrary-valued properties, prototypical examples of the task inputs, and prototypical examples of the task outputs. The precise definition of the RP and the signature of the launch operation can be found in the WSDL of the Engine port-type.

The Task port-type is the the point of contacts for clients that wish to monitor the execution of tasks. The port-type is stateful, in that it maintains information about the execution of tasks in dedicated task resources. Task resources are created by the engine when tasks are launched, and are bound to the port-type into WS-Resources available via the implied resource pattern of WSRF. The execution state is modelled as RPs of the WS-Resources and published in all the scopes of the WS-Resources at regular intervals. RPs include the start time, inputs, and current state of the execution, as well as the logs, outputs and errors produced by the task. The port-type does not expose ad-hoc operations for monitoring purpose but relies on the standard operations of the gCube Notification Provider.

Task resources interact with running tasks by injecting them with a task runtime object in which they can find what they need to consume (e.g. inputs) and place what they need to produce (e.g. logs, outputs, errors). Task resources also inject tasks with a logger that redirects transparently to the resources all the logs produced by the tasks. It is through the runtime that task resources can publish the current state of execution in RPs of WS-Resources.

Finally, special treatment is given to scheduled tasks, i.e. tasks that execute at fixed intervals indefinitely or as long as certain conditions are verified. First, their task resources are persisted and the execution of the task resumed after a container startup (it would be unsound to restart non scheduled tasks). Second, their task resources subscribe with the plugin manager to be notified if the plugin of the task is deregistered; in this case, the task is stopped at the end of the current schedule (this is not generally possible with non scheduled tasks).

Sample Usage

The examples below use the high-level facilities of the client library of the Executor, partly because it is the recommended way to interact with the service and partly because the use of plain stubs (also included in the library) can be inferred from the public interfaces of the service.

Conceptually, most clients engage in the following interactions:

• discover service instances that can execute the target task. This requires interaction with the Information System.
• launch the execution of the task with one the discovered instances. This requires interaction with the Engine port-type of the Executor.
• monitor the execution of the task. This requires interaction with the Task port-type of the Executor.

These interactions are conveniently subsumed by instances of ExecutorCall, a class that model high-level calls to the Executor service. ExecutorCall is instantiated with the name of the target task, the intended scope of the call, and, optionally, security settings. Scope information may be provided with a GCUBEScope or a GCUBEScopeManagers, while security settings are provided by a GCUBEScopeManager. If the call is issued from another service, scope and security information can also be provided by a GCUBEServiceContexts. The example below illustrates the instantiation possibilites:

String name = ....
GCUBEScope scope = ....
GCUBEScopeManager smanager = .....
GCUBESecurityManager secmanager = .....
GCUBEServiceContext context = ....

//some call
Executor call;
call = new ExecutorCall(name,scope);
call = new ExecutorCall(name,smanager);
call = new ExecutorCall(name,scope, secmanager);
call = new ExecutorCall(name,smanager, secmanager);
call = new ExecutorCall(name,context);

The call is now configured to transparently discover instances of the Executor service that can execute the target task. Published properties of the target task can be set on the call to further disambiguate discovery:

String propertyName = ...
String propertyValue = ...
call.setTaskProperty(propertyName,propertyValue);

Discovery, on the other hand, can be entirely bypassed if the endpoint of a suitable Executor instance is already known:

String hostname = ...
String port = ...
call.setEndpoint(hostname,port);

The method launch can then be invoked on the call to execute the target task. This may require the preliminary definition of task inputs as a Map of string keys and arbitrary object values, e.g.:

Map<String,Object> inputs = ...
String inputName = ...
String inputValue = ...
inputs.put(inputName,inputValue)
...

The name and value the inputs must of course align with task expectations (specified in the task documentation and also manifest is in the RP published by the Engine port-type of all service instances that can execute the target task). Here we assume a string valued input, though any input type provided by the plugin is allowed, e.g.:

String input2name = ....
MyType input2value = ...
inputs.put(input2Name, input2Value);

In this case, the call must be configured with the type mapping required to serialise MyType instances (a type mapping is a correspondence between a type and its serialisation on the wire). As type mappings are explicitly provided by the context of the plugin of the target task (see below), e.g. an instance of MyPluginContext, the client can conveniently set them on the call as follows:

MyPluginContext pcontext = new MyPluginContext();
call.addTypeMapping(pcontext.getTypeMappings());

Do notice that clients that use task-specific types have an explicit dependency on the plugin of the task in addition to the service client libraries.

The target task can finally be executed as follows:

ExecutorCall.TaskProxy proxy = call.launch(inputs);

where TaskProxy is the type of a local proxy of the running task. Clients can use it to poll the execution state (consult the documentatio for the full list of methods that can be invoked on a task proxy):

System.out.format("Task invoked started at %Tc with %s state",proxy.getStartTime(),proxy.getState());

The task proxy reflects the value of the RPs of the WS-Resource that models the execution of the target task. Its methods, however, execute against a local cache of the RP values and do not engage the remote WS-Resource. The cache is first populated immediately after the execution of the task but must be explicitly synchronized by clients whenever fresh information is required:

proxy.synchronize();

Typically, clients will wish to synchronise proxies when there is some change to the execution of the target task. The Executor allows clients to subscribe for changes to the overall state of the execution and to its output, and the client library offers a convenient abstraction for this purpose. TaskMonitor is an abstract class that defines callbacks for event notifications and clients can subclass it to implement the callbacks for the events of interest. One common way of doing so is with an anonymous class, e.g.:

TaskMonitor monitor = new TaskMonitor() {
      public void onStateChange(String newState) throws Exception {
        //state values are the string conversion of handler's states
        if (state.equals(State.Failed.INSTANCE.toString())) { 
	  this.getProxy().synchronize();//synchronise to get error
          System.out.println("task has failed with error "+this.getProxy().getError());
        }
        else if (state.equals(State.Done.INSTANCE.toString())) {
          this.getProxy().synchronize();//synchronise to get output
	   System.out.println("task has completed with: "+this.getProxy().getOutput().get("endresult")));
        }
        else logger.info("task has moved to status "+state);
     }

     public void onOutputChange(Map<String, Object> output) { 
        if (output.containsKey("endresult"))
	logger.info("output message is "+output.get("endresult")));
     }
};

This monitor defines callbacks for both type of events and will thus receive both. Optionally, a monitor can express interest in either type of event by passing a TaskMonitor.TaskTopic to the constructor of TaskMonitor. With the anonymous class approach used above this can be accomplished as follows:

TaskMonitor monitor = new TaskMonitor(TaskMonitor.STATECHANGE) {
      public void onStateChange(String newState) throws Exception {...}
};

The task monitor can finally be passed to the ExecutorCall as a parameter of the launch:

ExecutorCall.TaskProxy proxy = call.launch(inputs, monitor);

The call will then transparently subscribe the monitor with the WS-Resource that models the execution of the target task. It will also inject the local proxy in the monitor (as well as returning it from the launch method as we have already seen). This is why the callback implementations can retrieve it with this.getProxy(), as can be seen above.

Finally, note that tasks that require no inputs can be simply invoked as follows:

proxy = call.launch();
proxy = call.launch(monitor);

Plugin Development

Executor plugins may have arbitrary size and dependencies but must include the following components:

• an implementation of the ExecutorTask interface which embodies the task;

• a subclass of the ExecutorPluginContext class which provides information about the task;

• a profile that binds the plugin to the Executor service.

Tasks

ExecutorTask is the interface required of all gCube Tasks. It is a tagging interface defined by the Executor service:

public interface ExecutorTask extends Lifetime<TaskRuntime> {}

Lifetime is in turn a parametric interface defined in gCore:

public interface Lifetime<T> extends GCUBEIHandler<T> {

   public State getLifetimeState();
   public void setLifetimeState(State state) throws IllegalArgumentException,IllegalStateException;

}

where GCUBEIHandler<T> is the interface all of gCore handlers that handle objects of type T for some T (we assume basic familiarity with gCore handlers).

By convention, a handler that implements Lifetime declares to manage its own lifetime, i.e. invokes setLifetimeState() whenever the state of its execution assumes one of the states pre-defined below State (CREATED, RUNNING, SUSPENDED, FAILED, DONE) or in fact any other user-defined state under State. By doing so, the handler guarantees that interested clients can use the methods getLifeTimeState() to know the state of its execution. In fact, Lifetime handlers do not need to implement these methods (and in fact should not) as the base GCUBEHandler already does. They only need to declare that they make use of them.

It is now clear that an ExecutorTask is a handler that manages its own lifetime (so that the Executor can monitor it) and that handles objects of type TaskRuntime, the role of which we discuss below. Notice that, like any other handler, a ExecutorTask can subclass any of the generic handler types defined in gCore, such as GCUBEScheduledHandler, GCUBESequentialHandler, GCUBEParallelHandler, GCUBEServiceHandler and its subclasses.

The following is possibly the simplest ExecutorTask, for which we only need to subclass the generic GCUBEHandler. Note that the task honors the commitment to manage its own lifetime.

class HelloTask extends GCUBEHandler<TaskRuntime> implements ExecutorTask {

   public void run() throws Exception {
      this.setLifetimeState(Running.INSTANCE);
      this.getLogger().trace("hello world");
      this.setLifetimeState(Done.INSTANCE);
   }

}

What really differentiates a task from any other handler is the handled TaskRuntime object. As discussed above, the task can use it to retrieve any inputs that clients may pass to the Executor, to access the context of the plugin, and to produce outputs and errors. Here is an example:

class EchoTask extends GCUBEHandler<TaskRuntime> implements ExecutorTask {

   public void run() throws Exception {
      TaskRuntime r = this.getHandled();
      this.setLifetimeState(Running.INSTANCE);
      this.setLifetimeState(Suspended.INSTANCE);
      this.getLogger().info("pausing...");
      Thread.sleep(1000);
      if (Math.random()>.8) {
   	   r.throwException(new Exception("random problem"));
   	   this.setLifetimeState(Failed.INSTANCE);
      }
      else  {
   	   r.addOutput("echo","hello "+r.getInput("clientname"));
          this.setLifetimeState(Done.INSTANCE);
      }
   }

}

Here the task uses its runtime to throw exceptions (method throwException()), to produce outputs (method addOutput()) and to retrieve inputs by name (method getInput()). Consult the documentation of TaskRuntime for a complete list of available methods. Note also that the logs emitted by the task will be transparently intercepted by the Executor service, published in the scope where the task will be launched, and made available to clients that will monitor its execution.

In conclusion, developing an ExecutorTask is no different from developing any other handler. One must only make sure to honor the Lifetime interface and to make appropriate of the handled TaskRuntime object.

Contexts

(in progress)

Plugin Profiles

(in progress)