Changeset 94

Timestamp:

10/18/07 01:19:28 (1 year ago)

Author:

edsuom

Message:

Prototype code for job updating...needs testing

Files:

• projects/AsynCluster/trunk/asyncluster/master/control.py (modified) (9 diffs)
• projects/AsynCluster/trunk/asyncluster/master/jobs.py (modified) (4 diffs)

projects/AsynCluster/trunk/asyncluster/master/control.py

@@ -34 +34 @@
-
-
+def log(msgProto, nodeID, *args):
+    """
+    Logs a message based on the supplied message prototype about some
+    particular node and, if specified as a third argument, some particular
+    user. Any arguments beyond the first three are applied to string
+    substitution of the message prototype.
+    """
+    msg = "N%02d" % nodeID
+    if args:
+        msg = "%s (%s): %s" % (msg, args[0], msgProto)
+        if len(args) > 1:
+            msg = msg % args[1:]
+    else:
+        msg = "%s: %s" % (msg, msgProto)
+    print msg
+
+
-class RemoteCallError(Exception):
-    """
@@ -66 +83 @@
-            hoursLeft -= float(self.updateInterval) / 3600.0
-            self.sessionMap[ID][1] = hoursLeft
-self._
+
-            if hoursLeft >= 0.0:
-                d = self.ctl.nodeRemote(ID, 'setTimeLeft', hoursLeft)
@@ -73 +90 @@
-            d.addCallback(
-                lambda _:
-self._
+
-            dList.append(d)
-        return defer.DeferredList(dList)
-    
-    def _log(self, msgProto, nodeID, *args):
-        msg = "N%02d" % nodeID
-        if args:
-            msg = "%s (%s): %s" % (msg, args[0], msgProto)
-            if len(args) > 1:
-                msg = msg % args[1:]
-        else:
-            msg = "%s: %s" % (msg, msgProto)
-        print msg
-    
-    def view_begin(self, node, userID, password):
@@ -98 +105 @@
-        def gotAuthorizationResult(authorized):
-            if not authorized:
-self._
+
-                return False
-            if ID is None:
-                d = defer.succeed(None)
-            else:
-
-
+
-                d = defer.maybeDeferred(self.end, ID)
-            d.addCallback(lambda _: self.t.restricted(userID))
@@ -112 +118 @@
-        def gotRestrictionResult(restricted):
-            if not restricted:
-self._
+
-                self.sessionMap[node.ID] = [userID, UNRESTRICTED_LOGIN_HOURS]
-                return True
@@ -121 +127 @@
-        def gotHoursAvailable(hours):
-            if hours <= 0.0:
-self._
+
-                return False
-            self.sessionMap[node.ID] = [userID, hours]
-self._
+
-            self.t.recordSessionStartTime(userID)
-            return True
@@ -142 +148 @@
-        if node.ID in self.sessionMap:
-            userID, hoursLeft = self.sessionMap[node.ID]
-self._
+
-            return hoursLeft
-        return 0.0
@@ -163 +169 @@
-        else:
-            userID, hoursLeft = self.sessionMap.pop(ID)
-
-
-
+
-            d = self.t.sessionEnd(userID)
-            if callClient:
@@ -304 +308 @@
-        """
-        return self.ctl.jobber.new(jobCode, niceness)
+
+    def remote_updateJob(self, jobID, callName, *args, **kw):
+        """
+        Updates the job specified by I{jobID} by arranging to dispatch a call
+        to the callable object specified in the job's namespace on each node
+        before it runs another call for that job.
+        """
+        return self.ctl.jobber.update(jobID, callName, *args, **kw)
-    
-    def remote_runJob(self, jobID, callName, *args, **kw):

projects/AsynCluster/trunk/asyncluster/master/jobs.py

@@ -24 +24 @@
-
-import os.path
+
+from zope.interface import implements
-from twisted.internet import reactor
-from twisted.spread import pb
+
+from asynqueue import workers, TaskQueue
-
-
@@ -32 +36 @@
-    I connect to the master TCP server via a UNIX socket and provide an
-    interface for running jobs defined by a Python code file.
+
+    @ivar root: A remote reference to the root object provided by the server.
+    
-    """
-    root = None
@@ -98 +105 @@
-        return d
-
+    def update(self, cmd, *args, **kw):
+        """
+        Arranges for a job update to be done on all present and future nodes
+        before they next run a job. Note that the update will apply to any jobs
+        currently queued up, too!
+
+        The update is done via the specified I{cmd} with any args and keywords
+        supplied.
+        """
+        if hasattr(self, 'jobID'):
+            return self.root.callRemote('updateJob', self.jobID, cmd, *args, **kw)
+        raise Exception("No job registered!")
+
-    def run(self, cmd, *args, **kw):
-        """
@@ -105 +125 @@
-            return self.root.callRemote('runJob', self.jobID, cmd, *args, **kw)
-        raise Exception("No job registered!")
+
+
+class JobWorker(workers.RemoteCallWorker):
+    """
+    Instantiate me with a started instance of L{JobClient} and I'll use its
+    root reference and job runner.
+    """
+    N = 30
+
+    def __init__(self, jobClient):
+        if not hasattr(jobClient, 'jobID'):
+            raise Exception("Supplied job client not started!")
+        self.client = jobClient
+        self.iQualified = [jobClient.jobID]
+        self.startup(jobClient.root)
+
+    def runNow(self, null, task):
+        cmd, args, kw = task.callTuple
+        d = self.client.run(cmd, *args, **kw)
+        job = (task, d)
+        self.jobs.append(job)
+        d.addCallback(self.doneTrying, job)
+        d.addErrback(self.oops)
+        # The task's deferred is NOT returned!
+
+    def stop(self):
+        d = workers.RemoteCallWorker.stop(self)
+        d.addCallback(lambda _: self.client.shutdown())
+        return d
+