From c74aea589d5a79d7048470d44e457dffc8919ad3 Mon Sep 17 00:00:00 2001 From: Brion Vibber Date: Tue, 16 Feb 2010 09:01:59 -0800 Subject: Stomp queue restructuring for mass scalability: - Multiplexing queues into groups and for multiple sites. - Sharing vs breakout configurable per site and per queue via $config['queue']['breakout'] - Detect how many times a message is redelivered, discard if it's killed too many daemons - count configurable with $config['queue']['max_retries'] - can dump the items to files in $config['queue']['dead_letter_dir'] Queue daemon memory & resource leak fixes: - avoid unnecessary reconnections to memcached server (switch persistent connections back in on second initialization, assuming it's child process) - monkey-patch for leaky .ini loads in DB_DataObject::databaseStructure() - was leaking 200k per active switch - applied leak fixes to Status_network as well, using intermediate base Safe_DataObject for both it and Memcache_DataObject Misc queue fixes: - correct handling of child processes exiting due to signal termination instead of regular exit - shutdown instead of infinite respawn loop if we're already past the soft memory limit at startup - Added --all option for xmppdaemon... still opens one xmpp connection per site that has xmpp active Cache updates: - add Cache::increment() method with native support for memcached atomic increment --- lib/queuemanager.php | 97 ++++++++++++++++++++++++++++++++++++---------------- 1 file changed, 67 insertions(+), 30 deletions(-) (limited to 'lib/queuemanager.php') diff --git a/lib/queuemanager.php b/lib/queuemanager.php index 149617eb5..8f8c8f133 100644 --- a/lib/queuemanager.php +++ b/lib/queuemanager.php @@ -39,9 +39,10 @@ abstract class QueueManager extends IoManager { static $qm = null; - public $master = null; - public $handlers = array(); - public $groups = array(); + protected $master = null; + protected $handlers = array(); + protected $groups = array(); + protected $activeGroups = array(); /** * Factory function to pull the appropriate QueueManager object @@ -215,55 +216,64 @@ abstract class QueueManager extends IoManager if (class_exists($class)) { return new $class(); } else { - common_log(LOG_ERR, "Nonexistent handler class '$class' for queue '$queue'"); + $this->_log(LOG_ERR, "Nonexistent handler class '$class' for queue '$queue'"); } } else { - common_log(LOG_ERR, "Requested handler for unkown queue '$queue'"); + $this->_log(LOG_ERR, "Requested handler for unkown queue '$queue'"); } return null; } /** * Get a list of registered queue transport names to be used - * for this daemon. + * for listening in this daemon. * * @return array of strings */ - function getQueues() + function activeQueues() { - $group = $this->activeGroup(); - return array_keys($this->groups[$group]); + $queues = array(); + foreach ($this->activeGroups as $group) { + if (isset($this->groups[$group])) { + $queues = array_merge($queues, $this->groups[$group]); + } + } + + return array_keys($queues); } /** - * Initialize the list of queue handlers + * Initialize the list of queue handlers for the current site. * * @event StartInitializeQueueManager * @event EndInitializeQueueManager */ function initialize() { - // @fixme we'll want to be able to listen to particular queues... + $this->handlers = array(); + $this->groups = array(); + $this->groupsByTransport = array(); + if (Event::handle('StartInitializeQueueManager', array($this))) { - $this->connect('plugin', 'PluginQueueHandler'); + $this->connect('distrib', 'DistribQueueHandler'); $this->connect('omb', 'OmbQueueHandler'); $this->connect('ping', 'PingQueueHandler'); - $this->connect('distrib', 'DistribQueueHandler'); if (common_config('sms', 'enabled')) { $this->connect('sms', 'SmsQueueHandler'); } // XMPP output handlers... - $this->connect('jabber', 'JabberQueueHandler'); - $this->connect('public', 'PublicQueueHandler'); - // @fixme this should get an actual queue - //$this->connect('confirm', 'XmppConfirmHandler'); + if (common_config('xmpp', 'enabled')) { + // Delivery prep, read by queuedaemon.php: + $this->connect('jabber', 'JabberQueueHandler'); + $this->connect('public', 'PublicQueueHandler'); + + // Raw output, read by xmppdaemon.php: + $this->connect('xmppout', 'XmppOutQueueHandler', 'xmpp'); + } // For compat with old plugins not registering their own handlers. $this->connect('plugin', 'PluginQueueHandler'); - - $this->connect('xmppout', 'XmppOutQueueHandler', 'xmppdaemon'); - } Event::handle('EndInitializeQueueManager', array($this)); } @@ -276,25 +286,41 @@ abstract class QueueManager extends IoManager * @param string $class * @param string $group */ - public function connect($transport, $class, $group='queuedaemon') + public function connect($transport, $class, $group='main') { $this->handlers[$transport] = $class; $this->groups[$group][$transport] = $class; + $this->groupsByTransport[$transport] = $group; } /** - * @return string queue group to use for this request + * Set the active group which will be used for listening. + * @param string $group */ - function activeGroup() + function setActiveGroup($group) { - $group = 'queuedaemon'; - if ($this->master) { - // hack hack - if ($this->master instanceof XmppMaster) { - return 'xmppdaemon'; - } + $this->activeGroups = array($group); + } + + /** + * Set the active group(s) which will be used for listening. + * @param array $groups + */ + function setActiveGroups($groups) + { + $this->activeGroups = $groups; + } + + /** + * @return string queue group for this queue + */ + function queueGroup($queue) + { + if (isset($this->groupsByTransport[$queue])) { + return $this->groupsByTransport[$queue]; + } else { + throw new Exception("Requested group for unregistered transport $queue"); } - return $group; } /** @@ -318,4 +344,15 @@ abstract class QueueManager extends IoManager $monitor->stats($key, $owners); } } + + protected function _log($level, $msg) + { + $class = get_class($this); + if ($this->activeGroups) { + $groups = ' (' . implode(',', $this->activeGroups) . ')'; + } else { + $groups = ''; + } + common_log($level, "$class$groups: $msg"); + } } -- cgit v1.2.3-54-g00ecf