Update of /cvsroot/mailman/mailman/Mailman/Queue
In directory sc8-pr-cvs1:/tmp/cvs-serv5952/Mailman/Queue
Modified Files:
Tag: Release_2_1-maint
BounceRunner.py NewsRunner.py OutgoingRunner.py Runner.py
Switchboard.py
Added Files:
Tag: Release_2_1-maint
RetryRunner.py
Log Message:
Backporting from the HEAD -- updated queue runners
--- NEW FILE: RetryRunner.py ---
# Copyright (C) 2003 by the Free Software Foundation, Inc.
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
import time
from Mailman import mm_cfg
from Mailman.Queue.Runner import Runner
from Mailman.Queue.Switchboard import Switchboard
try:
True, False
except NameError:
True = 1
False = 0
class RetryRunner(Runner):
QDIR = mm_cfg.RETRYQUEUE_DIR
SLEEPTIME = mm_cfg.minutes(15)
def __init__(self, slice=None, numslices=1):
Runner.__init__(self, slice, numslices)
self.__outq = Switchboard(mm_cfg.OUTQUEUE_DIR)
def _dispose(self, mlist, msg, msgdata):
# Move it to the out queue for another retry
self.__outq.enqueue(msg, msgdata)
return False
def _snooze(self, filecnt):
# We always want to snooze
time.sleep(self.SLEEPTIME)
Index: BounceRunner.py
===================================================================
RCS file: /cvsroot/mailman/mailman/Mailman/Queue/BounceRunner.py,v
retrieving revision 2.27
retrieving revision 2.27.2.1
diff -u -d -r2.27 -r2.27.2.1
--- BounceRunner.py 2 Dec 2002 14:34:18 -0000 2.27
+++ BounceRunner.py 22 Sep 2003 02:34:10 -0000 2.27.2.1
@@ -17,6 +17,8 @@
"""Bounce queue runner."""
import re
+import time
+
from email.MIMEText import MIMEText
from email.MIMEMessage import MIMEMessage
from email.Utils import parseaddr
@@ -33,12 +35,22 @@
COMMASPACE = ', '
+REGISTER_BOUNCES_EVERY = mm_cfg.minutes(15)
+
class BounceRunner(Runner):
QDIR = mm_cfg.BOUNCEQUEUE_DIR
- # We only do bounce processing once per minute.
- SLEEPTIME = mm_cfg.minutes(1)
+
+ def __init__(self, slice=None, numslices=1):
+ Runner.__init__(self, slice, numslices)
+ # This is a simple sequence of bounce score events. Each entry in the
+ # list is a tuple of (address, day, msg) where day is a tuple of
+ # (YYYY, MM, DD). We'll sort and collate all this information in
+ # _register_bounces() below.
+ self._bounces = {}
+ self._bouncecnt = 0
+ self._next_registration = time.time() + REGISTER_BOUNCES_EVERY
def _dispose(self, mlist, msg, msgdata):
# Make sure we have the most up-to-date state
@@ -84,57 +96,48 @@
# although I'm unsure how that could happen. Possibly ScanMessages()
# can let None's sneak through. In any event, this will kill them.
addrs = filter(None, addrs)
- # Okay, we have some recognized addresses. We now need to register
- # the bounces for each of these. If the bounce came to the site list,
- # then we'll register the address on every list in the system, but
- # note: this could be VERY resource intensive!
- foundp = 0
- listname = mlist.internal_name()
- if listname == mm_cfg.MAILMAN_SITE_LIST:
- foundp = 1
- for listname in Utils.list_names():
- xlist = self._open_list(listname)
- xlist.Load()
- for addr in addrs:
- if xlist.isMember(addr):
- unlockp = 0
- if not xlist.Locked():
- try:
- xlist.Lock(timeout=mm_cfg.LIST_LOCK_TIMEOUT)
- except LockFile.TimeOutError:
- # Oh well, forget aboutf this list
- continue
- unlockp = 1
- try:
- xlist.registerBounce(addr, msg)
- foundp = 1
- xlist.Save()
- finally:
- if unlockp:
- xlist.Unlock()
+ # Store the bounce score events so we can register them periodically
+ today = time.localtime()[:3]
+ events = [(addr, today, msg) for addr in addrs]
+ self._bounces.setdefault(mlist.internal_name(), []).extend(events)
+ self._bouncecnt += len(addrs)
+
+ def _doperiodic(self):
+ now = time.time()
+ if self._next_registration > now or not self._bounces:
+ return
+ # Let's go ahead and register the bounces we've got stored up
+ self._next_registration = now + REGISTER_BOUNCES_EVERY
+ self._register_bounces()
+
+ def _register_bounces(self):
+ syslog('bounce', 'Processing %s queued bounces', self._bouncecnt)
+ # First, get the list of bounces register against the site list. For
+ # these addresses, we want to register a bounce on every list the
+ # address is a member of -- which we don't know yet.
+ sitebounces = self._bounces.get(mm_cfg.MAILMAN_SITE_LIST, [])
+ if sitebounces:
+ listnames = Utils.list_names()
else:
+ listnames = self._bounces.keys()
+ for listname in listnames:
+ mlist = self._open_list(listname)
+ mlist.Lock()
try:
- mlist.Lock(timeout=mm_cfg.LIST_LOCK_TIMEOUT)
- except LockFile.TimeOutError:
- # Try again later
- syslog('bounce', "%s: couldn't get list lock", listname)
- return 1
- else:
- try:
- for addr in addrs:
- if mlist.isMember(addr):
- mlist.registerBounce(addr, msg)
- foundp = 1
- mlist.Save()
- finally:
- mlist.Unlock()
- if not foundp:
- # It means an address was recognized but it wasn't an address
- # that's on any mailing list at this site. BAW: don't forward
- # these, but do log it.
- syslog('bounce', 'bounce message with non-members of %s: %s',
- listname, COMMASPACE.join(addrs))
- #maybe_forward(mlist, msg)
+ events = self._bounces.get(listname, []) + sitebounces
+ for addr, day, msg in events:
+ mlist.registerBounce(addr, msg, day=day)
+ mlist.Save()
+ finally:
+ mlist.Unlock()
+ # Reset and free all the cached memory
+ self._bounces = {}
+ self._bouncecnt = 0
+
+ def _cleanup(self):
+ if self._bounces:
+ self._register_bounces()
+ Runner._cleanup(self)
Index: NewsRunner.py
===================================================================
RCS file: /cvsroot/mailman/mailman/Mailman/Queue/NewsRunner.py,v
retrieving revision 2.14
retrieving revision 2.14.2.1
diff -u -d -r2.14 -r2.14.2.1
--- NewsRunner.py 19 Oct 2002 17:36:28 -0000 2.14
+++ NewsRunner.py 22 Sep 2003 02:34:10 -0000 2.14.2.1
@@ -1,4 +1,4 @@
-# Copyright (C) 2000,2001,2002 by the Free Software Foundation, Inc.
+# Copyright (C) 2000-2003 by the Free Software Foundation, Inc.
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
@@ -45,6 +45,13 @@
""", re.VERBOSE)
+try:
+ True, False
+except NameError:
+ True = 1
+ False = 0
+
+
class NewsRunner(Runner):
QDIR = mm_cfg.NEWSQUEUE_DIR
@@ -60,7 +67,9 @@
conn = None
try:
try:
- conn = nntplib.NNTP(mlist.nntp_host, readermode=1,
+ nntp_host, nntp_port = Utils.nntpsplit(mlist.nntp_host)
+ conn = nntplib.NNTP(nntp_host, nntp_port,
+ readermode=True,
user=mm_cfg.NNTP_USERNAME,
password=mm_cfg.NNTP_PASSWORD)
conn.post(fp)
@@ -79,8 +88,8 @@
# Some other exception occurred, which we definitely did not
# expect, so set this message up for requeuing.
self._log(e)
- return 1
- return 0
+ return True
+ return False
@@ -123,13 +132,13 @@
#
# Our Message-ID format is <[EMAIL PROTECTED]>
msgid = msg['message-id']
- hackmsgid = 1
+ hackmsgid = True
if msgid:
mo = mcre.search(msgid)
if mo:
lname, hname = mo.group('listname', 'hostname')
if lname == mlist.internal_name() and hname == mlist.host_name:
- hackmsgid = 0
+ hackmsgid = False
if hackmsgid:
del msg['message-id']
msg['Message-ID'] = Utils.unique_message_id(mlist)
@@ -155,4 +164,4 @@
for v in values[1:]:
msg[rewrite] = v
# Mark this message as prepared in case it has to be requeued
- msgdata['prepped'] = 1
+ msgdata['prepped'] = True
Index: OutgoingRunner.py
===================================================================
RCS file: /cvsroot/mailman/mailman/Mailman/Queue/OutgoingRunner.py,v
retrieving revision 2.15.2.1
retrieving revision 2.15.2.2
diff -u -d -r2.15.2.1 -r2.15.2.2
--- OutgoingRunner.py 31 Mar 2003 21:49:43 -0000 2.15.2.1
+++ OutgoingRunner.py 22 Sep 2003 02:34:11 -0000 2.15.2.2
@@ -29,11 +29,12 @@
from Mailman import Errors
from Mailman import LockFile
from Mailman.Queue.Runner import Runner
+from Mailman.Queue.Switchboard import Switchboard
from Mailman.Logging.Syslog import syslog
# This controls how often _doperiodic() will try to deal with deferred
# permanent failures. It is a count of calls to _doperiodic()
-DEAL_WITH_PERMFAILURES_EVERY = 1
+DEAL_WITH_PERMFAILURES_EVERY = 10
try:
True, False
@@ -59,6 +60,7 @@
# error log. It gets reset if the message was successfully sent, and
# set if there was a socket.error.
self.__logged = False
+ self.__retryq = Switchboard(mm_cfg.RETRYQUEUE_DIR)
def _dispose(self, mlist, msg, msgdata):
# See if we should retry delivery of this message again.
@@ -98,33 +100,30 @@
# For permanent failures, make a copy of the message for bounce
# handling. I'm not sure this is necessary, or the right thing to
# do.
- pcnt = len(e.permfailures)
- msgcopy = copy.deepcopy(msg)
- self._permfailures.setdefault(mlist, []).extend(
- zip(e.permfailures, [msgcopy] * pcnt))
- # Temporary failures
- if not e.tempfailures:
- # Don't need to keep the message queued if there were only
- # permanent failures.
- return False
- now = time.time()
- recips = e.tempfailures
- last_recip_count = msgdata.get('last_recip_count', 0)
- deliver_until = msgdata.get('deliver_until', now)
- if len(recips) == last_recip_count:
- # We didn't make any progress, so don't attempt delivery any
- # longer. BAW: is this the best disposition?
- if now > deliver_until:
- return False
- else:
- # Keep trying to delivery this message for a while
- deliver_until = now + mm_cfg.DELIVERY_RETRY_PERIOD
- msgdata['last_recip_count'] = len(recips)
- msgdata['deliver_until'] = deliver_until
- msgdata['deliver_after'] = now + mm_cfg.DELIVERY_RETRY_WAIT
- msgdata['recips'] = recips
- # Requeue
- return True
+ if e.permfailures:
+ pcnt = len(e.permfailures)
+ msgcopy = copy.deepcopy(msg)
+ self._permfailures.setdefault(mlist, []).extend(
+ zip(e.permfailures, [msgcopy] * pcnt))
+ # Move temporary failures to the qfiles/retry queue which will
+ # occasionally move them back here for another shot at delivery.
+ if e.tempfailures:
+ now = time.time()
+ recips = e.tempfailures
+ last_recip_count = msgdata.get('last_recip_count', 0)
+ deliver_until = msgdata.get('deliver_until', now)
+ if len(recips) == last_recip_count:
+ # We didn't make any progress, so don't attempt delivery
+ # any longer. BAW: is this the best disposition?
+ if now > deliver_until:
+ return False
+ else:
+ # Keep trying to delivery this message for a while
+ deliver_until = now + mm_cfg.DELIVERY_RETRY_PERIOD
+ msgdata['last_recip_count'] = len(recips)
+ msgdata['deliver_until'] = deliver_until
+ msgdata['recips'] = recips
+ self.__retryq.enqueue(msg, msgdata)
# We've successfully completed handling of this message
return False
@@ -134,6 +133,9 @@
self._permfail_counter += 1
if self._permfail_counter < DEAL_WITH_PERMFAILURES_EVERY:
return
+ self._handle_permfailures()
+
+ def _handle_permfailures(self):
# Reset the counter
self._permfail_counter = 0
# And deal with the deferred permanent failures.
@@ -149,3 +151,7 @@
mlist.Save()
finally:
mlist.Unlock()
+
+ def _cleanup(self):
+ self._handle_permfailures()
+ Runner._cleanup(self)
Index: Runner.py
===================================================================
RCS file: /cvsroot/mailman/mailman/Mailman/Queue/Runner.py,v
retrieving revision 2.25
retrieving revision 2.25.2.1
diff -u -d -r2.25 -r2.25.2.1
--- Runner.py 13 Dec 2002 16:52:07 -0000 2.25
+++ Runner.py 22 Sep 2003 02:34:11 -0000 2.25.2.1
@@ -1,4 +1,4 @@
-# Copyright (C) 1998,1999,2000,2001,2002 by the Free Software Foundation, Inc.
+# Copyright (C) 1998-2003 by the Free Software Foundation, Inc.
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
@@ -31,6 +31,12 @@
from Mailman.Queue.Switchboard import Switchboard
from Mailman.Logging.Syslog import syslog
+try:
+ True, False
+except NameError:
+ True = 1
+ False = 0
+
class Runner:
@@ -44,16 +50,16 @@
self._switchboard = Switchboard(self.QDIR, slice, numslices)
# Create the shunt switchboard
self._shunt = Switchboard(mm_cfg.SHUNTQUEUE_DIR)
- self._stop = 0
+ self._stop = False
def stop(self):
- self._stop = 1
+ self._stop = True
def run(self):
# Start the main loop for this queue runner.
try:
try:
- while 1:
+ while True:
# Once through the loop that processes all the files in
# the queue directory.
filecnt = self._oneloop()
@@ -64,10 +70,10 @@
# If the stop flag is set, we're done.
if self._stop:
break
- # If there were no files to process, then we'll simply
- # sleep for a little while and expect some to show up.
- if not filecnt:
- self._snooze()
+ # Give the runner an opportunity to snooze for a while,
+ # but pass it the file count so it can decide whether to
+ # do more work now or not.
+ self._snooze(filecnt)
except KeyboardInterrupt:
pass
finally:
@@ -79,8 +85,7 @@
# First, list all the files in our queue directory.
# Switchboard.files() is guaranteed to hand us the files in FIFO
# order. Return an integer count of the number of files that were
- # available for this qrunner to process. A non-zero value tells run()
- # not to snooze for a while.
+ # available for this qrunner to process.
files = self._switchboard.files()
for filebase in files:
# Ask the switchboard for the message and metadata objects
@@ -105,12 +110,12 @@
self._onefile(msg, msgdata)
except Exception, e:
self._log(e)
- syslog('error', 'SHUNTING: %s', filebase)
# Put a marker in the metadata for unshunting
msgdata['whichq'] = self._switchboard.whichq()
- self._shunt.enqueue(msg, msgdata)
+ filebase = self._shunt.enqueue(msg, msgdata)
+ syslog('error', 'SHUNTING: %s', filebase)
# Other work we want to do each time through the loop
- Utils.reap(self._kids, once=1)
+ Utils.reap(self._kids, once=True)
self._doperiodic()
if self._shortcircuit():
break
@@ -172,7 +177,7 @@
mlist = self._listcache.get(listname)
if not mlist:
try:
- mlist = MailList.MailList(listname, lock=0)
+ mlist = MailList.MailList(listname, lock=False)
except Errors.MMListError, e:
syslog('error', 'error opening list: %s\n%s', listname, e)
return None
@@ -219,18 +224,18 @@
from the Runner's hash slice processing loop. You can do whatever
special periodic processing you want here, and the return value is
irrelevant.
-
"""
pass
- def _snooze(self):
- """Sleep for a little while, because there was nothing to do.
+ def _snooze(self, filecnt):
+ """Sleep for a little while.
- This is called from the Runner's main loop, but only when the last
- processing loop had no work to do (i.e. there were no messages in it's
- little slice of hash space).
+ filecnt is the number of messages in the queue the last time through.
+ Sub-runners can decide to continue to do work, or sleep for a while
+ based on this value. By default, we only snooze if there was nothing
+ to do last time around.
"""
- if self.SLEEPTIME <= 0:
+ if filecnt or self.SLEEPTIME <= 0:
return
time.sleep(self.SLEEPTIME)
Index: Switchboard.py
===================================================================
RCS file: /cvsroot/mailman/mailman/Mailman/Queue/Switchboard.py,v
retrieving revision 2.15.2.2
retrieving revision 2.15.2.3
diff -u -d -r2.15.2.2 -r2.15.2.3
--- Switchboard.py 20 Apr 2003 04:02:07 -0000 2.15.2.2
+++ Switchboard.py 22 Sep 2003 02:34:11 -0000 2.15.2.3
@@ -50,7 +50,20 @@
# 20 bytes of all bits set, maximum sha.digest() value
shamax = 0xffffffffffffffffffffffffffffffffffffffffL
-SAVE_MSGS_AS_PICKLES = 1
+try:
+ True, False
+except NameError:
+ True = 1
+ False = 0
+
+# This flag causes Mailman to fsync() the file after writing and flushing its
+# contents. While this ensures the data is written to disk, avoiding data
+# loss, it is a huge performance killer.
+SYNC_AFTER_WRITE = False
+# This flag causes messages to be written as pickles (when True) or text files
+# (when False). Pickles are more efficient because the message doesn't need
+# to be re-parsed every time it's unqueued, but pickles are not human readable.
+SAVE_MSGS_AS_PICKLES = True
@@ -127,6 +140,7 @@
tmpfile = dbfile + '.tmp'
self._ext_write(tmpfile, data)
os.rename(tmpfile, dbfile)
+ return filebase
def dequeue(self, filebase):
# Calculate the .db and .msg filenames from the given filebase.
@@ -239,6 +253,10 @@
else:
dict[attr] = repr(fval)
marshal.dump(dict, fp)
+ # Make damn sure that the data we just wrote gets flushed to disk
+ fp.flush()
+ if SYNC_AFTER_WRITE:
+ os.fsync(fp.fileno())
fp.close()
def _ext_read(self, filename):
@@ -307,6 +325,10 @@
os.umask(omask)
for k, v in dict.items():
print >> fp, '%s = %s' % (k, repr(v))
+ # Make damn sure that the data we just wrote gets flushed to disk
+ fp.flush()
+ if SYNC_AFTER_WRITE:
+ os.fsync(fp.fileno())
fp.close()
def _ext_read(self, filename):
_______________________________________________
Mailman-checkins mailing list
[EMAIL PROTECTED]
http://mail.python.org/mailman/listinfo/mailman-checkins
