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(true);
 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         //only include global singletons on the first thread
 180         $this->runLoop($thread == 1);
 181     }
 182
 183     /**
 184      * Reconnect to the database for each child process,
 185      * or they'll get very confused trying to use the
 186      * same socket.
 187      */
 188     function resetDb()
 189     {
 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']);
 194
 195         // Reconnect main memcached, or threads will stomp on
 196         // each other and corrupt their requests.
 197         $cache = common_memcache();
 198         if ($cache) {
 199             $cache->reconnect();
 200         }
 201
 202         // Also reconnect memcached for status_network table.
 203         if (!empty(Status_network::$cache)) {
 204             Status_network::$cache->close();
 205             Status_network::$cache = null;
 206         }
 207     }
 208
 209     /**
 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.
 214      *
 215      * Most of the time this won't need to be overridden in a subclass.
 216      *
 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)
 220      *
 221      * @return boolean true on success, false on failure
 222      */
 223     function runLoop($includeGlobalSingletons)
 224     {
 225         $this->log(LOG_INFO, 'checking for queued notices');
 226
 227         $master = new IoMaster($this->get_id());
 228         $master->init($this->all, $includeGlobalSingletons);
 229         $master->service();
 230
 231         $this->log(LOG_INFO, 'finished servicing the queue');
 232
 233         $this->log(LOG_INFO, 'terminating normally');
 234
 235         return true;
 236     }
 237
 238     function log($level, $msg)
 239     {
 240         common_log($level, get_class($this) . ' ('. $this->get_id() .'): '.$msg);
 241     }
 242 }
 243
 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) {
 249     $id = $args[0];
 250 } else {
 251     $id = null;
 252 }
 253
 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'));
 258 } else {
 259     $threads = 0;
 260 }
 261 if (!$threads) {
 262     $threads = getProcessorCount();
 263 }
 264
 265 $daemonize = !(have_option('f') || have_option('--foreground'));
 266 $all = have_option('a') || have_option('--all');
 267
 268 if (have_option('--skip-xmpp')) {
 269     define('XMPP_EMERGENCY_FLAG', true);
 270 }
 271 if (have_option('--xmpp-only')) {
 272     define('XMPP_ONLY_FLAG', true);
 273 }
 274
 275 $daemon = new QueueDaemon($id, $daemonize, $threads, $all);
 276 $daemon->runOnce();
 277