From ec145b73fc91dd54695dd374c8a71a11e233b8c0 Mon Sep 17 00:00:00 2001 From: Brion Vibber Date: Tue, 12 Jan 2010 19:57:15 -0800 Subject: 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 --- scripts/queuedaemon.php | 265 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 265 insertions(+) create mode 100755 scripts/queuedaemon.php (limited to 'scripts/queuedaemon.php') 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 +. + */ + +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 = << Spawn 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(); + -- cgit v1.2.3-54-g00ecf