Major refactoring of queue handlers to support running multiple sites in one daemon.

Key changes:
* Initialization code moved from common.php to StatusNet class;
  can now switch configurations during runtime.
* As a consequence, configuration files must now be idempotent...
  Be careful with constant, function or class definitions.
* Control structure for daemons/QueueManager/QueueHandler has been refactored;
  the run loop is now managed by IoMaster run via scripts/queuedaemon.php
  IoManager subclasses are woken to handle socket input or polling, and may
  cover multiple sites.
* Plugins can implement notice queue handlers more easily by registering a
  QueueHandler class; no more need to add a daemon.

The new QueueDaemon runs from scripts/queuedaemon.php:

* This replaces most of the old *handler.php scripts; they've been refactored
  to the bare handler classes.
* Spawns multiple child processes to spread load; defaults to CPU count on
  Linux and Mac OS X systems, or override with --threads=N
* When multithreaded, child processes are automatically respawned on failure.
* Threads gracefully shut down and restart when passing a soft memory limit
  (defaults to 90% of memory_limit), limiting damage from memory leaks.
* Support for UDP-based monitoring: http://www.gitorious.org/snqmon

Rough control flow diagram:
QueueDaemon -> IoMaster -> IoManager
                           QueueManager [listen or poll] -> QueueHandler
                           XmppManager [ping & keepalive]
                           XmppConfirmManager [poll updates]

Todo:

* Respawning features not currently available running single-threaded.
* When running single-site, configuration changes aren't picked up.
* New sites or config changes affecting queue subscriptions are not yet
  handled without a daemon restart.
* SNMP monitoring output to integrate with general tools (nagios, ganglia)
* Convert XMPP confirmation message sends to use stomp queue instead of polling
* Convert xmppdaemon.php to IoManager?
* Convert Twitter status, friends import polling daemons to IoManager
* Clean up some error reporting and failure modes
* May need to adjust queue priorities for best perf in backlog/flood cases

Detailed code history available in my daemon-work branch:
http://www.gitorious.org/~brion/statusnet/brion-fixes/commits/daemon-work

10 files changed, 289 insertions, 706 deletions

diff --git a/scripts/enjitqueuehandler.php b/scripts/enjitqueuehandler.php
deleted file mode 100755
index afcac539a..000000000
--- a/scripts/enjitqueuehandler.php
+++ /dev/null
@@ -1,121 +0,0 @@
-#!/usr/bin/env php
-<?php
-/*
- * StatusNet - the distributed open-source microblogging tool
- * Copyright (C) 2008, 2009, StatusNet, Inc.
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as published by
- * the Free Software Foundation, either version 3 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 Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_ENJIT_HELP
-Daemon script for watching new notices and posting to enjit.
-
-    -i --id           Identity (default none)
-
-END_OF_ENJIT_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-
-require_once INSTALLDIR . '/lib/mail.php';
-require_once INSTALLDIR . '/lib/queuehandler.php';
-
-set_error_handler('common_error_handler');
-
-class EnjitQueueHandler extends QueueHandler
-{
-    function transport()
-    {
-        return 'enjit';
-    }
-
-    function start()
-    {
-        $this->log(LOG_INFO, "Starting EnjitQueueHandler");
-        $this->log(LOG_INFO, "Broadcasting to ".common_config('enjit', 'apiurl'));
-        return true;
-    }
-
-    function handle_notice($notice)
-    {
-
-        $profile = Profile::staticGet($notice->profile_id);
-
-        $this->log(LOG_INFO, "Posting Notice ".$notice->id." from ".$profile->nickname);
-
-        if ( ! $notice->is_local ) {
-            $this->log(LOG_INFO, "Skipping remote notice");
-            return "skipped";
-        }
-
-        #
-        # Build an Atom message from the notice
-        #
-        $noticeurl = common_local_url('shownotice', array('notice' => $notice->id));
-        $msg = $profile->nickname . ': ' . $notice->content;
-
-        $atom  = "<entry xmlns='http://www.w3.org/2005/Atom'>\n";
-        $atom .= "<apisource>".common_config('enjit','source')."</apisource>\n";
-        $atom .= "<source>\n";
-        $atom .= "<title>" . $profile->nickname . " - " . common_config('site', 'name') . "</title>\n";
-        $atom .= "<link href='" . $profile->profileurl . "'/>\n";
-        $atom .= "<link rel='self' type='application/rss+xml' href='" . common_local_url('userrss', array('nickname' => $profile->nickname)) . "'/>\n";
-        $atom .= "<author><name>" . $profile->nickname . "</name></author>\n";
-        $atom .= "<icon>" . $profile->avatarUrl(AVATAR_PROFILE_SIZE) . "</icon>\n";
-        $atom .= "</source>\n";
-        $atom .= "<title>" . htmlspecialchars($msg) . "</title>\n";
-        $atom .= "<summary>" . htmlspecialchars($msg) . "</summary>\n";
-        $atom .= "<link rel='alternate' href='" . $noticeurl . "' />\n";
-        $atom .= "<id>". $notice->uri . "</id>\n";
-        $atom .= "<published>".common_date_w3dtf($notice->created)."</published>\n";
-        $atom .= "<updated>".common_date_w3dtf($notice->modified)."</updated>\n";
-        $atom .= "</entry>\n";
-
-        $url  = common_config('enjit', 'apiurl') . "/submit/". common_config('enjit','apikey');
-        $data = array(
-            'msg' => $atom,
-        );
-
-        #
-        # POST the message to $config['enjit']['apiurl']
-        #
-        $request = HTTPClient::start();
-        $response = $request->post($url, null, $data);
-
-        return $response->isOk();
-    }
-
-}
-
-if (have_option('-i')) {
-    $id = get_option_value('-i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new EnjitQueueHandler($id);
-
-if ($handler->start()) {
-    $handler->handle_queue();
-}
-
-$handler->finish();
diff --git a/scripts/getvaliddaemons.php b/scripts/getvaliddaemons.php
index 99ad41b37..a332e06b5 100755
--- a/scripts/getvaliddaemons.php
+++ b/scripts/getvaliddaemons.php
@@ -37,19 +37,10 @@ require_once INSTALLDIR.'/scripts/commandline.inc';
 
 $daemons = array();
 
-$daemons[] = INSTALLDIR.'/scripts/pluginqueuehandler.php';
-$daemons[] = INSTALLDIR.'/scripts/ombqueuehandler.php';
-$daemons[] = INSTALLDIR.'/scripts/pingqueuehandler.php';
+$daemons[] = INSTALLDIR.'/scripts/queuedaemon.php';
 
 if(common_config('xmpp','enabled')) {
     $daemons[] = INSTALLDIR.'/scripts/xmppdaemon.php';
-    $daemons[] = INSTALLDIR.'/scripts/jabberqueuehandler.php';
-    $daemons[] = INSTALLDIR.'/scripts/publicqueuehandler.php';
-    $daemons[] = INSTALLDIR.'/scripts/xmppconfirmhandler.php';
-}
-
-if (common_config('sms', 'enabled')) {
-    $daemons[] = INSTALLDIR.'/scripts/smsqueuehandler.php';
 }
 
 if (Event::handle('GetValidDaemons', array(&$daemons))) {
diff --git a/scripts/pluginqueuehandler.php b/scripts/handlequeued.php
index fa39bdda6..9031437aa 100755
--- a/scripts/pluginqueuehandler.php
+++ b/scripts/handlequeued.php
@@ -20,45 +20,37 @@
 
 define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
 
-$shortoptions = 'i::';
-$longoptions = array('id::');
+$helptext = <<<END_OF_QUEUE_HELP
+USAGE: handlequeued.php <queue> <notice id>
+Run a single queued notice through background processing
+as if it were being run through the queue.
 
-$helptext = <<<END_OF_OMB_HELP
-Daemon script for letting plugins handle stuff at queue time
 
-    -i --id           Identity (default none)
-
-END_OF_OMB_HELP;
+END_OF_QUEUE_HELP;
 
 require_once INSTALLDIR.'/scripts/commandline.inc';
-require_once INSTALLDIR . '/lib/queuehandler.php';
-
-class PluginQueueHandler extends QueueHandler
-{
 
-    function transport()
-    {
-        return 'plugin';
-    }
+if (count($args) != 2) {
+    show_help();
+}
 
-    function start()
-    {
-        $this->log(LOG_INFO, "INITIALIZE");
-        return true;
-    }
+$queue = trim($args[0]);
+$noticeId = intval($args[1]);
 
-    function handle_notice($notice)
-    {
-        Event::handle('HandleQueuedNotice', array(&$notice));
-        return true;
-    }
+$qm = QueueManager::get();
+$handler = $qm->getHandler($queue);
+if (!$handler) {
+    print "No handler for queue '$queue'.\n";
+    exit(1);
 }
 
-if (have_option('i', 'id')) {
-    $id = get_option_value('i', 'id');
-} else {
-    $id = null;
+$notice = Notice::staticGet('id', $noticeId);
+if (empty($notice)) {
+    print "Invalid notice id $noticeId\n";
+    exit(1);
 }
 
-$handler = new PluginQueueHandler($id);
-$handler->runOnce();
+if (!$handler->handle_notice($notice)) {
+    print "Failed to handle notice id $noticeId on queue '$queue'.\n";
+    exit(1);
+}
diff --git a/scripts/jabberqueuehandler.php b/scripts/jabberqueuehandler.php
deleted file mode 100755
index 8f3a56944..000000000
--- a/scripts/jabberqueuehandler.php
+++ /dev/null
@@ -1,78 +0,0 @@
-#!/usr/bin/env php
-<?php
-/*
- * StatusNet - the distributed open-source microblogging tool
- * Copyright (C) 2008, 2009, StatusNet, Inc.
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as published by
- * the Free Software Foundation, either version 3 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 Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_JABBER_HELP
-Daemon script for pushing new notices to Jabber users.
-
-    -i --id           Identity (default none)
-
-END_OF_JABBER_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-
-require_once INSTALLDIR . '/lib/common.php';
-require_once INSTALLDIR . '/lib/jabber.php';
-require_once INSTALLDIR . '/lib/xmppqueuehandler.php';
-
-class JabberQueueHandler extends XmppQueueHandler
-{
-    var $conn = null;
-
-    function transport()
-    {
-        return 'jabber';
-    }
-
-    function handle_notice($notice)
-    {
-        try {
-            return jabber_broadcast_notice($notice);
-        } catch (XMPPHP_Exception $e) {
-            $this->log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
-            exit(1);
-        }
-    }
-}
-
-// Abort immediately if xmpp is not enabled, otherwise the daemon chews up
-// lots of CPU trying to connect to unconfigured servers
-if (common_config('xmpp','enabled')==false) {
-    print "Aborting daemon - xmpp is disabled\n";
-    exit();
-}
-
-if (have_option('i')) {
-    $id = get_option_value('i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new JabberQueueHandler($id);
-
-$handler->runOnce();
diff --git a/scripts/ombqueuehandler.php b/scripts/ombqueuehandler.php
deleted file mode 100755
index be33b9821..000000000
--- a/scripts/ombqueuehandler.php
+++ /dev/null
@@ -1,87 +0,0 @@
-#!/usr/bin/env php
-<?php
-/*
- * StatusNet - the distributed open-source microblogging tool
- * Copyright (C) 2008, 2009, StatusNet, Inc.
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as published by
- * the Free Software Foundation, either version 3 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 Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_OMB_HELP
-Daemon script for pushing new notices to OpenMicroBlogging subscribers.
-
-    -i --id           Identity (default none)
-
-END_OF_OMB_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-
-require_once INSTALLDIR . '/lib/omb.php';
-require_once INSTALLDIR . '/lib/queuehandler.php';
-
-set_error_handler('common_error_handler');
-
-class OmbQueueHandler extends QueueHandler
-{
-
-    function transport()
-    {
-        return 'omb';
-    }
-
-    function start()
-    {
-        $this->log(LOG_INFO, "INITIALIZE");
-        return true;
-    }
-
-    function handle_notice($notice)
-    {
-        if ($this->is_remote($notice)) {
-            $this->log(LOG_DEBUG, 'Ignoring remote notice ' . $notice->id);
-            return true;
-        } else {
-            return omb_broadcast_notice($notice);
-        }
-    }
-
-    function finish()
-    {
-    }
-
-    function is_remote($notice)
-    {
-        $user = User::staticGet($notice->profile_id);
-        return is_null($user);
-    }
-}
-
-if (have_option('i')) {
-    $id = get_option_value('i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new OmbQueueHandler($id);
-
-$handler->runOnce();
diff --git a/scripts/pingqueuehandler.php b/scripts/pingqueuehandler.php
deleted file mode 100755
index c92337e36..000000000
--- a/scripts/pingqueuehandler.php
+++ /dev/null
@@ -1,69 +0,0 @@
-#!/usr/bin/env php
-<?php
-/*
- * StatusNet - the distributed open-source microblogging tool
- * Copyright (C) 2008, 2009, StatusNet, Inc.
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as published by
- * the Free Software Foundation, either version 3 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 Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_PING_HELP
-Daemon script for pushing new notices to ping servers.
-
-    -i --id           Identity (default none)
-
-END_OF_PING_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-
-require_once INSTALLDIR . '/lib/ping.php';
-require_once INSTALLDIR . '/lib/queuehandler.php';
-
-class PingQueueHandler extends QueueHandler {
-
-	function transport() {
-		return 'ping';
-	}
-
-	function start() {
-		$this->log(LOG_INFO, "INITIALIZE");
-		return true;
-	}
-
-	function handle_notice($notice) {
-		return ping_broadcast_notice($notice);
-	}
-
-	function finish() {
-	}
-}
-
-if (have_option('i')) {
-    $id = get_option_value('i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new PingQueueHandler($id);
-
-$handler->runOnce();
diff --git a/scripts/publicqueuehandler.php b/scripts/publicqueuehandler.php
deleted file mode 100755
index 50a11bcba..000000000
--- a/scripts/publicqueuehandler.php
+++ /dev/null
@@ -1,76 +0,0 @@
-#!/usr/bin/env php
-<?php
-/*
- * StatusNet - the distributed open-source microblogging tool
- * Copyright (C) 2008, 2009, StatusNet, Inc.
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as published by
- * the Free Software Foundation, either version 3 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 Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_PUBLIC_HELP
-Daemon script for pushing new notices to public XMPP subscribers.
-
-    -i --id           Identity (default none)
-
-END_OF_PUBLIC_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-
-require_once INSTALLDIR . '/lib/jabber.php';
-require_once INSTALLDIR . '/lib/xmppqueuehandler.php';
-
-class PublicQueueHandler extends XmppQueueHandler
-{
-
-    function transport()
-    {
-        return 'public';
-    }
-
-    function handle_notice($notice)
-    {
-        try {
-            return jabber_public_notice($notice);
-        } catch (XMPPHP_Exception $e) {
-            $this->log(LOG_ERR, "Got an XMPPHP_Exception: " . $e->getMessage());
-            die($e->getMessage());
-        }
-    }
-}
-
-// Abort immediately if xmpp is not enabled, otherwise the daemon chews up
-// lots of CPU trying to connect to unconfigured servers
-if (common_config('xmpp','enabled')==false) {
-    print "Aborting daemon - xmpp is disabled\n";
-    exit();
-}
-
-if (have_option('i')) {
-    $id = get_option_value('i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new PublicQueueHandler($id);
-
-$handler->runOnce();
diff --git a/scripts/queuedaemon.php b/scripts/queuedaemon.php
new file mode 100755
index 000000000..8ef364fe7
--- /dev/null
+++ b/scripts/queuedaemon.php
@@ -0,0 +1,265 @@
+#!/usr/bin/env php
+<?php
+/*
+ * StatusNet - the distributed open-source microblogging tool
+ * Copyright (C) 2008, 2009, StatusNet, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 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 Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
+
+$shortoptions = 'fi:at:';
+$longoptions = array('id=', 'foreground', 'all', 'threads=');
+
+/**
+ * Attempts to get a count of the processors available on the current system
+ * to fan out multiple threads.
+ *
+ * Recognizes Linux and Mac OS X; others will return default of 1.
+ *
+ * @return intval
+ */
+function getProcessorCount()
+{
+    $cpus = 0;
+    switch (PHP_OS) {
+    case 'Linux':
+        $cpuinfo = file('/proc/cpuinfo');
+        foreach (file('/proc/cpuinfo') as $line) {
+            if (preg_match('/^processor\s+:\s+(\d+)\s?$/', $line)) {
+                $cpus++;
+            }
+        }
+        break;
+    case 'Darwin':
+        $cpus = intval(shell_exec("/usr/sbin/sysctl -n hw.ncpu 2>/dev/null"));
+        break;
+    }
+    if ($cpus) {
+        return $cpus;
+    }
+    return 1;
+}
+
+$threads = getProcessorCount();
+$helptext = <<<END_OF_QUEUE_HELP
+Daemon script for running queued items.
+
+    -i --id           Identity (default none)
+    -f --foreground   Stay in the foreground (default background)
+    -a --all          Handle queues for all local sites
+                      (requires Stomp queue handler, status_network setup)
+    -t --threads=<n>  Spawn <n> processing threads (default $threads)
+
+
+END_OF_QUEUE_HELP;
+
+require_once INSTALLDIR.'/scripts/commandline.inc';
+
+require_once(INSTALLDIR.'/lib/daemon.php');
+require_once(INSTALLDIR.'/classes/Queue_item.php');
+require_once(INSTALLDIR.'/classes/Notice.php');
+
+define('CLAIM_TIMEOUT', 1200);
+
+/**
+ * Queue handling daemon...
+ *
+ * The queue daemon by default launches in the background, at which point
+ * it'll pass control to the configured QueueManager class to poll for updates.
+ *
+ * We can then pass individual items through the QueueHandler subclasses
+ * they belong to.
+ */
+class QueueDaemon extends Daemon
+{
+    protected $allsites;
+    protected $threads=1;
+
+    function __construct($id=null, $daemonize=true, $threads=1, $allsites=false)
+    {
+        parent::__construct($daemonize);
+
+        if ($id) {
+            $this->set_id($id);
+        }
+        $this->all = $allsites;
+        $this->threads = $threads;
+    }
+
+    /**
+     * How many seconds a polling-based queue manager should wait between
+     * checks for new items to handle.
+     *
+     * Defaults to 60 seconds; override to speed up or slow down.
+     *
+     * @return int timeout in seconds
+     */
+    function timeout()
+    {
+        return 60;
+    }
+
+    function name()
+    {
+        return strtolower(get_class($this).'.'.$this->get_id());
+    }
+
+    function run()
+    {
+        if ($this->threads > 1) {
+            return $this->runThreads();
+        } else {
+            return $this->runLoop();
+        }
+    }
+    
+    function runThreads()
+    {
+        $children = array();
+        for ($i = 1; $i <= $this->threads; $i++) {
+            $pid = pcntl_fork();
+            if ($pid < 0) {
+                print "Couldn't fork for thread $i; aborting\n";
+                exit(1);
+            } else if ($pid == 0) {
+                $this->runChild($i);
+                exit(0);
+            } else {
+                $this->log(LOG_INFO, "Spawned thread $i as pid $pid");
+                $children[$i] = $pid;
+            }
+        }
+        
+        $this->log(LOG_INFO, "Waiting for children to complete.");
+        while (count($children) > 0) {
+            $status = null;
+            $pid = pcntl_wait($status);
+            if ($pid > 0) {
+                $i = array_search($pid, $children);
+                if ($i === false) {
+                    $this->log(LOG_ERR, "Unrecognized child pid $pid exited!");
+                    continue;
+                }
+                unset($children[$i]);
+                $this->log(LOG_INFO, "Thread $i pid $pid exited.");
+                
+                $pid = pcntl_fork();
+                if ($pid < 0) {
+                    print "Couldn't fork to respawn thread $i; aborting thread.\n";
+                } else if ($pid == 0) {
+                    $this->runChild($i);
+                    exit(0);
+                } else {
+                    $this->log(LOG_INFO, "Respawned thread $i as pid $pid");
+                    $children[$i] = $pid;
+                }
+            }
+        }
+        $this->log(LOG_INFO, "All child processes complete.");
+        return true;
+    }
+
+    function runChild($thread)
+    {
+        $this->set_id($this->get_id() . "." . $thread);
+        $this->resetDb();
+        $this->runLoop();
+    }
+
+    /**
+     * Reconnect to the database for each child process,
+     * or they'll get very confused trying to use the
+     * same socket.
+     */
+    function resetDb()
+    {
+        // @fixme do we need to explicitly open the db too
+        // or is this implied?
+        global $_DB_DATAOBJECT;
+        unset($_DB_DATAOBJECT['CONNECTIONS']);
+
+        // Reconnect main memcached, or threads will stomp on
+        // each other and corrupt their requests.
+        $cache = common_memcache();
+        if ($cache) {
+            $cache->reconnect();
+        }
+
+        // Also reconnect memcached for status_network table.
+        if (!empty(Status_network::$cache)) {
+            Status_network::$cache->close();
+            Status_network::$cache = null;
+        }
+    }
+
+    /**
+     * Setup and start of run loop for this queue handler as a daemon.
+     * Most of the heavy lifting is passed on to the QueueManager's service()
+     * method, which passes control on to the QueueHandler's handle_notice()
+     * method for each notice that comes in on the queue.
+     *
+     * Most of the time this won't need to be overridden in a subclass.
+     *
+     * @return boolean true on success, false on failure
+     */
+    function runLoop()
+    {
+        $this->log(LOG_INFO, 'checking for queued notices');
+
+        $master = new IoMaster($this->get_id());
+        $master->init($this->all);
+        $master->service();
+
+        $this->log(LOG_INFO, 'finished servicing the queue');
+
+        $this->log(LOG_INFO, 'terminating normally');
+
+        return true;
+    }
+
+    function log($level, $msg)
+    {
+        common_log($level, get_class($this) . ' ('. $this->get_id() .'): '.$msg);
+    }
+}
+
+if (have_option('i')) {
+    $id = get_option_value('i');
+} else if (have_option('--id')) {
+    $id = get_option_value('--id');
+} else if (count($args) > 0) {
+    $id = $args[0];
+} else {
+    $id = null;
+}
+
+if (have_option('t')) {
+    $threads = intval(get_option_value('t'));
+} else if (have_option('--threads')) {
+    $threads = intval(get_option_value('--threads'));
+} else {
+    $threads = 0;
+}
+if (!$threads) {
+    $threads = getProcessorCount();
+}
+
+$daemonize = !(have_option('f') || have_option('--foreground'));
+$all = have_option('a') || have_option('--all');
+
+$daemon = new QueueDaemon($id, $daemonize, $threads, $all);
+$daemon->runOnce();
+
diff --git a/scripts/smsqueuehandler.php b/scripts/smsqueuehandler.php
deleted file mode 100755
index 6583a77da..000000000
--- a/scripts/smsqueuehandler.php
+++ /dev/null
@@ -1,73 +0,0 @@
-#!/usr/bin/env php
-<?php
-/*
- * StatusNet - the distributed open-source microblogging tool
- * Copyright (C) 2008, 2009, StatusNet, Inc.
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as published by
- * the Free Software Foundation, either version 3 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 Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_SMS_HELP
-Daemon script for pushing new notices to local subscribers using SMS.
-
-    -i --id           Identity (default none)
-
-END_OF_SMS_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-
-require_once INSTALLDIR . '/lib/mail.php';
-require_once INSTALLDIR . '/lib/queuehandler.php';
-
-class SmsQueueHandler extends QueueHandler
-{
-    function transport()
-    {
-        return 'sms';
-    }
-
-    function start()
-    {
-        $this->log(LOG_INFO, "INITIALIZE");
-        return true;
-    }
-
-    function handle_notice($notice)
-    {
-        return mail_broadcast_notice_sms($notice);
-    }
-
-    function finish()
-    {
-    }
-}
-
-if (have_option('i')) {
-    $id = get_option_value('i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new SmsQueueHandler($id);
-
-$handler->runOnce();
diff --git a/scripts/xmppconfirmhandler.php b/scripts/xmppconfirmhandler.php
deleted file mode 100755
index 2e3974136..000000000
--- a/scripts/xmppconfirmhandler.php
+++ /dev/null
@@ -1,161 +0,0 @@
-#!/usr/bin/env php
-<?php
-/*
- * StatusNet - the distributed open-source microblogging tool
- * Copyright (C) 2008, 2009, StatusNet, Inc.
- *
- * This program is free software: you can redistribute it and/or modify
- * it under the terms of the GNU Affero General Public License as published by
- * the Free Software Foundation, either version 3 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 Affero General Public License for more details.
- *
- * You should have received a copy of the GNU Affero General Public License
- * along with this program.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
-
-$shortoptions = 'i::';
-$longoptions = array('id::');
-
-$helptext = <<<END_OF_JABBER_HELP
-Daemon script for pushing new confirmations to Jabber users.
-
-    -i --id           Identity (default none)
-
-END_OF_JABBER_HELP;
-
-require_once INSTALLDIR.'/scripts/commandline.inc';
-require_once INSTALLDIR . '/lib/jabber.php';
-require_once INSTALLDIR . '/lib/xmppqueuehandler.php';
-
-class XmppConfirmHandler extends XmppQueueHandler
-{
-    var $_id = 'confirm';
-
-    function class_name()
-    {
-        return 'XmppConfirmHandler';
-    }
-
-    function run()
-    {
-        if (!$this->start()) {
-            return false;
-        }
-        $this->log(LOG_INFO, 'checking for queued confirmations');
-        do {
-            $confirm = $this->next_confirm();
-            if ($confirm) {
-                $this->log(LOG_INFO, 'Sending confirmation for ' . $confirm->address);
-                $user = User::staticGet($confirm->user_id);
-                if (!$user) {
-                    $this->log(LOG_WARNING, 'Confirmation for unknown user ' . $confirm->user_id);
-                    continue;
-                }
-                $success = jabber_confirm_address($confirm->code,
-                                                  $user->nickname,
-                                                  $confirm->address);
-                if (!$success) {
-                    $this->log(LOG_ERR, 'Confirmation failed for ' . $confirm->address);
-                    # Just let the claim age out; hopefully things work then
-                    continue;
-                } else {
-                    $this->log(LOG_INFO, 'Confirmation sent for ' . $confirm->address);
-                    # Mark confirmation sent; need a dupe so we don't have the WHERE clause
-                    $dupe = Confirm_address::staticGet('code', $confirm->code);
-                    if (!$dupe) {
-                        common_log(LOG_WARNING, 'Could not refetch confirm', __FILE__);
-                        continue;
-                    }
-                    $orig = clone($dupe);
-                    $dupe->sent = $dupe->claimed;
-                    $result = $dupe->update($orig);
-                    if (!$result) {
-                        common_log_db_error($dupe, 'UPDATE', __FILE__);
-                        # Just let the claim age out; hopefully things work then
-                        continue;
-                    }
-                    $dupe->free();
-                    unset($dupe);
-                }
-                $user->free();
-                unset($user);
-                $confirm->free();
-                unset($confirm);
-                $this->idle(0);
-            } else {
-#                $this->clear_old_confirm_claims();
-                $this->idle(10);
-            }
-        } while (true);
-        if (!$this->finish()) {
-            return false;
-        }
-        return true;
-    }
-
-    function next_confirm()
-    {
-        $confirm = new Confirm_address();
-        $confirm->whereAdd('claimed IS null');
-        $confirm->whereAdd('sent IS null');
-        # XXX: eventually we could do other confirmations in the queue, too
-        $confirm->address_type = 'jabber';
-        $confirm->orderBy('modified DESC');
-        $confirm->limit(1);
-        if ($confirm->find(true)) {
-            $this->log(LOG_INFO, 'Claiming confirmation for ' . $confirm->address);
-                # working around some weird DB_DataObject behaviour
-            $confirm->whereAdd(''); # clears where stuff
-            $original = clone($confirm);
-            $confirm->claimed = common_sql_now();
-            $result = $confirm->update($original);
-            if ($result) {
-                $this->log(LOG_INFO, 'Succeeded in claim! '. $result);
-                return $confirm;
-            } else {
-                $this->log(LOG_INFO, 'Failed in claim!');
-                return false;
-            }
-        }
-        return null;
-    }
-
-    function clear_old_confirm_claims()
-    {
-        $confirm = new Confirm();
-        $confirm->claimed = null;
-        $confirm->whereAdd('now() - claimed > '.CLAIM_TIMEOUT);
-        $confirm->update(DB_DATAOBJECT_WHEREADD_ONLY);
-        $confirm->free();
-        unset($confirm);
-    }
-}
-
-// Abort immediately if xmpp is not enabled, otherwise the daemon chews up
-// lots of CPU trying to connect to unconfigured servers
-if (common_config('xmpp','enabled')==false) {
-    print "Aborting daemon - xmpp is disabled\n";
-    exit();
-}
-
-if (have_option('i')) {
-    $id = get_option_value('i');
-} else if (have_option('--id')) {
-    $id = get_option_value('--id');
-} else if (count($args) > 0) {
-    $id = $args[0];
-} else {
-    $id = null;
-}
-
-$handler = new XmppConfirmHandler($id);
-
-$handler->runOnce();
-