4 * StatusNet - the distributed open-source microblogging tool
5 * Copyright (C) 2008, 2009, StatusNet, Inc.
7 * This program is free software: you can redistribute it and/or modify
8 * it under the terms of the GNU Affero General Public License as published by
9 * the Free Software Foundation, either version 3 of the License, or
10 * (at your option) any later version.
12 * This program is distributed in the hope that it will be useful,
13 * but WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 * GNU Affero General Public License for more details.
17 * You should have received a copy of the GNU Affero General Public License
18 * along with this program. If not, see <http://www.gnu.org/licenses/>.
21 define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
23 $shortoptions = 'fi:at:';
24 $longoptions = array('id=', 'foreground', 'all', 'threads=', 'skip-xmpp', 'xmpp-only');
27 * Attempts to get a count of the processors available on the current system
28 * to fan out multiple threads.
30 * Recognizes Linux and Mac OS X; others will return default of 1.
34 function getProcessorCount()
39 $cpuinfo = file('/proc/cpuinfo');
40 foreach (file('/proc/cpuinfo') as $line) {
41 if (preg_match('/^processor\s+:\s+(\d+)\s?$/', $line)) {
47 $cpus = intval(shell_exec("/usr/sbin/sysctl -n hw.ncpu 2>/dev/null"));
56 $threads = getProcessorCount();
57 $helptext = <<<END_OF_QUEUE_HELP
58 Daemon script for running queued items.
60 -i --id Identity (default none)
61 -f --foreground Stay in the foreground (default background)
62 -a --all Handle queues for all local sites
63 (requires Stomp queue handler, status_network setup)
64 -t --threads=<n> Spawn <n> processing threads (default $threads)
69 require_once INSTALLDIR.'/scripts/commandline.inc';
71 require_once(INSTALLDIR.'/lib/daemon.php');
72 require_once(INSTALLDIR.'/classes/Queue_item.php');
73 require_once(INSTALLDIR.'/classes/Notice.php');
75 define('CLAIM_TIMEOUT', 1200);
78 * Queue handling daemon...
80 * The queue daemon by default launches in the background, at which point
81 * it'll pass control to the configured QueueManager class to poll for updates.
83 * We can then pass individual items through the QueueHandler subclasses
86 class QueueDaemon extends Daemon
91 function __construct($id=null, $daemonize=true, $threads=1, $allsites=false)
93 parent::__construct($daemonize);
98 $this->all = $allsites;
99 $this->threads = $threads;
103 * How many seconds a polling-based queue manager should wait between
104 * checks for new items to handle.
106 * Defaults to 60 seconds; override to speed up or slow down.
108 * @return int timeout in seconds
117 return strtolower(get_class($this).'.'.$this->get_id());
122 if ($this->threads > 1) {
123 return $this->runThreads();
125 return $this->runLoop();
129 function runThreads()
132 for ($i = 1; $i <= $this->threads; $i++) {
135 print "Couldn't fork for thread $i; aborting\n";
137 } else if ($pid == 0) {
141 $this->log(LOG_INFO, "Spawned thread $i as pid $pid");
142 $children[$i] = $pid;
146 $this->log(LOG_INFO, "Waiting for children to complete.");
147 while (count($children) > 0) {
149 $pid = pcntl_wait($status);
151 $i = array_search($pid, $children);
153 $this->log(LOG_ERR, "Unrecognized child pid $pid exited!");
156 unset($children[$i]);
157 $this->log(LOG_INFO, "Thread $i pid $pid exited.");
161 print "Couldn't fork to respawn thread $i; aborting thread.\n";
162 } else if ($pid == 0) {
166 $this->log(LOG_INFO, "Respawned thread $i as pid $pid");
167 $children[$i] = $pid;
171 $this->log(LOG_INFO, "All child processes complete.");
175 function runChild($thread)
177 $this->set_id($this->get_id() . "." . $thread);
183 * Reconnect to the database for each child process,
184 * or they'll get very confused trying to use the
189 // @fixme do we need to explicitly open the db too
190 // or is this implied?
191 global $_DB_DATAOBJECT;
192 unset($_DB_DATAOBJECT['CONNECTIONS']);
194 // Reconnect main memcached, or threads will stomp on
195 // each other and corrupt their requests.
196 $cache = common_memcache();
201 // Also reconnect memcached for status_network table.
202 if (!empty(Status_network::$cache)) {
203 Status_network::$cache->close();
204 Status_network::$cache = null;
209 * Setup and start of run loop for this queue handler as a daemon.
210 * Most of the heavy lifting is passed on to the QueueManager's service()
211 * method, which passes control on to the QueueHandler's handle_notice()
212 * method for each notice that comes in on the queue.
214 * Most of the time this won't need to be overridden in a subclass.
216 * @return boolean true on success, false on failure
220 $this->log(LOG_INFO, 'checking for queued notices');
222 $master = new IoMaster($this->get_id());
223 $master->init($this->all);
226 $this->log(LOG_INFO, 'finished servicing the queue');
228 $this->log(LOG_INFO, 'terminating normally');
233 function log($level, $msg)
235 common_log($level, get_class($this) . ' ('. $this->get_id() .'): '.$msg);
239 if (have_option('i')) {
240 $id = get_option_value('i');
241 } else if (have_option('--id')) {
242 $id = get_option_value('--id');
243 } else if (count($args) > 0) {
249 if (have_option('t')) {
250 $threads = intval(get_option_value('t'));
251 } else if (have_option('--threads')) {
252 $threads = intval(get_option_value('--threads'));
257 $threads = getProcessorCount();
260 $daemonize = !(have_option('f') || have_option('--foreground'));
261 $all = have_option('a') || have_option('--all');
263 if (have_option('--skip-xmpp')) {
264 define('XMPP_EMERGENCY_FLAG', true);
266 if (have_option('--xmpp-only')) {
267 define('XMPP_ONLY_FLAG', true);
270 $daemon = new QueueDaemon($id, $daemonize, $threads, $all);