]> git.mxchange.org Git - quix0rs-gnu-social.git/blob - scripts/queuedaemon.php
Merge commit 'refs/merge-requests/117' of git://gitorious.org/statusnet/mainline...
[quix0rs-gnu-social.git] / scripts / queuedaemon.php
1 #!/usr/bin/env php
2 <?php
3 /*
4  * StatusNet - the distributed open-source microblogging tool
5  * Copyright (C) 2008, 2009, StatusNet, Inc.
6  *
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.
11  *
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.
16  *
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/>.
19  */
20
21 define('INSTALLDIR', realpath(dirname(__FILE__) . '/..'));
22
23 $shortoptions = 'fi:at:';
24 $longoptions = array('id=', 'foreground', 'all', 'threads=', 'skip-xmpp', 'xmpp-only');
25
26 /**
27  * Attempts to get a count of the processors available on the current system
28  * to fan out multiple threads.
29  *
30  * Recognizes Linux and Mac OS X; others will return default of 1.
31  *
32  * @return intval
33  */
34 function getProcessorCount()
35 {
36     $cpus = 0;
37     switch (PHP_OS) {
38     case 'Linux':
39         $cpuinfo = file('/proc/cpuinfo');
40         foreach (file('/proc/cpuinfo') as $line) {
41             if (preg_match('/^processor\s+:\s+(\d+)\s?$/', $line)) {
42                 $cpus++;
43             }
44         }
45         break;
46     case 'Darwin':
47         $cpus = intval(shell_exec("/usr/sbin/sysctl -n hw.ncpu 2>/dev/null"));
48         break;
49     }
50     if ($cpus) {
51         return $cpus;
52     }
53     return 1;
54 }
55
56 $threads = getProcessorCount();
57 $helptext = <<<END_OF_QUEUE_HELP
58 Daemon script for running queued items.
59
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)
65
66
67 END_OF_QUEUE_HELP;
68
69 require_once INSTALLDIR.'/scripts/commandline.inc';
70
71 require_once(INSTALLDIR.'/lib/daemon.php');
72 require_once(INSTALLDIR.'/classes/Queue_item.php');
73 require_once(INSTALLDIR.'/classes/Notice.php');
74
75 define('CLAIM_TIMEOUT', 1200);
76
77 /**
78  * Queue handling daemon...
79  *
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.
82  *
83  * We can then pass individual items through the QueueHandler subclasses
84  * they belong to.
85  */
86 class QueueDaemon extends Daemon
87 {
88     protected $allsites;
89     protected $threads=1;
90
91     function __construct($id=null, $daemonize=true, $threads=1, $allsites=false)
92     {
93         parent::__construct($daemonize);
94
95         if ($id) {
96             $this->set_id($id);
97         }
98         $this->all = $allsites;
99         $this->threads = $threads;
100     }
101
102     /**
103      * How many seconds a polling-based queue manager should wait between
104      * checks for new items to handle.
105      *
106      * Defaults to 60 seconds; override to speed up or slow down.
107      *
108      * @return int timeout in seconds
109      */
110     function timeout()
111     {
112         return 60;
113     }
114
115     function name()
116     {
117         return strtolower(get_class($this).'.'.$this->get_id());
118     }
119
120     function run()
121     {
122         if ($this->threads > 1) {
123             return $this->runThreads();
124         } else {
125             return $this->runLoop();
126         }
127     }
128     
129     function runThreads()
130     {
131         $children = array();
132         for ($i = 1; $i <= $this->threads; $i++) {
133             $pid = pcntl_fork();
134             if ($pid < 0) {
135                 print "Couldn't fork for thread $i; aborting\n";
136                 exit(1);
137             } else if ($pid == 0) {
138                 $this->runChild($i);
139                 exit(0);
140             } else {
141                 $this->log(LOG_INFO, "Spawned thread $i as pid $pid");
142                 $children[$i] = $pid;
143             }
144         }
145         
146         $this->log(LOG_INFO, "Waiting for children to complete.");
147         while (count($children) > 0) {
148             $status = null;
149             $pid = pcntl_wait($status);
150             if ($pid > 0) {
151                 $i = array_search($pid, $children);
152                 if ($i === false) {
153                     $this->log(LOG_ERR, "Unrecognized child pid $pid exited!");
154                     continue;
155                 }
156                 unset($children[$i]);
157                 $this->log(LOG_INFO, "Thread $i pid $pid exited.");
158                 
159                 $pid = pcntl_fork();
160                 if ($pid < 0) {
161                     print "Couldn't fork to respawn thread $i; aborting thread.\n";
162                 } else if ($pid == 0) {
163                     $this->runChild($i);
164                     exit(0);
165                 } else {
166                     $this->log(LOG_INFO, "Respawned thread $i as pid $pid");
167                     $children[$i] = $pid;
168                 }
169             }
170         }
171         $this->log(LOG_INFO, "All child processes complete.");
172         return true;
173     }
174
175     function runChild($thread)
176     {
177         $this->set_id($this->get_id() . "." . $thread);
178         $this->resetDb();
179         $this->runLoop();
180     }
181
182     /**
183      * Reconnect to the database for each child process,
184      * or they'll get very confused trying to use the
185      * same socket.
186      */
187     function resetDb()
188     {
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']);
193
194         // Reconnect main memcached, or threads will stomp on
195         // each other and corrupt their requests.
196         $cache = common_memcache();
197         if ($cache) {
198             $cache->reconnect();
199         }
200
201         // Also reconnect memcached for status_network table.
202         if (!empty(Status_network::$cache)) {
203             Status_network::$cache->close();
204             Status_network::$cache = null;
205         }
206     }
207
208     /**
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.
213      *
214      * Most of the time this won't need to be overridden in a subclass.
215      *
216      * @return boolean true on success, false on failure
217      */
218     function runLoop()
219     {
220         $this->log(LOG_INFO, 'checking for queued notices');
221
222         $master = new IoMaster($this->get_id());
223         $master->init($this->all);
224         $master->service();
225
226         $this->log(LOG_INFO, 'finished servicing the queue');
227
228         $this->log(LOG_INFO, 'terminating normally');
229
230         return true;
231     }
232
233     function log($level, $msg)
234     {
235         common_log($level, get_class($this) . ' ('. $this->get_id() .'): '.$msg);
236     }
237 }
238
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) {
244     $id = $args[0];
245 } else {
246     $id = null;
247 }
248
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'));
253 } else {
254     $threads = 0;
255 }
256 if (!$threads) {
257     $threads = getProcessorCount();
258 }
259
260 $daemonize = !(have_option('f') || have_option('--foreground'));
261 $all = have_option('a') || have_option('--all');
262
263 if (have_option('--skip-xmpp')) {
264     define('XMPP_EMERGENCY_FLAG', true);
265 }
266 if (have_option('--xmpp-only')) {
267     define('XMPP_ONLY_FLAG', true);
268 }
269
270 $daemon = new QueueDaemon($id, $daemonize, $threads, $all);
271 $daemon->runOnce();
272