lib/spawningdaemon.php

   1 <?php
   2 /*
   3  * StatusNet - the distributed open-source microblogging tool
   4  * Copyright (C) 2010, StatusNet, Inc.
   5  *
   6  * This program is free software: you can redistribute it and/or modify
   7  * it under the terms of the GNU Affero General Public License as published by
   8  * the Free Software Foundation, either version 3 of the License, or
   9  * (at your option) any later version.
  10  *
  11  * This program is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14  * GNU Affero General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Affero General Public License
  17  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  18  */
  19
  20 /**
  21  * Base class for daemon that can launch one or more processing threads,
  22  * respawning them if they exit.
  23  *
  24  * This is mainly intended for indefinite workloads such as monitoring
  25  * a queue or maintaining an IM channel.
  26  *
  27  * Child classes should implement the
  28  *
  29  * We can then pass individual items through the QueueHandler subclasses
  30  * they belong to. We additionally can handle queues for multiple sites.
  31  *
  32  * @package QueueHandler
  33  * @author Brion Vibber <brion@status.net>
  34  */
  35 abstract class SpawningDaemon extends Daemon
  36 {
  37     protected $threads=1;
  38
  39     const EXIT_OK = 0;
  40     const EXIT_ERR = 1;
  41     const EXIT_SHUTDOWN = 100;
  42     const EXIT_RESTART = 101;
  43
  44     function __construct($id=null, $daemonize=true, $threads=1)
  45     {
  46         parent::__construct($daemonize);
  47
  48         if ($id) {
  49             $this->set_id($id);
  50         }
  51         $this->threads = $threads;
  52     }
  53
  54     /**
  55      * Perform some actual work!
  56      *
  57      * @return int exit code; use self::EXIT_SHUTDOWN to request not to respawn.
  58      */
  59     public abstract function runThread();
  60
  61     /**
  62      * Spawn one or more background processes and let them start running.
  63      * Each individual process will execute whatever's in the runThread()
  64      * method, which should be overridden.
  65      *
  66      * Child processes will be automatically respawned when they exit.
  67      *
  68      * @todo possibly allow for not respawning on "normal" exits...
  69      *       though ParallelizingDaemon is probably better for workloads
  70      *       that have forseeable endpoints.
  71      */
  72     function run()
  73     {
  74         $this->initPipes();
  75
  76         $children = array();
  77         for ($i = 1; $i <= $this->threads; $i++) {
  78             $pid = pcntl_fork();
  79             if ($pid < 0) {
  80                 $this->log(LOG_ERROR, "Couldn't fork for thread $i; aborting\n");
  81                 exit(1);
  82             } else if ($pid == 0) {
  83                 $this->initAndRunChild($i);
  84             } else {
  85                 $this->log(LOG_INFO, "Spawned thread $i as pid $pid");
  86                 $children[$i] = $pid;
  87             }
  88             sleep(common_config('queue', 'spawndelay'));
  89         }
  90
  91         $this->log(LOG_INFO, "Waiting for children to complete.");
  92         while (count($children) > 0) {
  93             $status = null;
  94             $pid = pcntl_wait($status);
  95             if ($pid > 0) {
  96                 $i = array_search($pid, $children);
  97                 if ($i === false) {
  98                     $this->log(LOG_ERR, "Ignoring exit of unrecognized child pid $pid");
  99                     continue;
 100                 }
 101                 if (pcntl_wifexited($status)) {
 102                     $exitCode = pcntl_wexitstatus($status);
 103                     $info = "status $exitCode";
 104                 } else if (pcntl_wifsignaled($status)) {
 105                     $exitCode = self::EXIT_ERR;
 106                     $signal = pcntl_wtermsig($status);
 107                     $info = "signal $signal";
 108                 }
 109                 unset($children[$i]);
 110
 111                 if ($this->shouldRespawn($exitCode)) {
 112                     $this->log(LOG_INFO, "Thread $i pid $pid exited with $info; respawing.");
 113
 114                     $pid = pcntl_fork();
 115                     if ($pid < 0) {
 116                         $this->log(LOG_ERROR, "Couldn't fork to respawn thread $i; aborting thread.\n");
 117                     } else if ($pid == 0) {
 118                         $this->initAndRunChild($i);
 119                     } else {
 120                         $this->log(LOG_INFO, "Respawned thread $i as pid $pid");
 121                         $children[$i] = $pid;
 122                     }
 123                     sleep(common_config('queue', 'spawndelay'));
 124                 } else {
 125                     $this->log(LOG_INFO, "Thread $i pid $pid exited with status $exitCode; closing out thread.");
 126                 }
 127             }
 128         }
 129         $this->log(LOG_INFO, "All child processes complete.");
 130         return true;
 131     }
 132
 133     /**
 134      * Create an IPC socket pair which child processes can use to detect
 135      * if the parent process has been killed.
 136      */
 137     function initPipes()
 138     {
 139         $sockets = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
 140         if ($sockets) {
 141             $this->parentWriter = $sockets[0];
 142             $this->parentReader = $sockets[1];
 143         } else {
 144             $this->log(LOG_ERROR, "Couldn't create inter-process sockets");
 145             exit(1);
 146         }
 147     }
 148
 149     /**
 150      * Build an IOManager that simply ensures that we have a connection
 151      * to the parent process open. If it breaks, the child process will
 152      * die.
 153      *
 154      * @return ProcessManager
 155      */
 156     public function processManager()
 157     {
 158         return new ProcessManager($this->parentReader);
 159     }
 160
 161     /**
 162      * Determine whether to respawn an exited subprocess based on its exit code.
 163      * Otherwise we'll respawn all exits by default.
 164      *
 165      * @param int $exitCode
 166      * @return boolean true to respawn
 167      */
 168     protected function shouldRespawn($exitCode)
 169     {
 170         if ($exitCode == self::EXIT_SHUTDOWN) {
 171             // Thread requested a clean shutdown.
 172             return false;
 173         } else {
 174             // Otherwise we should always respawn!
 175             return true;
 176         }
 177     }
 178
 179     /**
 180      * Initialize things for a fresh thread, call runThread(), and
 181      * exit at completion with appropriate return value.
 182      */
 183     protected function initAndRunChild($thread)
 184     {
 185         // Close the writer end of our parent<->children pipe.
 186         fclose($this->parentWriter);
 187         $this->set_id($this->get_id() . "." . $thread);
 188         $this->resetDb();
 189         $exitCode = $this->runThread();
 190         exit($exitCode);
 191     }
 192
 193     /**
 194      * Reconnect to the database for each child process,
 195      * or they'll get very confused trying to use the
 196      * same socket.
 197      */
 198     protected function resetDb()
 199     {
 200         // @fixme do we need to explicitly open the db too
 201         // or is this implied?
 202         global $_DB_DATAOBJECT;
 203         unset($_DB_DATAOBJECT['CONNECTIONS']);
 204
 205         // Reconnect main memcached, or threads will stomp on
 206         // each other and corrupt their requests.
 207         $cache = Cache::instance();
 208         if ($cache) {
 209             $cache->reconnect();
 210         }
 211
 212         // Also reconnect memcached for status_network table.
 213         if (!empty(Status_network::$cache)) {
 214             Status_network::$cache->close();
 215             Status_network::$cache = null;
 216         }
 217     }
 218
 219     function log($level, $msg)
 220     {
 221         common_log($level, get_class($this) . ' ('. $this->get_id() .'): '.$msg);
 222     }
 223
 224     function name()
 225     {
 226         return strtolower(get_class($this).'.'.$this->get_id());
 227     }
 228 }
 229