Changeset 126

Timestamp:

02/22/08 22:41:33 (9 months ago)

Author:

edsuom

Message:

Worker specialties, improved timouts

Files:

• projects/AsynQueue/trunk/asynqueue/base.py (modified) (3 diffs)
• projects/AsynQueue/trunk/asynqueue/jobs.py (modified) (6 diffs)
• projects/AsynQueue/trunk/asynqueue/tasks.py (modified) (6 diffs)
• projects/AsynQueue/trunk/asynqueue/test/util.py (modified) (2 diffs)
• projects/AsynQueue/trunk/asynqueue/workers.py (modified) (1 diff)
• projects/AsynQueue/trunk/setup.py (modified) (2 diffs)

projects/AsynQueue/trunk/asynqueue/base.py

@@ -26 +26 @@
-from zope.interface import implements
-from twisted.python import failure
-
+
+
+
+
+
+
+
+
-
-import tasks
@@ -328 +335 @@
-          task has been queued up.
-
+        @keyword timeout: A timeout interval in seconds from when a worker gets
+          a task assignment for the call, after which the call will be retried.
+
-        """
-        def oneLessPending(result):
@@ -351 +361 @@
-        niceness = kw.pop('niceness', 0)
-        series = kw.pop('series', None)
-
+
+
-        if kw.pop('doNext', False):
-            task.priority = -1000000

projects/AsynQueue/trunk/asynqueue/jobs.py

@@ -25 +25 @@
-from zope.interface import implements, Interface
-from twisted.internet import defer, reactor
+from twisted.python import reflect
+from twisted.python.rebuild import rebuild
-from twisted.python.failure import Failure
-from twisted.python import reflect
-from twisted.spread import pb, flavors
+# Use C Deferreds if possible, for efficiency
+try:
+    from twisted.internet import cdefer
+except:
+    pass
+else:
+    defer.Deferred = cdefer.Deferred
-
-import base, workers
@@ -89 +97 @@
-            if cls.__module__ not in modules:
-                modules.append(cls.__module__)
-reloa
+buil
-        # since the last run
-        for module in modules:
-            try:
-load(reflect.namedModule(module)
+build(reflect.namedModule(module), doLog=False
-            except:
-                pass
@@ -221 +229 @@
-    
-    """
-2
+1
-    
-    def __init__(self, queue=None):
@@ -409 +417 @@
-        All keywords except for the following are passed to the call:
-
-
-
+
+
+
-          
-        @note: The task object generated contains the name of a callable (as a
@@ -424 +433 @@
-                kw['doNext'] = True
-            dq = self.queue.call(callName, *args, **kw)
-
-
-
-
-
+
-            dq.addCallback(jobRan)
-
+
-        def jobRan(result):
-            status, result = result
- and jobID in self.callsPending
-[jobID]
+
+.get(jobID, [])
-                    del self.callsPending[jobID][d]
-                    d.callback(result)
+            elif result == 'Timeout':
+                log("Timeout on job %d, retrying", jobID)
+                tryAgain()
-            else:
-                log("Error running job %d:\n%s", jobID, result)
-                tryAgain()
-
-        def jobFailed(failure):
-            return False, failure.getTraceback()
-
-        def tryAgain():
@@ -456 +461 @@
-        if jobID not in self.jobs:
-            raise ValueError("No job '%s' registered" % jobID)
-        timeout = kw.pop('timeout', None)
-        kw['series'] = jobID
-        kw['niceness'] = self.jobs[jobID][1]

projects/AsynQueue/trunk/asynqueue/tasks.py

@@ -24 +24 @@
-# Imports
-from twisted.internet import defer, reactor
+# Use C Deferreds if possible, for efficiency
+try:
+    from twisted.internet import cdefer
+except:
+    pass
+else:
+    defer.Deferred = cdefer.Deferred
-
-from workers import IWorker
@@ -46 +53 @@
-
-    """
-
+, timeout=None
-        if not isinstance(args, (tuple, list)):
-            raise TypeError("Second argument 'args' isn't a sequence")
@@ -55 +62 @@
-        self.series = series
-        self.d = defer.Deferred()
+        self.timeout = timeout
+
+    def startTimer(self):
+        if self.timeout:
+            self.callID = reactor.callLater(self.timeout, self.timedout)
+        else:
+            self.callID = None
-
-    def callback(self, result):
+        if self.callID:
+            self.callID.cancel()
+            self.callID = None
-        if not self.d.called:
-            self.d.callback(result)
-
-    def errback(self, result):
+        if self.callID:
+            self.callID.cancel()
+            self.callID = None
-        self.d.errback(result)
+
+    def timedout(self):
+        if not self.d.called:
+            self.d.callback((False, "Timeout"))
+        self.callID = None
-    
-    def __repr__(self):
@@ -105 +130 @@
-        self.seriesNumbers = {}
-
-
+, timeout=None
-        """
-        Call this to obtain a L{Task} instance that will run in the specified
@@ -131 +156 @@
-        positivized = niceness + 20
-        priority = self._serial(series) * (1 + (float(positivized)/10)**2)
-
+, timeout
-    
-    def _serial(self, series):
@@ -176 +201 @@
-        """
-        self.d.callback(None)
+        self.task.startTimer()
-        return worker.run(self.task)
-

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

@@ -38 +38 @@
-
-class MockTask(object):
-
+, timeout=None
-        self.ran = False
-        self.callTuple = (f, args, kw)
@@ -52 +52 @@
-    def __str__(self):
-        return str(self.callTuple[0])
+
+    def startTimer(self):
+        pass
-
-

projects/AsynQueue/trunk/asynqueue/workers.py

@@ -118 +118 @@
-    """
-    I implement an L{IWorker} that runs tasks in a dedicated worker thread.
+
+    You can define one or more specialties that I am qualified to handle with
+    string arguments.
-    """
-    implements(IWorker)
-    cQualified = []
-
-
+, *specialties
-        import threading
-[]
+list(specialties)
-        self.event = threading.Event()
-        self.thread = threading.Thread(target=self._loop)

projects/AsynQueue/trunk/setup.py

@@ -30 +30 @@
-
-### Define setup options
-2
+3
-      'license':'GPL',
-      'platforms':'OS Independent',
@@ -47 +47 @@
-kw['keywords'] = [
-    'Twisted', 'asynchronous', 'threads',
-
+
+
-
-kw['classifiers'] = [
-    'Development Status :: 4 - Beta',
+
-    'Intended Audience :: Developers',
+    'Intended Audience :: Science/Research',
+
-    'License :: OSI Approved :: GNU General Public License (GPL)',
-    'Operating System :: OS Independent',
-    'Programming Language :: Python',
+
+    'Topic :: System :: Distributed Computing',
+    'Topic :: Software Development :: Object Brokering',
-    'Topic :: Software Development :: Libraries :: Python Modules',
-    ]
+
-
-kw['description'] = " ".join("""