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(true);
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);
179 //only include global singletons on the first thread
180 $this->runLoop($thread == 1);
184 * Reconnect to the database for each child process,
185 * or they'll get very confused trying to use the
190 // @fixme do we need to explicitly open the db too
191 // or is this implied?
192 global $_DB_DATAOBJECT;
193 unset($_DB_DATAOBJECT['CONNECTIONS']);
195 // Reconnect main memcached, or threads will stomp on
196 // each other and corrupt their requests.
197 $cache = common_memcache();
202 // Also reconnect memcached for status_network table.
203 if (!empty(Status_network::$cache)) {
204 Status_network::$cache->close();
205 Status_network::$cache = null;
210 * Setup and start of run loop for this queue handler as a daemon.
211 * Most of the heavy lifting is passed on to the QueueManager's service()
212 * method, which passes control on to the QueueHandler's handle_notice()
213 * method for each notice that comes in on the queue.
215 * Most of the time this won't need to be overridden in a subclass.
217 * @param boolean $includeGlobalSingletons Include IoManagers that are
218 * global singletons (should only be one instance - regardless of how
219 * many processes or sites there are)
221 * @return boolean true on success, false on failure
223 function runLoop($includeGlobalSingletons)
225 $this->log(LOG_INFO, 'checking for queued notices');
227 $master = new IoMaster($this->get_id());
228 $master->init($this->all, $includeGlobalSingletons);
231 $this->log(LOG_INFO, 'finished servicing the queue');
233 $this->log(LOG_INFO, 'terminating normally');
238 function log($level, $msg)
240 common_log($level, get_class($this) . ' ('. $this->get_id() .'): '.$msg);
244 if (have_option('i')) {
245 $id = get_option_value('i');
246 } else if (have_option('--id')) {
247 $id = get_option_value('--id');
248 } else if (count($args) > 0) {
254 if (have_option('t')) {
255 $threads = intval(get_option_value('t'));
256 } else if (have_option('--threads')) {
257 $threads = intval(get_option_value('--threads'));
262 $threads = getProcessorCount();
265 $daemonize = !(have_option('f') || have_option('--foreground'));
266 $all = have_option('a') || have_option('--all');
268 if (have_option('--skip-xmpp')) {
269 define('XMPP_EMERGENCY_FLAG', true);
271 if (have_option('--xmpp-only')) {
272 define('XMPP_ONLY_FLAG', true);
275 $daemon = new QueueDaemon($id, $daemonize, $threads, $all);