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     function __construct($id=null, $daemonize=true, $threads=1)
  40     {
  41         parent::__construct($daemonize);
  42
  43         if ($id) {
  44             $this->set_id($id);
  45         }
  46         $this->threads = $threads;
  47     }
  48
  49     /**
  50      * Perform some actual work!
  51      *
  52      * @return boolean true on success, false on failure
  53      */
  54     public abstract function runThread();
  55
  56     /**
  57      * Spawn one or more background processes and let them start running.
  58      * Each individual process will execute whatever's in the runThread()
  59      * method, which should be overridden.
  60      *
  61      * Child processes will be automatically respawned when they exit.
  62      *
  63      * @todo possibly allow for not respawning on "normal" exits...
  64      *       though ParallelizingDaemon is probably better for workloads
  65      *       that have forseeable endpoints.
  66      */
  67     function run()
  68     {
  69         $children = array();
  70         for ($i = 1; $i <= $this->threads; $i++) {
  71             $pid = pcntl_fork();
  72             if ($pid < 0) {
  73                 $this->log(LOG_ERROR, "Couldn't fork for thread $i; aborting\n");
  74                 exit(1);
  75             } else if ($pid == 0) {
  76                 $this->initAndRunChild($i);
  77             } else {
  78                 $this->log(LOG_INFO, "Spawned thread $i as pid $pid");
  79                 $children[$i] = $pid;
  80             }
  81         }
  82
  83         $this->log(LOG_INFO, "Waiting for children to complete.");
  84         while (count($children) > 0) {
  85             $status = null;
  86             $pid = pcntl_wait($status);
  87             if ($pid > 0) {
  88                 $i = array_search($pid, $children);
  89                 if ($i === false) {
  90                     $this->log(LOG_ERR, "Unrecognized child pid $pid exited!");
  91                     continue;
  92                 }
  93                 unset($children[$i]);
  94                 $this->log(LOG_INFO, "Thread $i pid $pid exited.");
  95
  96                 $pid = pcntl_fork();
  97                 if ($pid < 0) {
  98                     $this->log(LOG_ERROR, "Couldn't fork to respawn thread $i; aborting thread.\n");
  99                 } else if ($pid == 0) {
 100                     $this->initAndRunChild($i);
 101                 } else {
 102                     $this->log(LOG_INFO, "Respawned thread $i as pid $pid");
 103                     $children[$i] = $pid;
 104                 }
 105             }
 106         }
 107         $this->log(LOG_INFO, "All child processes complete.");
 108         return true;
 109     }
 110
 111     /**
 112      * Initialize things for a fresh thread, call runThread(), and
 113      * exit at completion with appropriate return value.
 114      */
 115     protected function initAndRunChild($thread)
 116     {
 117         $this->set_id($this->get_id() . "." . $thread);
 118         $this->resetDb();
 119         $ok = $this->runThread();
 120         exit($ok ? 0 : 1);
 121     }
 122
 123     /**
 124      * Reconnect to the database for each child process,
 125      * or they'll get very confused trying to use the
 126      * same socket.
 127      */
 128     protected function resetDb()
 129     {
 130         // @fixme do we need to explicitly open the db too
 131         // or is this implied?
 132         global $_DB_DATAOBJECT;
 133         unset($_DB_DATAOBJECT['CONNECTIONS']);
 134
 135         // Reconnect main memcached, or threads will stomp on
 136         // each other and corrupt their requests.
 137         $cache = common_memcache();
 138         if ($cache) {
 139             $cache->reconnect();
 140         }
 141
 142         // Also reconnect memcached for status_network table.
 143         if (!empty(Status_network::$cache)) {
 144             Status_network::$cache->close();
 145             Status_network::$cache = null;
 146         }
 147     }
 148
 149     function log($level, $msg)
 150     {
 151         common_log($level, get_class($this) . ' ('. $this->get_id() .'): '.$msg);
 152     }
 153
 154     function name()
 155     {
 156         return strtolower(get_class($this).'.'.$this->get_id());
 157     }
 158 }
 159