A major rewrite to 1) better support the new mailmanctl script and 2)

to hopefully lay the groundwork for a MM2.0.x compatible cron-invoked
qrunner process.

This qrunner responds to SIGINT, SIGTERM, and SIGHUP as described in
the mailmanctl docstring.  It also added a more robust -r/--runner
flag to support mailmanctl invoking multiple qrunners per queue via
slicing.

The default now is to run continuously, unless -o/--once is given, and
if multiple -r/--runner flags are given, they will be run in
roundrobin format.  This should be all we need to implement a
backwards compatible cron/qrunner script.

1 files changed, 173 insertions, 23 deletions

diff --git a/bin/qrunner b/bin/qrunner
index 77055fd4b..f26b63898 100644
--- a/bin/qrunner
+++ b/bin/qrunner
@@ -16,35 +16,61 @@
 # along with this program; if not, write to the Free Software
 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
 
-"""One-shot qrunner, er, runner.
+"""Run one or more qrunners, once or repeatedly.
 
-This script is primarily for debugging purposes, and is used to invoke a
-specific sub-qrunner.  The invoked qrunner will run once through its
-processing loop and then exit.  You can only call this script if there is no
-master qrunner daemon running.
+Each named runner class is run in round-robin fashion.  In other words, the
+first named runner is run to consume all the files currently in its
+directory.  When that qrunner is done, the next one is run to consume all the
+files in /its/ directory, and so on.  The number of total iterations can be
+given on the command line.
 
-Usage: %(PROGRAM)s [options] runnername
+Usage: %(PROGRAM)s [options]
 
 Options:
 
+    -r runner[:slice:range]
+    --runner=runner[:slice:range]
+        Run the named qrunner, which must be one of the strings returned by
+        the -l option.  Optional slice:range if given, is used to assign
+        multiple qrunner processes to a queue.  range is the total number of
+        qrunners for this queue while slice is the number of this qrunner from
+        [0..range).
+
+        If using the slice:range form, you better make sure that each qrunner
+        for the queue is given the same range value.  If slice:runner is not
+        given, then 1:1 is used.
+
+        Multiple -r options may be given, in which case each qrunner will run
+        once in round-robin fashion.  The special runner `All' is shorthand
+        for a qrunner for each listed by the -l option.
+
+    --once
+    -o
+        Run each named qrunner exactly once through its main loop.  Otherwise,
+        each qrunner runs indefinitely, until the process receives a SIGTERM
+        or SIGINT.
+
     -l/--list
-        Shows the available qrunner names.
+        Shows the available qrunner names and exit.
+
+    -v/--verbose
+        Spit out more debugging information to the logs/qrunner log file.
 
     -h/--help
         Print this message and exit.
 
-runnername is required unless -l or -h is given, and it must be one of the
-names displayed by the -l switch.
+runner is required unless -l or -h is given, and it must be one of the names
+displayed by the -l switch.
 """
 
 import sys
 import getopt
+import signal
 
 import paths
 from Mailman import mm_cfg
-from Mailman import LockFile
-from Mailman.Queue import Control
 from Mailman.i18n import _
+from Mailman.Logging.Syslog import syslog
 
 PROGRAM = sys.argv[0]
 COMMASPACE = ', '
@@ -59,12 +85,63 @@ def usage(code, msg=''):
 
 
 
+def make_qrunner(name, slice, range, once=0):
+    modulename = 'Mailman.Queue.' + name
+    try:
+        __import__(modulename)
+    except ImportError, e:
+        usage(1, e)
+    qrclass = getattr(sys.modules[modulename], name)
+    if once:
+        # Subclass to hack in the setting of the stop flag in _doperiodic()
+        class Once(qrclass):
+            def _doperiodic(self):
+                self.stop()
+        qrunner = Once(slice, range)
+    else:
+        qrunner = qrclass(slice, range)
+    return qrunner
+
+
+
+def set_signals(loop):
+    # Set up the SIGTERM handler for stopping the loop
+    def sigterm_handler(signum, frame, loop=loop):
+        # Exit the qrunner cleanly
+        loop.stop()
+        loop.status = signal.SIGTERM
+        syslog('qrunner', '%s qrunner caught SIGTERM.  Stopping.', loop.name())
+    signal.signal(signal.SIGTERM, sigterm_handler)
+    # Set up the SIGINT handler for stopping the loop.  For us, SIGINT is
+    # the same as SIGTERM, but our parent treats the exit statuses
+    # differently (it restarts a SIGINT but not a SIGTERM).
+    def sigint_handler(signum, frame, loop=loop):
+        # Exit the qrunner cleanly
+        loop.stop()
+        loop.status = signal.SIGINT
+        syslog('qrunner', '%s qrunner caught SIGINT.  Stopping.', loop.name())
+    signal.signal(signal.SIGINT, sigint_handler)
+    # SIGHUP just tells us to close our log files.  They'll be
+    # automatically reopened at the next log print :)
+    def sighup_handler(signum, frame, loop=loop):
+        syslog.close()
+        syslog('qrunner', '%s qrunner caught SIGHUP.  Reopening logs.',
+               loop.name())
+    signal.signal(signal.SIGHUP, sighup_handler)
+
+
+
 def main():
     try:
-        opts, args = getopt.getopt(sys.argv[1:], 'hl', ['help', 'list'])
+        opts, args = getopt.getopt(
+            sys.argv[1:], 'hlor:v',
+            ['help', 'list', 'once', 'runner=', 'verbose'])
     except getopt.error, msg:
         usage(1, msg)
 
+    once = 0
+    runners = []
+    verbose = 0
     for opt, arg in opts:
         if opt in ('-h', '--help'):
             usage(0)
@@ -74,21 +151,94 @@ def main():
                     name = runnername[:-len('Runner')]
                 else:
                     name = runnername
-                print _('-r %(name)s runs the %(runnername)s qrunner')
+                print _('%(name)s runs the %(runnername)s qrunner')
+            print _('All runs all the above qrunners')
             sys.exit(0)
+        elif opt in ('-o', '--once'):
+            once = 1
+        elif opt in ('-r', '--runner'):
+            runnerspec = arg
+            parts = runnerspec.split(':')
+            if len(parts) == 1:
+                runner = parts[0]
+                slice = 1
+                range = 1
+            elif len(parts) == 3:
+                runner = parts[0]
+                try:
+                    slice = int(parts[1])
+                    range = int(parts[2])
+                except ValueError:
+                    usage(1, 'Bad runner specification: %(runnerspec)s')
+            else:
+                usage(1, 'Bad runner specification: %(runnerspec)s')
+            if runner == 'All':
+                for runnername, slices in mm_cfg.QRUNNERS:
+                    runners.append((runnername, slice, range))
+            else:
+                if runner.endswith('Runner'):
+                    runners.append((runner, slice, range))
+                else:
+                    runners.append((runner + 'Runner', slice, range))
+        elif opt in ('-v', '--verbose'):
+            verbose = 1
 
-    if len(args) < 1:
+    if len(args) <> 0:
+        usage(1)
+    if len(runners) == 0:
         usage(1, _('No runner name given.'))
-    elif len(args) > 1:
-        command = COMMASPACE.join(args)
-        usage(1, _('Bad command: %(command)s'))
 
-    try:
-        Control.start(0, args[0])
-    except ImportError, e:
-        print >> sys.stderr, e
-    except LockFile.TimeOutError:
-        print >> sys.stderr, 'Another qrunner is already running, exiting.'
+    # Fast track for one infinite runner
+    if len(runners) == 1 and not once:
+        qrunner = make_qrunner(*runners[0])
+        class Loop:
+            status = 0
+            def __init__(self, qrunner):
+                self.__qrunner = qrunner
+            def name(self):
+                return self.__qrunner.__class__.__name__
+            def stop(self):
+                self.__qrunner.stop()
+        loop = Loop(qrunner)
+        set_signals(loop)
+        # Now start up the main loop
+        syslog('qrunner', '%s qrunner started.', loop.name())
+        qrunner.run()
+        syslog('qrunner', '%s qrunner exiting.', loop.name())
+    else:
+        # Anything else we have to handle a bit more specially
+        qrunners = []
+        for runner, slice, range in runners:
+            qrunner = make_qrunner(runner, slice, range, 1)
+            qrunners.append(qrunner)
+        # This class is used to manage the main loop
+        class Loop:
+            status = 0
+            def __init__(self):
+                self.__isdone = 0
+            def name(self):
+                return 'Main loop'
+            def stop(self):
+                self.__isdone = 1
+            def isdone(self):
+                return self.__isdone
+        loop = Loop()
+        set_signals(loop)
+        syslog('qrunner', 'Main qrunner loop started.')
+        while not loop.isdone():
+            for qrunner in qrunners:
+                # In case the SIGTERM came in the middle of this iteration
+                if loop.isdone():
+                    break
+                if verbose:
+                    syslog('qrunner', 'Now doing a %s qrunner iteration',
+                           qrunner.__class__.__bases__[0].__name__)
+                qrunner.run()
+            if once:
+                break
+        syslog('qrunner', 'Main qrunner loop exiting.')
+    # All done
+    sys.exit(loop.status)