Changeset 95

Timestamp:

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

Author:

edsuom

Message:

Prototype code for job updating...needs testing

Files:

• projects/AsynQueue/trunk/asynqueue/jobs.py (modified) (7 diffs)
• projects/AsynQueue/trunk/asynqueue/test/test_jobs.py (modified) (2 diffs)
• projects/AsynQueue/trunk/asynqueue/workers.py (modified) (9 diffs)

projects/AsynQueue/trunk/asynqueue/jobs.py

@@ -144 +144 @@
-        workers.RemoteCallWorker.__init__(self, *args, **kw)
-
-_
+
-        funcName, args, kw = task.callTuple
-        d = self.remoteCaller('runJob', task.series, funcName, *args, **kw)
-        job = (task, d)
-        self.jobs.append(job)
-_
+
-        # This task's deferred is NOT returned!
-
@@ -174 +174 @@
-    def __init__(self, queue=None):
-        self.jobs  = {}
+        self.updates = {}
-        if queue is None:
-            self.queue = base.TaskQueue()
@@ -228 +229 @@
-            if status is False:
-                mutable.append(None)
+            elif jobID in self.updates:
+                return self._runUpdate(jobID, worker)
-        
-        def allDone(null):
@@ -281 +284 @@
-        d.addCallback(lambda _: jobID)
-        return d
+
+    def _runUpdate(self, jobID, worker):
+        funcName, args, kw = self.updates[jobID]
+        return worker.remoteCaller('runJob', jobID, funcName, *args, **kw)
+
+    def update(self, jobID, callName, *args, **kw):
+        """
+        """
+        self.updates[jobID] = callName, args, kw
+        dList = [
+            self._runUpdate(jobID, worker)
+            for worker in self.queue.workers()]
+        return defer.DeferredList(dList)
-    
-    def run(self, jobID, callName, *args, **kw):
@@ -307 +323 @@
-                return result
-            log("Error running job %d:\n%s", jobID, result)
-
+
+
+
+
+
-        jobID = int(jobID)
-        if jobID not in self.jobs:
@@ -314 +334 @@
-        kw['niceness'] = self.jobs[jobID][1]
-        d = self.queue.call(callName, *args, **kw)
-Both(jobRan
+Callbacks(jobRan, jobFailed
-        return d
-
@@ -324 +344 @@
-        self.queue.cancelSeries(jobID)
-        self.jobs.pop(jobID, None)
-
-
+
+
+

projects/AsynQueue/trunk/asynqueue/test/test_jobs.py

@@ -189 +189 @@
-        return self._attach(bogus=True).addCallback(self.failUnlessEqual, None)
-
+    def test_attachChild_withUpdate(self):
+        self.fail(
+            "Test that attached child gets updated before running any jobs")
+
-    def test_new(self):
-        def check(jobID):
@@ -234 +238 @@
-        yield defer.waitForDeferred(defer.DeferredList(dList))
-        self.failUnlessEqual(results, range(10))
+
+    def test_update(self):
+        self.fail("Test updates before next job runs")

projects/AsynQueue/trunk/asynqueue/workers.py

@@ -25 +25 @@
-from twisted.python import failure
-from twisted.internet import defer, reactor
+from twisted.spread import pb
-
-import errors
@@ -225 +226 @@
-
-    def __init__(self, remoteReference, N=3, noTypeCheck=False):
-
+
+
+
-        # Check supplied remote reference object
-        if not noTypeCheck:
@@ -233 +236 @@
-                raise TypeError(
-                    "You must construct me with a PB RemoteReference")
+        self.startup(remoteReference)
+
+    def startup(self, remoteReference):
+        """
+        Starts things up with the remote reference in hand. Useful to have this
+        as a separate method when you're subclassing and doing difference
+        constructor stuff.
+        """
-        # Setup resignation-upon-disconnect
-        self.resignators = []
-        self.disconnectErrors = (pb.DeadReferenceError, pb.PBConnectionLost)
-
-
-
-
-
+
-        # Prepare the run request queue
-        self.jobs = []
@@ -247 +254 @@
-            self.runRequestQueue.put(None)
-
-_
+
-        suffix, args, kw = task.callTuple
-        d = self.remoteCaller(suffix, *args, **kw)
-        job = (task, d)
-        self.jobs.append(job)
-_
-_
+
+
-        # The task's deferred is NOT returned!
-
-_
+
-        if failure.check(*self.disconnectErrors):
-_
+
-        else:
-            return failure
-    
-_
+
-        self.jobs.remove(job)
-        self.runRequestQueue.put(None)
@@ -268 +275 @@
-        task.d.callback(result)
-    
-_
+
-        while self.resignators:
-            callableObject = self.resignators.pop()
@@ -291 +298 @@
-        if getattr(self, 'isShuttingDown', False):
-            raise errors.QueueRunError
-_
+
-    
-    def stop(self):
@@ -329 +336 @@
-        self.suffixCache = []
-
-_
+
-        nameListing = [x.__name__ for x in items]
-        nameListing[-1] = "or " + nameListing[-1]
@@ -335 +342 @@
-        return joinString.join(nameListing)
-
-_
+
-        for interface in self.interfaces:
-            for attrName in interface:
@@ -341 +348 @@
-                    self.suffixCache.append(suffix)
-                    return
-_
+
-        raise AttributeError(
-            "No remote method *_%s provided by interface %s" % (suffix, names))
-
-_
+
-        suffix, args, kw = task.callTuple
-        if suffix not in self.suffixCache:
-_
+
-        d = self.remoteCaller(suffix, *args, **kw)
-        job = (task, d)
-        self.jobs.append(job)
-_
+
-        # The task's deferred is NOT returned!