round robin

Author: jaredmauch

Mailman/Defaults.py.in

@@ -933,6 +933,11 @@ DEFAULT_RESPOND_TO_POST_REQUESTS = Yes
 # BAW: Eventually we may support weighted hash spaces.
 # BAW: Although not enforced, the # of slices must be a power of 2
 
+# Distribution method for queue runners: 'hash' (default) or 'round_robin'
+# Hash-based distribution ensures same message always goes to same runner
+# Round-robin distribution provides more even load distribution
+QUEUE_DISTRIBUTION_METHOD = 'hash'
+
 QRUNNERS = [
     ('ArchRunner',     1), # messages for the archiver
     ('BounceRunner',   2), # for processing the qfile/bounces directory

Mailman/Queue/Runner.py

@@ -43,7 +43,8 @@ def __init__(self, slice=None, numslices=1):
         self._kids = {}
         # Create our own switchboard.  Don't use the switchboard cache because
         # we want to provide slice and numslice arguments.
-        self._switchboard = Switchboard(self.QDIR, slice, numslices, True)
+        distribution = getattr(mm_cfg, 'QUEUE_DISTRIBUTION_METHOD', 'hash')
+        self._switchboard = Switchboard(self.QDIR, slice, numslices, True, distribution)
         # Create the shunt switchboard
         self._shunt = Switchboard(mm_cfg.SHUNTQUEUE_DIR)
         self._stop = False

Mailman/Queue/Switchboard.py

@@ -65,8 +65,9 @@
 
 
 class Switchboard:
-    def __init__(self, whichq, slice=None, numslices=1, recover=False):
+    def __init__(self, whichq, slice=None, numslices=1, recover=False, distribution='hash'):
         self.__whichq = whichq
+        self.__distribution = distribution
         # Create the directory if it doesn't yet exist.
         # FIXME
         omask = os.umask(0)                       # rwxrws---
@@ -82,8 +83,13 @@ def __init__(self, whichq, slice=None, numslices=1, recover=False):
         self.__upper = None
         # BAW: test performance and end-cases of this algorithm
         if numslices != 1:
-            self.__lower = (((shamax+1) * slice) / numslices)
-            self.__upper = ((((shamax+1) * (slice+1)) / numslices)) - 1
+            if distribution == 'hash':
+                self.__lower = (((shamax+1) * slice) / numslices)
+                self.__upper = ((((shamax+1) * (slice+1)) / numslices)) - 1
+            elif distribution == 'round_robin':
+                self.__slice = slice
+                self.__numslices = numslices
+            # Add more distribution methods here as needed
         if recover:
             self.recover_backup_files()
 
@@ -105,7 +111,23 @@ def enqueue(self, _msg, _metadata={}, **_kws):
         else:
             protocol = 0
             msgsave = pickle.dumps(str(_msg), protocol, fix_imports=True)
-        hashfood = msgsave + listname.encode() + repr(now).encode()
+        
+        # Choose distribution method
+        if self.__distribution == 'round_robin':
+            # Use a simple counter for round-robin distribution
+            import threading
+            if not hasattr(self, '_counter'):
+                self._counter = 0
+                self._counter_lock = threading.Lock()
+            
+            with self._counter_lock:
+                self._counter = (self._counter + 1) % self.__numslices
+                current_slice = self._counter
+            hashfood = msgsave + listname.encode() + repr(now).encode() + str(current_slice).encode()
+        else:
+            # Default hash-based distribution
+            hashfood = msgsave + listname.encode() + repr(now).encode()
+        
         # Encode the current time into the file name for FIFO sorting in
         # files().  The file name consists of two parts separated by a `+':
         # the received time for this message (i.e. when it first showed up on
@@ -192,14 +214,26 @@ def files(self, extension='.pck'):
             if ext != extension:
                 continue
             when, digest = filebase.split('+')
-            # Throw out any files which don't match our bitrange.  BAW: test
-            # performance and end-cases of this algorithm.  MAS: both
-            # comparisons need to be <= to get complete range.
-            if lower is None or (lower <= int(digest, 16) <= upper):
-                key = float(when)
-                while key in times:
-                    key += DELTA
-                times[key] = filebase
+            
+            # Choose distribution method for file filtering
+            if self.__distribution == 'round_robin':
+                # For round-robin, use modulo of digest to determine slice
+                slice_num = int(digest, 16) % self.__numslices
+                if slice_num == self.__slice:
+                    key = float(when)
+                    while key in times:
+                        key += DELTA
+                    times[key] = filebase
+            else:
+                # Default hash-based distribution
+                # Throw out any files which don't match our bitrange.  BAW: test
+                # performance and end-cases of this algorithm.  MAS: both
+                # comparisons need to be <= to get complete range.
+                if lower is None or (lower <= int(digest, 16) <= upper):
+                    key = float(when)
+                    while key in times:
+                        key += DELTA
+                    times[key] = filebase
         # FIFO sort
         keys = list(times.keys())
         keys.sort()