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