Difference between revisions of "Executor"

Revision as of 10:54, 20 September 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:

 extends GCUBEIHandler<T> {
 
   public State getState();
 
}

where GCUBEIHandler<T> is the interface all of gCore handlers that handle objects of type T for some T (we assume basic familiarity with [https://wiki.gcore.research-infrastructures.eu/gCore/index.php/The_Handler_Framework gCore handlers|).

By convention, a handler that implements Lifetime declares to manage its own lifetime and return accurate values from invocations of getState(). In practice, this means that implementers of ExecutorTask will invoke the method setState() in GCUBEHandler -- the base implementation of GCUBEIHandler) -- to set the current state of the task to one of the pre-defined states shown below (State (CREATED, RUNNING, SUSPENDED, FAILED, DONE) or in fact any other custom state defined as a subclass of State (see here for more details).

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 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.setState(Running.INSTANCE);
       this.getLogger().trace("hello world");
       this.setState(Done.INSTANCE);
    }
 
}

What really differentiates a task from any other handler is the handled TaskRuntime object. As discussed above, the Executor injects it into the task before executing it and that 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.setState(Running.INSTANCE);
       this.setState(Suspended.INSTANCE);
       this.getLogger().info("pausing...");
       Thread.sleep(1000);
       if (Math.random()>.8) {
    	   r.throwException(new Exception("random problem"));
    	   this.setState(Failed.INSTANCE);
       }
       else  {
    	   r.addOutput("echo","hello "+r.getInput("clientname"));
           this.setState(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, make use of the handled TaskRuntime object, and satsfy the following requirement:

Note:Tasks are instantiated by the Executor and this requires them to have a zero-argument constructor.

Contexts

A plugin context is the entry point to the implementation of a plugin and exposes information that allows the Executor to manage it.

Part of this information is common to all plugins and is required by gCore to act on behalf of the Executor for the most generic aspects of plugin management. This includes:

• zero or more descriptive properties whereby the plugin may be identified by the clients of the Executor;
• zero or more type mappings that may be required to deserialise object types that are statically unknown to the Executor but must be exchanged on the wire between its clients and the plugin (cf. discussion above).

Other information is instead required specifically by the Executor, including:

• the implementation of ExecutorTask which embodies the task included in the plugin;
• zero or more sample inputs required by the task included in the plugin;
• zero or more sample outputs produced by the task included in the plugin;
• the time after which any trace of tasks that have failed or successfully completed will be removed.

GCUBEPluginContext is defined in gCore as the root of all plugin contexts. It defines a number of callbacks whereby gCore can obtain the generic information it requires. ExecutorPluginContext derives GCUBEPluginContext in the Executor and repeats the pattern for information required specifically by it. Plugin developers must derive ExecutorPluginContext to implement the callbacks and return information specific to their plugin.

To simplify the process, GCUBEPluginContext implements its callbacks in terms of empty data structures (e.g. a list of DescriptivePropertys and a list of TypeMappings). Plugin developers may then simply populate these structures in the constructors of their own contexts (e.g. by invoking addDescriptiveProperty() and addTypeMapping()). The <code>ExecutorPluginContext does the same for its own callbacks (cf. addSampleOutput() and addSampleInput()). One callback, getTaskClass(), is left abstract and plugin developers must implement it and return the class the class that implements ExecutorTask in their plugin. Finally, developers may override the implementation of getTimeToLive() to override the default time-to-live of the tasks in their plugin. The following example illustrates:

public class MyPluginContext extends ExecutorPluginContext {
 
   QName name = new QName("...", "ComplexProperty");
 
   public MyPluginContext() {
	this.addProperty(
	   new DescriptiveProperty("A simple property","property1","value1"),
	   new DescriptiveProperty("A complex property","property2",new ComplexProperty("value2"))
        );
 
	this.addTypeMappings(new TypeMapping(ComplexProperty.class,name)));
 
	this.addSampleInput(new DescriptiveProperty("A complex input","parameter1",new ComplexProperty("sample")));
	this.addSampleOutput(new DescriptiveProperty("A complex output","endresult",new ComplexProperty("sample")));	
   }
 
 
   /**{@inheritDoc} */
   public Class<? extends ExecutorTask> getTaskClass() {return MyTask.class;}
 
   /**{@inheritDoc} */
   public int getTimeToLive(){return 10;} //seconds!
 
}

Here the context constructor is used to populate the inherited data structures with properties, type mappings, sample inputs and sample outputs. ComplexProperty is a type defined within the plugin and used for the value of one plugin property, the value of the single task input, and the value the single task output. A type mapping is defined for it, accordingly. The class of MyTask is returned from getTaskClass() and getTimeToLive() is overridden to return a value of 10 seconds. Consult the documentation of ExecutorPluginContext for the full list of available methods and for the details of their signatures.

Note:Plugin contexts are instantiated by gCore and this requires them to have a zero-argument constructor.

Plugin Profiles

(in progress)

Development Environment and Testing

gCore ships with an archived Eclipse project that can be used to kick-start the development of a plugin and its testing. This is how the project would look like after import:

The project:

• assumes the definition of user libraries for gCore and plugin dependencies in the hosting workspace (cf. GCORELIBS and PLUGINDEPS). The developer must at least add to the latter dependencies to the implementation of the Executor and its stubs. * pre-defines a context, a task, and a profile for the plugin (cf. PluginContext, Task, and profile.xml). The developer can choose the physical placement of dependencies, though the project assumes them to be outside the service implementation (cf. build.properties).
• includes a customisation of the standard gCore buildfile that processes the types under the folder schema and output corresponding stub class under the package types, from which they can be imported in the rest of the implementation. As usual, configurable build properties are in build.properties.
• defines a mock of the TaskRuntime that a running instance of the Executor will inject into the task before running it (cf. MockRuntime). This allows most tasks to be tested without deploying them in a running instance of the Executor. The sample test included in the project exemplifies how to use these facilities (cf. Test). Here is the significant excerpt:

....
//load plugin profile as classpath resource
GCUBEService plugin = GHNContext.getImplementation(GCUBEService.class);
plugin.load(new InputStreamReader(Test.class.getResourceAsStream("profile.xml")));
 
MockRuntime runtime = new MockRuntime(plugin);
 
//prepare manually mock runtime as a running instance of the Executor would.
runtime.getInputs().put("parameter1", new ComplexType("testvalue"));
 
 
//prepare task as a running instance of the Executor would.
GCUBEScopeManager manager = new GCUBEScopeManagerImpl();
manager.setScope(GCUBEScope.getScope("/gcube/devsec")); //would use args normally
 
//launch task as a running instance of the Executor would.
Task task = new Task();
task.setScopeManager(manager);
task.setHandled(runtime);
task.run();

Note:The project archive can be found in $GLOBUS_LOCATION/share/gcore_tools/.