Author: tmckay
Date: 2011-07-14 20:22:03 +0000 (Thu, 14 Jul 2011)
New Revision: 4870
Added:
trunk/sage/python/sage/aviary/
trunk/sage/python/sage/aviary/__init__.py
trunk/sage/python/sage/aviary/aviaryoperations.py
trunk/sage/python/sage/qmf/
trunk/sage/python/sage/qmf/__init__.py
trunk/sage/python/sage/qmf/qmfoperations.py
Removed:
trunk/sage/python/sage/aviaryoperations.py
trunk/sage/python/sage/qmfoperations.py
Modified:
trunk/cumin/bin/cumin-web
trunk/cumin/etc/cumin.conf
trunk/cumin/python/cumin/config.py
trunk/cumin/python/cumin/main.py
trunk/cumin/python/cumin/task.py
trunk/sage/python/sage/util.py
Log:
Initial pass at aviary job operations integration.
Modified: trunk/cumin/bin/cumin-web
===================================================================
--- trunk/cumin/bin/cumin-web 2011-07-12 20:02:12 UTC (rev 4869)
+++ trunk/cumin/bin/cumin-web 2011-07-14 20:22:03 UTC (rev 4870)
@@ -13,7 +13,15 @@
def restore_IO():
sys.stderr = sys.__stderr__
sys.stdout = sys.__stdout__
-
+
+def set_aviary_configs(cumin, values):
+ cumin.use_aviary = values.use_aviary
+ cumin.aviary_host = values.aviary_host
+ cumin.aviary_job_host = values.aviary_job_host
+ cumin.aviary_query_host = values.aviary_query_host
+ cumin.aviary_job_port = values.aviary_job_port
+ cumin.aviary_query_port = values.aviary_query_port
+
def main():
# Do our own simple option check so we can redirect IO early
# without worrying about other options or the behavior of optParse
@@ -88,8 +96,7 @@
if type(values.sasl_mech_list) == str:
cumin.sasl_mech_list = values.sasl_mech_list.upper()
- cumin.use_aviary = values.use_aviary
-
+ set_aviary_configs(cumin, values)
cumin.debug = opts.debug
cumin.user = values.user
cumin.update_interval = values.update_interval
Modified: trunk/cumin/etc/cumin.conf
===================================================================
--- trunk/cumin/etc/cumin.conf 2011-07-12 20:02:12 UTC (rev 4869)
+++ trunk/cumin/etc/cumin.conf 2011-07-14 20:22:03 UTC (rev 4870)
@@ -14,6 +14,10 @@
# database: dbname=cumin user=cumin host=localhost
# brokers: localhost:5672
# sasl-mech-list: [allow all available mechanisms]
+# use-aviary: True
+# aviary-host: localhost
+# aviary-job-port: 9090
+# aviary-query-port: 9091
# log-level: info
# log-max-mb: 10
# log-max-archives: 1
Modified: trunk/cumin/python/cumin/config.py
===================================================================
--- trunk/cumin/python/cumin/config.py 2011-07-12 20:02:12 UTC (rev 4869)
+++ trunk/cumin/python/cumin/config.py 2011-07-14 20:22:03 UTC (rev 4870)
@@ -149,6 +149,27 @@
# previous behavior
param = ConfigParameter(self, "sasl-mech-list", str)
+ param = ConfigParameter(self, "use-aviary", bool)
+ param.default = True
+
+ param = ConfigParameter(self, "aviary-host", str)
+ param.default = "localhost"
+
+ # Use aviary-host for all services, unless these are
+ # set explicitly as an override. This allows different
+ # services to be hosted on different machines if desired.
+ param = ConfigParameter(self, "aviary-job-host", str)
+ param.default = ""
+
+ param = ConfigParameter(self, "aviary-query-host", str)
+ param.default = ""
+
+ param = ConfigParameter(self, "aviary-job-port", int)
+ param.default = 9090
+
+ param = ConfigParameter(self, "aviary-query-port", int)
+ param.default = 9091
+
self.log_file = ConfigParameter(self, "log-file", str)
param = ConfigParameter(self, "log-level", str)
@@ -163,9 +184,6 @@
param = ConfigParameter(self, "debug", bool)
param.default = False
- param = ConfigParameter(self, "use-aviary", bool)
- param.default = True
-
class CuminOptionParser(OptionParser,object):
def __init__(self):
OptionParser.__init__(self)
Modified: trunk/cumin/python/cumin/main.py
===================================================================
--- trunk/cumin/python/cumin/main.py 2011-07-12 20:02:12 UTC (rev 4869)
+++ trunk/cumin/python/cumin/main.py 2011-07-14 20:22:03 UTC (rev 4870)
@@ -22,8 +22,7 @@
from util import *
from widgets import *
from sage.remoteoperations import RemoteOperations
-from sage.qmfoperations import QmfOperations
-from sage.aviaryoperations import AviaryOperations
+from sage.qmf.qmfoperations import QmfOperations
from wooly import Session
from cumin.stat import PieChartPage
from cumin.widgets import AboutPage
@@ -81,9 +80,16 @@
self.sasl_mech_list = None
# Whether or not to use Aviary interface for
- # queries and job submissions. May be set via cumin.conf
+ # queries and job submissions plus configs.
+ # These values are expected to be filled in if
+ # use_aviary is left True.
self.use_aviary = True
-
+ self.aviary_host = None
+ self.aviary_job_host = None
+ self.aviary_query_host = None
+ self.aviary_job_port = None
+ self.aviary_query_port = None
+
def server_alive(self):
return self.server.server_alive()
@@ -155,7 +161,20 @@
self.remote = RemoteOperations()
ops = [QmfOperations("qmf", self.session)]
if self.use_aviary:
- ops.insert(0, AviaryOperations("aviary"))
+ from sage.aviary.aviaryoperations import AviaryOperations
+ aviary_dir = os.path.join(self.home, "rpc-defs/aviary")
+
+ # If aviary_job_host or aviary_query_host are "" or
+ # None then they inherit the value of aviary_host.
+ if not self.aviary_job_host:
+ self.aviary_job_host = self.aviary_host
+ if not self.aviary_query_host:
+ self.aviary_query_host = self.aviary_host
+ ops.insert(0, AviaryOperations("aviary", aviary_dir,
+ self.aviary_job_host,
+ self.aviary_job_port,
+ self.aviary_query_host,
+ self.aviary_query_port))
self.remote.add_mechanisms(ops)
self.model.init()
Modified: trunk/cumin/python/cumin/task.py
===================================================================
--- trunk/cumin/python/cumin/task.py 2011-07-12 20:02:12 UTC (rev 4869)
+++ trunk/cumin/python/cumin/task.py 2011-07-14 20:22:03 UTC (rev 4870)
@@ -173,8 +173,17 @@
def make_callback(self):
def completion(status_code, output_args):
- self.status_code = status_code
- self.output_args = output_args
+ # Make results recorded directly through the callback
+ # match what is done when TaskInvocation values are
+ # set by an ObjectTask or Task on failure
+ # (see Task.invoke() and Task.exception() above)
+ # Previously the different mechanisms produced different results.
+ if status_code not in (0, "OK"):
+ self.status = self.FAILED
+ self.exception = Exception(status_code)
+ else:
+ self.status_code = status_code
+ self.output_args = output_args
self.end()
return completion
Added: trunk/sage/python/sage/aviary/__init__.py
===================================================================
Copied: trunk/sage/python/sage/aviary/aviaryoperations.py (from rev 4863,
trunk/sage/python/sage/aviaryoperations.py)
===================================================================
--- trunk/sage/python/sage/aviary/aviaryoperations.py (rev 0)
+++ trunk/sage/python/sage/aviary/aviaryoperations.py 2011-07-14 20:22:03 UTC (rev 4870)
@@ -0,0 +1,331 @@
+import os
+
+from suds import *
+from suds.client import Client
+from sage.util import CallSync, CallThread, ObjectPool
+import threading
+import logging
+
+#f = open("./suds.client.log", 'a+')
+#h = logging.StreamHandler(f)
+#logging.getLogger('suds.client').setLevel(logging.DEBUG)
+
+#f = open("./suds.client.log", 'a+')
+#h = logging.StreamHandler(f)
+#logging.getLogger('suds.client').addHandler(h)
+
+#logging.getLogger('suds.transport').setLevel(logging.DEBUG)
+#f = open("./suds.transport.log", 'a+')
+#h = logging.StreamHandler(f)
+#logging.getLogger('suds.transport').addHandler(h)
+
+#logging.getLogger('suds.xsd.schema').setLevel(logging.DEBUG)
+#f = open("./suds.xsd.log", 'a+')
+#h = logging.StreamHandler(f)
+#logging.getLogger('suds.xsd.schema').addHandler(h)
+
+#logging.getLogger('suds.wsdl').setLevel(logging.DEBUG)
+#f = open("./suds.wsdl.log", 'a+')
+#h = logging.StreamHandler(f)
+#logging.getLogger('suds.wsdl').addHandler(h)
+
+'''
+Stuff to do:
+
+- Add logging to Aviary
+
+- See if we can fill in the scheduler/pool information.
+What is cumin passing? (the scheduler object has this stuff,
+but what about hierarchical collectors?)
+
+- get the submission objects passed down if possible
+for job control functions. This would be extra stuff for QMF, but helpful with
+Aviary.
+
+Add a summary comment at the top like QMF opreations
+
+can we use default/timeout with suds?
+yes, there is at least a timeout that can be set once,
+it might be possible to change the timeout per call.
+'''
+
+class AviaryOperations(object):
+ def __init__(self, name, datadir, job_host, job_port, query_host, query_port):
+ assert type(job_host) is str
+ assert type(query_host) is str
+
+ self.name = name
+ self.datadir = datadir
+
+ # Add http:// if there is no protocol...
+ if job_host.find("://") < 0:
+ job_host = "http://" + job_host
+ if query_host.find("://") < 0:
+ query_host = "http://" + query_host
+
+ self.job_url = job_host + ":" + str(job_port) +
"/services/job/"
+ self.query_url = query_host + ":" + str(query_port) +
"/services/query/"
+
+ job_wsdl = "file:" + os.path.join(self.datadir,
"aviary-job.wsdl")
+ query_wsdl = "file:" + os.path.join(self.datadir,
"aviary-query.wsdl")
+
+ self.job_client_pool = JobClientPool(job_wsdl, None)
+ self.query_client_pool = QueryClientPool(query_wsdl, None)
+
+ self.type_map = self._type_map()
+
+# scheduler operations
+
+ def set_job_attribute(self, scheduler, job_id, name, value, callback):
+ assert callback
+
+ job_client = self.job_client_pool.get_object()
+
+ def my_callback(result):
+ self.job_client_pool.return_object(job_client)
+ # massage results for use by standard callback
+ cb_args = self._cb_args_dataless(result)
+ callback(*cb_args)
+
+ url = self.job_url + "setJobAttribute"
+ #print "Aviary set job attribute " + url
+
+ # Have to set the URL for the method. This might go away someday...
+ job_client.set_options(location=url)
+
+ # Make a job id parameter (see job wsdl)
+ jobId = job_client.factory.create('ns0:JobID')
+ jobId.job = job_id
+ jobId.pool = scheduler.Pool
+ jobId.scheduler = scheduler.Name
+
+ # Make attribute parameter from name and value
+ aviary_attr = job_client.factory.create('ns0:Attribute')
+ aviary_attr.name = name
+ aviary_attr.type = "STRING"
+ aviary_attr.value = value
+
+ t = CallThread(job_client.service.setJobAttribute, my_callback, jobId,
aviary_attr)
+ t.start()
+
+ def submit_job(self, scheduler, ad, callback):
+ assert callback
+
+ job_client = self.job_client_pool.get_object()
+
+ def my_callback(result):
+ # Turn this back off before we put it back in the pool
+ # so allow_overrides isn't set for someone else...
+ job_client.set_allow_overrides(False)
+ self.job_client_pool.return_object(job_client)
+ if isinstance(result, Exception):
+ callback(result, None)
+ else:
+ # the aviary response has the job id available,
+ # we'll pass it anyway even though Cumin does not care
+ # at the present time
+ status = AviaryOperations._get_status(result.status)
+ callback(status, result.id)
+
+ url = self.job_url + "submitJob"
+ #print "Aviary submit job " + url
+
+ # Have to set the URL for the method. This might go away someday...
+ job_client.set_options(location=url)
+
+ # Set basic attributes in the order defined by aviary-job.wsdl.
+ args = list()
+ basic_attrs = ("Cmd", "Args", "Owner",
"Iwd", "Submission")
+ for attr in basic_attrs:
+ args.append(ad[attr])
+
+ # Add empty list for Aviary's basic requirement value...
+ args.append([])
+
+ # and let's let Requirements remain an unrestricted expression so that
+ # we can just pass through the value from Cumin without interfering.
+ # To do that, we need to specify Requirements through the
+ # "extras" fields and set allowOverrides to True.
+ # (otherwise, Requirements will be limited to particular
+ # resource constraint types defined by aviary)
+ job_client.set_allow_overrides(True)
+ extras = list()
+ for k, v in ad.iteritems():
+ # We don't need to send descriptors down to aviary
+ # and basic_attrs have already been filled in
+ if k == "!!descriptors" or k in basic_attrs:
+ continue
+
+ extra = job_client.factory.create('ns0:Attribute')
+ extra.name = k
+ # But we do need to look in descriptors to find expressions...
+ if k in ad["!!descriptors"]:
+ extra.type = "EXPRESSION"
+ else:
+ try:
+ extra.type = self.type_map[type(v)]
+ except KeyError:
+ extra.type = "UNDEFINED"
+ extra.value = v
+ extras.append(extra)
+
+ # Important, extras itself must be added as an embedded list or
+ # suds will consider only a single item
+ args.append(extras)
+
+ t = CallThread(job_client.service.submitJob, my_callback, *args)
+ t.start()
+
+ def hold_job(self, scheduler, job_id, reason,
+ callback=None, default=None, timeout=5):
+ '''
+ This method is asynchronous iff 'callback' is supplied.
+ '''
+ return self._control_job(scheduler, job_id, reason,
+ "holdJob",
+ callback, default, timeout)
+
+ def release_job(self, scheduler, job_id, reason,
+ callback=None, default=None, timeout=5):
+ '''
+ This method is asynchronous iff 'callback' is supplied.
+ '''
+ return self._control_job(scheduler, job_id, reason,
+ "releaseJob",
+ callback, default, timeout)
+
+ def remove_job(self, scheduler, job_id, reason,
+ callback=None, default=None, timeout=5):
+ '''
+ This method is asynchronous iff 'callback' is supplied.
+ '''
+ return self._control_job(scheduler, job_id, reason,
+ "removeJob",
+ callback, default, timeout)
+
+# Secret private implementation stuff, don't look!
+
+ @classmethod
+ def _type_map(cls):
+ # Need to be able to turn simple Python types into Aviary types for attributes
+ return {int: "INTEGER", float: "FLOAT", str:
"STRING", bool: "BOOLEAN"}
+
+ @classmethod
+ def _get_status(cls, result):
+ # For Aviary operations, if the operation
+ # did not work the reason is in the text field.
+ # In cumin, we want to pass any error text as
+ # the status parameter to callbacks
+ if result.code != "OK":
+ return result.text
+ return result.code
+
+ @classmethod
+ def _cb_args_dataless(cls, result):
+ # Marshal data in result for passing to standard callback.
+ # This routine is for results that contain status only, no data.
+ if isinstance(result, Exception):
+ status = result
+ else:
+ status = AviaryOperations._get_status(result)
+ return (status, None)
+
+ def _call_sync(self, process_results, meth, *meth_args, **meth_kwargs):
+ # Common interface with QMF operations requires that a MethodResult
+ # object (or something just like it) be returned for synchronous calls.
+ # Might as well use the CallSync since the completion() routine
+ # handles status semantics.
+ sync = CallSync()
+ try:
+ result = meth(*meth_args, **meth_kwargs)
+ except Exception, e:
+ result = e
+ cb_args = process_results(result)
+ sync.get_completion()(*cb_args)
+ return sync
+
+ def _control_job(self, scheduler, job_id, reason, meth_name,
+ callback, default, timeout):
+
+ url = self.job_url + meth_name
+ client = self.job_client_pool.get_object()
+ meth = getattr(client.service, meth_name)
+
+ # Have to set the URL for the method. This might go away someday...
+ #print "Aviary control job " + url
+ client.set_options(location=url)
+
+ # Make a job id parameter (see job wsdl)
+ jobId = client.factory.create('ns0:JobID')
+ jobId.job = job_id
+ jobId.pool = scheduler.Pool
+ jobId.scheduler = scheduler.Name
+
+ if callback:
+ def my_callback(result):
+ self.job_client_pool.return_object(client)
+ cb_args = self._cb_args_dataless(result)
+ callback(*cb_args)
+
+ t = CallThread(meth, my_callback, jobId, reason)
+ t.start()
+ else:
+ res = self._call_sync(self._cb_args_dataless, meth, jobId, reason)
+ self.job_client_pool.return_object(client)
+ return res;
+
+try:
+ # Some of this stuff does not exist pre suds 0.4.1
+ # Make it work anyway for testing on such hosts by
+ # declaring simple OverridesPlugin in the exception case.
+ from suds.plugin import MessagePlugin
+ from suds.sax.attribute import Attribute
+ class OverridesPlugin(MessagePlugin):
+ '''
+ Plugin which allows optional addition of attributes
+ to the suds message after marshalling.
+ '''
+ def __init__(self):
+ super(OverridesPlugin, self).__init__()
+
+ self.allow = False
+
+ def marshalled(self, context):
+ if self.allow:
+ sj_body = context.envelope.getChild('Body')[0]
+ sj_body.attributes.append(Attribute("allowOverrides",
"true"))
+except:
+ class OverridesPlugin(object):
+ def __init__(self):
+ self.allow = False
+
+class OverrideClient(Client):
+ '''
+ Instantiate a Client object with an OverridesPlugin and
+ retain a pointer to the plugin so that it can be controlled.
+ '''
+ def __init__(self, *args):
+ self.override = OverridesPlugin()
+ try:
+ super(OverrideClient, self).__init__(*args, plugins=[self.override])
+ except:
+ super(OverrideClient, self).__init__(*args)
+
+ def set_allow_overrides(self, truth):
+ self.override.allow = truth
+
+class JobClientPool(ObjectPool):
+ def __init__(self, job_wsdl, max_size):
+ super(JobClientPool, self).__init__(max_size)
+ self.job_wsdl = job_wsdl
+
+ def create_object(self):
+ return OverrideClient(self.job_wsdl)
+
+class QueryClientPool(ObjectPool):
+ def __init__(self, query_wsdl, max_size):
+ super(QueryClientPool, self).__init__(max_size)
+ self.query_wsdl = query_wsdl
+
+ def create_object(self):
+ return Client(self.query_wsdl)
Deleted: trunk/sage/python/sage/aviaryoperations.py
===================================================================
--- trunk/sage/python/sage/aviaryoperations.py 2011-07-12 20:02:12 UTC (rev 4869)
+++ trunk/sage/python/sage/aviaryoperations.py 2011-07-14 20:22:03 UTC (rev 4870)
@@ -1,4 +0,0 @@
-class AviaryOperations(object):
- def __init__(self, name):
-
- self.name = name
Added: trunk/sage/python/sage/qmf/__init__.py
===================================================================
Copied: trunk/sage/python/sage/qmf/qmfoperations.py (from rev 4863,
trunk/sage/python/sage/qmfoperations.py)
===================================================================
--- trunk/sage/python/sage/qmf/qmfoperations.py (rev 0)
+++ trunk/sage/python/sage/qmf/qmfoperations.py 2011-07-14 20:22:03 UTC (rev 4870)
@@ -0,0 +1,183 @@
+import logging
+from sage.util import CallSync, wait
+
+log = logging.getLogger("sage.remoteoperations")
+
+class QmfOperations(object):
+ '''
+ Provides an interface to QMF remote procedure calls.
+ Public methods may implement synchronous or asynchronous
+ calling semantics, or may support either one depending on the parameters
+ passed.
+ If a method takes a 'callback' parameter and that parameter is not
+ None, the method will have asynchronous calling semantics. Otherwise,
+ the method will have synchronous calling semantics and the 'default' and
+ 'timeout' parameters will be used.
+ The type of the result returned from all synchronous methods will be
+ sage.util.MethodResult or a descendent.
+ The 'callback' parameter if set is expected to be a reference to a
+ function that can be invoked by self.session.call_method() when an
+ operation is complete.
+ '''
+
+ def __init__(self, name, session):
+ '''
+ If this object is added to a RemoteOperations object as a mechanism,
+ the 'name' parameter will be used to create an attribute which points
+ to this object.
+ The 'session' parameter is expected to be an object that defines the
+ following method:
+
+ def call_method(callback, obj, meth, args)
+
+ session.call_method should be capable of invoking the 'meth' method
+ on the 'obj' object with the tuple 'args' as arguments. It
should
+ be capable of passing results to the 'callback' function. Any
+ 'callback' parameter passed to a public method of a QmfOperations object
+ should have a signature that is compatible with that expected by
+ session.call_method().
+ '''
+
+ self.name = name
+ self.session = session
+
+ # Note, methods below were written to be synchronous, asynchronous, or
+ # selectable to reflect existing usage of QMF methods in cumin at
+ # the time 'sage' was created. Any of these methods may be changed to
+ # have synchronous or asynchronous calling semantics by using the callback,
+ # default, and timeout paramters and the self._call method.
+
+# broker operations
+
+ def queue_move_messages(self, broker, src, dst, count, callback):
+ assert callback
+ self._call(broker, "queueMoveMessages", callback, 0, 0, src, dst,
count)
+
+# methods on various broker objects. close is implemented on multiple objects
+
+ def close(self, obj, callback):
+ assert callback
+ self._call(obj, "close", callback, 0, 0)
+
+ def detach(self, obj, callback):
+ assert callback
+ self._call(obj, "detach", callback, 0, 0)
+
+ def bridge(self, link, durable, src, dest, key, tag, excludes, srcIsQueue,
srcIsLocal, dynamic, sync, callback):
+ assert callback
+ self._call(link, "bridge", callback, 0, 0,
+ durable, src, dest, key, tag, excludes, srcIsQueue, srcIsLocal,
dynamic, sync)
+
+ def connect(self, obj, host, port, durable, mech, username, password, transport,
callback):
+ assert callback
+ self._call(obj, "connect", callback, 0, 0,
+ host, port, durable, mech, username, password, transport)
+
+# queue operations
+
+ def purge_queue(self, queue, count, callback):
+ assert callback
+ self._call(queue, "purge", callback, 0, 0, count)
+
+# scheduler operations
+
+ def set_job_attribute(self, scheduler, job_id, name, value, callback):
+ assert callback
+ self._call(scheduler, "SetJobAttribute", callback, 0, 0,
+ job_id, name, value)
+
+ def submit_job(self, scheduler, ad, callback):
+ assert callback
+ self._call(scheduler, "SubmitJob", callback, 0, 0, ad)
+
+ def hold_job(self, scheduler, job_id, reason,
+ callback=None, default=None, timeout=5):
+ '''
+ This method is asynchronous iff 'callback' is supplied.
+ '''
+ return self._call(scheduler, "HoldJob", callback, default,
+ timeout, job_id, reason)
+
+ def release_job(self, scheduler, job_id, reason,
+ callback=None, default=None, timeout=5):
+ '''
+ This method is asynchronous iff 'callback' is supplied.
+ '''
+ return self._call(scheduler, "ReleaseJob", callback, default,
+ timeout, job_id, reason)
+
+ def remove_job(self, scheduler, job_id, reason,
+ callback=None, default=None, timeout=5):
+ '''
+ This method is asynchronous iff 'callback' is supplied.
+ '''
+ return self._call(scheduler, "RemoveJob", callback, default,
+ timeout, job_id, reason)
+
+# jobserver operations
+
+ def get_job_ad(self, job_server, job_id, default=None, timeout=5):
+ return self._call(job_server, "GetJobAd", 0, default, timeout, job_id)
+
+ def fetch_job_data(self, job_server, job_id, file, start, end,
+ default=None, timeout=5):
+ return self._call(job_server, "FetchJobData", 0, default, timeout,
+ job_id, file, start, end)
+
+# negotiator operations
+
+ def set_limit(self, negotiator, lim_name, lim_max, default=None, timeout=5):
+ return self._call(negotiator, "SetLimit", 0, default, timeout,
+ lim_name, lim_max)
+
+ def get_limits(self, negotiator, callback):
+ assert callback
+ return self._call(negotiator, "GetLimits", callback, 0, 0)
+
+ def reconfig(self, negotiator, default=None, timeout=5):
+ return self._call(negotiator, "Reconfig", 0, default, timeout)
+
+ def set_raw_config(self, negotiator, name, value, default=None, timeout=5):
+ return self._call(negotiator, "SetRawConfig", 0, default, timeout,
+ name, value)
+
+ def get_raw_config(self, negotiator, name, callback):
+ assert callback
+ return self._call(negotiator, "GetRawConfig", callback, 0, 0, name)
+
+# master operations
+
+ def start(self, master, daemon, callback):
+ assert callback
+ return self._call(master, "Start", callback, 0, 0, daemon)
+
+ def stop(self, master, daemon, callback):
+ assert callback
+ return self._call(master, "Stop", callback, 0, 0, daemon)
+
+# submission operations
+
+ def get_job_summaries(self, submission, callback):
+ assert callback
+ return self._call(submission, "GetJobSummaries", callback, 0, 0)
+
+# Secret private implementation stuff, don't look!
+ def _call(self, obj, meth, cb, dflt, tout, *args):
+ if cb:
+ self.session.call_method(cb, obj, meth, args)
+ else:
+ try:
+ sync = CallSync(log, dflt)
+ self.session.call_method(sync.get_completion(),
+ obj, meth, args)
+ except Exception, e:
+ sync.error = e
+ log.exception(e)
+
+ wait(sync.done, timeout=tout)
+ if not sync.got_data and not sync.error:
+ sync.error = Exception("Request timed out")
+ msg = "QMF call %s timed out" % meth
+ log.error(msg)
+ return sync
+
Deleted: trunk/sage/python/sage/qmfoperations.py
===================================================================
--- trunk/sage/python/sage/qmfoperations.py 2011-07-12 20:02:12 UTC (rev 4869)
+++ trunk/sage/python/sage/qmfoperations.py 2011-07-14 20:22:03 UTC (rev 4870)
@@ -1,181 +0,0 @@
-import logging
-from sage.util import CallSync, wait
-
-log = logging.getLogger("sage.remoteoperations")
-
-class QmfOperations(object):
- '''
- Provides an interface to QMF remote procedure calls.
- Public methods may implement synchronous or asynchronous
- calling semantics, or may support either one depending on the parameters
- passed.
- If a method takes a 'callback' parameter and that parameter is not
- None, the method will have asynchronous calling semantics. Otherwise,
- the method will have synchronous calling semantics and the 'default' and
- 'timeout' parameters will be used.
- The 'callback' parameter if set is expected to be a reference to a
- function that can be invoked by self.session.call_method() when an
- operation is complete.
- '''
-
- def __init__(self, name, session):
- '''
- If this object is added to a RemoteOperations object as a mechanism,
- the 'name' parameter will be used to create an attribute which points
- to this object.
- The 'session' parameter is expected to be an object that defines the
- following method:
-
- def call_method(callback, obj, meth, args)
-
- session.call_method should be capable of invoking the 'meth' method
- on the 'obj' object with the tuple 'args' as arguments. It
should
- be capable of passing results to the 'callback' function. Any
- 'callback' parameter passed to a public method of a QmfOperations object
- should have a signature that is compatible with that expected by
- session.call_method().
- '''
-
- self.name = name
- self.session = session
-
- # Note, methods below were written to be synchronous, asynchronous, or
- # selectable to reflect existing usage of QMF methods in cumin at
- # the time 'sage' was created. Any of these methods may be changed to
- # have synchronous or asynchronous calling semantics by using the callback,
- # default, and timeout paramters and the self._call method.
-
-# broker operations
-
- def queue_move_messages(self, broker, src, dst, count, callback):
- assert callback
- self._call(broker, "queueMoveMessages", callback, 0, 0, src, dst,
count)
-
-# methods on various broker objects. close is implemented on multiple objects
-
- def close(self, obj, callback):
- assert callback
- self._call(obj, "close", callback, 0, 0)
-
- def detach(self, obj, callback):
- assert callback
- self._call(obj, "detach", callback, 0, 0)
-
- def bridge(self, link, durable, src, dest, key, tag, excludes, srcIsQueue,
srcIsLocal, dynamic, sync, callback):
- assert callback
- self._call(link, "bridge", callback, 0, 0,
- durable, src, dest, key, tag, excludes, srcIsQueue, srcIsLocal,
dynamic, sync)
-
- def connect(self, obj, host, port, durable, mech, username, password, transport,
callback):
- assert callback
- self._call(obj, "connect", callback, 0, 0,
- host, port, durable, mech, username, password, transport)
-
-# queue operations
-
- def purge_queue(self, queue, count, callback):
- assert callback
- self._call(queue, "purge", callback, 0, 0, count)
-
-# scheduler operations
-
- def set_job_attribute(self, scheduler, job_id, name, value, callback):
- assert callback
- self._call(scheduler, "SetJobAttribute", callback, 0, 0,
- job_id, name, value)
-
- def submit_job(self, scheduler, ad, callback):
- assert callback
- self._call(scheduler, "SubmitJob", callback, 0, 0, ad)
-
- def hold_job(self, scheduler, job_id, reason,
- callback=None, default=None, timeout=5):
- '''
- This method is asynchronous iff 'callback' is supplied.
- '''
- return self._call(scheduler, "HoldJob", callback, default,
- timeout, job_id, reason)
-
- def release_job(self, scheduler, job_id, reason,
- callback=None, default=None, timeout=5):
- '''
- This method is asynchronous iff 'callback' is supplied.
- '''
- return self._call(scheduler, "ReleaseJob", callback, default,
- timeout, job_id, reason)
-
- def remove_job(self, scheduler, job_id, reason,
- callback=None, default=None, timeout=5):
- '''
- This method is asynchronous iff 'callback' is supplied.
- '''
- return self._call(scheduler, "RemoveJob", callback, default,
- timeout, job_id, reason)
-
-# jobserver operations
-
- def get_job_ad(self, job_server, job_id, default=None, timeout=5):
- return self._call(job_server, "GetJobAd", 0, default, timeout, job_id)
-
- def fetch_job_data(self, job_server, job_id, file, start, end,
- default=None, timeout=5):
- return self._call(job_server, "FetchJobData", 0, default, timeout,
- job_id, file, start, end)
-
-# negotiator operations
-
- def set_limit(self, negotiator, lim_name, lim_max, default=None, timeout=5):
- return self._call(negotiator, "SetLimit", 0, default, timeout,
- lim_name, lim_max)
-
- def get_limits(self, negotiator, callback):
- assert callback
- return self._call(negotiator, "GetLimits", callback, 0, 0)
-
- def reconfig(self, negotiator, default=None, timeout=5):
- return self._call(negotiator, "Reconfig", 0, default, timeout)
-
- def set_raw_config(self, negotiator, name, value, default=None, timeout=5):
- return self._call(negotiator, "SetRawConfig", 0, default, timeout,
- name, value)
-
- def get_raw_config(self, negotiator, name, callback):
- assert callback
- return self._call(negotiator, "GetRawConfig", callback, 0, 0, name)
-
-# master operations
-
- def start(self, master, daemon, callback):
- assert callback
- return self._call(master, "Start", callback, 0, 0, daemon)
-
- def stop(self, master, daemon, callback):
- assert callback
- return self._call(master, "Stop", callback, 0, 0, daemon)
-
-# submission operations
-
- def get_job_summaries(self, submission, callback):
- assert callback
- return self._call(submission, "GetJobSummaries", callback, 0, 0)
-
-# Secret private implementation stuff, don't look!
- def _call(self, obj, meth, cb, dflt, tout, *args):
- if cb:
- self.session.call_method(cb, obj, meth, args)
- else:
- try:
- sync = CallSync(log, dflt)
- self.session.call_method(sync.get_completion(),
- obj, meth, args)
- except Exception, e:
- sync.error = e
- log.exception(e)
-
- wait(sync.done, timeout=tout)
- if not sync.got_data and not sync.error:
- sync.error = Exception("Request timed out")
- msg = "QMF call %s timed out" % meth
- log.error(msg)
- return sync
-
Modified: trunk/sage/python/sage/util.py
===================================================================
--- trunk/sage/python/sage/util.py 2011-07-12 20:02:12 UTC (rev 4869)
+++ trunk/sage/python/sage/util.py 2011-07-14 20:22:03 UTC (rev 4870)
@@ -1,7 +1,26 @@
from time import time, sleep
+from threading import Thread, Lock
-class CallSync(object):
+class MethodResult(object):
'''
+ Base type for synchronous method call results.
+ All synchronous methods that implement remote operations
+ in 'sage' will return a result whose type is this class
+ or a descendent of this class.
+ self.data will contain any data returned from the call
+ self.got_data will be True if the call succeeded
+ self.error will be True if the call raised an error
+ self.status will be 'OK' or 0 if the call succeeded and
+ may contain explanatory text otherwise
+ '''
+ def __init__(self):
+ self.data = None
+ self.got_data = False
+ self.error = False
+ self.status = None
+
+class CallSync(MethodResult):
+ '''
General callback object for asynchronous operations.
The 'get_completion' method will return a function that can be
used as a callback when an asynchronous operation completes.
@@ -11,10 +30,8 @@
The 'done' method can be polled to determine if the operation has completed.
'''
def __init__(self, log=None, default=None):
+ super(CallSync, self).__init__()
self.data = default
- self.got_data = False
- self.error = False
- self.status = None
self.log = log
def get_completion(self):
@@ -71,6 +88,10 @@
return self.syncs
def wait(predicate, timeout=30, args=None):
+ '''
+ Simple wait routine that tests predicate() with timeout.
+ If args is not None, calls predicate(args).
+ '''
start = time()
while True:
if args is not None:
@@ -82,3 +103,65 @@
if time() - start > timeout:
return False
sleep(1)
+
+class CallThread(Thread):
+ '''
+ Simple thread that executes result = routine(args).
+ This thread is useful for calling synchronous routines asynchronously.
+ Result is returned by calling callback(result).
+ If routine raises an exception, callback(exception) is called.
+ '''
+ def __init__(self, routine, callback, *args, **kwargs):
+ super(CallThread, self).__init__()
+
+ self.routine = routine
+ self.callback = callback
+ self.args = args
+ self.kwargs = kwargs
+
+ def run(self):
+ try:
+ result = self.routine(*self.args, **self.kwargs)
+ except Exception, e:
+ result = e
+ self.callback(result)
+
+class ObjectPool(object):
+ '''
+ Simple threadsafe pool class that holds up to max_size objects.
+ If the pool is empty when get_object() is called an object will be
+ created and returned; otherwise an object will be removed from
+ the pool and returned.
+ Objects are returned to the pool by a call to return_object(). If the pool
+ contains less than max_size items, a reference to the object will be
+ added to the pool.
+ ObjectPool must be derived from, and create_object must be overridden.
+ If max_size is set to None, the pool size is unlimited.
+ '''
+ def __init__(self, max_size):
+ self.max_size = max_size
+ self.pool = list()
+ self.lock = Lock()
+
+ def get_object(self):
+ self.lock.acquire()
+ if len(self.pool) == 0:
+ self.lock.release()
+ obj = self.create_object()
+ else:
+ obj = self.pool.pop()
+ self.lock.release()
+ return obj
+
+ def return_object(self, obj):
+ self.lock.acquire()
+ if self.max_size is None or len(self.pool) < self.max_size:
+ self.pool.append(obj)
+ self.lock.release()
+
+ def create_object(self):
+ '''
+ This routine must be overridden in a derived class
+ to create appropriate objects for addition to the pool.
+ '''
+ raise Exception("Not implemented")