3 * StatusNet - the distributed open-source microblogging tool
4 * Copyright (C) 2010, StatusNet, Inc.
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.
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.
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/>.
21 * Base class for daemon that can launch one or more processing threads,
22 * respawning them if they exit.
24 * This is mainly intended for indefinite workloads such as monitoring
25 * a queue or maintaining an IM channel.
27 * Child classes should implement the
29 * We can then pass individual items through the QueueHandler subclasses
30 * they belong to. We additionally can handle queues for multiple sites.
32 * @package QueueHandler
33 * @author Brion Vibber <brion@status.net>
35 abstract class SpawningDaemon extends Daemon
41 const EXIT_SHUTDOWN = 100;
42 const EXIT_RESTART = 101;
44 function __construct($id=null, $daemonize=true, $threads=1)
46 parent::__construct($daemonize);
51 $this->threads = $threads;
55 * Perform some actual work!
57 * @return int exit code; use self::EXIT_SHUTDOWN to request not to respawn.
59 public abstract function runThread();
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.
66 * Child processes will be automatically respawned when they exit.
68 * @todo possibly allow for not respawning on "normal" exits...
69 * though ParallelizingDaemon is probably better for workloads
70 * that have forseeable endpoints.
75 for ($i = 1; $i <= $this->threads; $i++) {
78 $this->log(LOG_ERROR, "Couldn't fork for thread $i; aborting\n");
80 } else if ($pid == 0) {
81 $this->initAndRunChild($i);
83 $this->log(LOG_INFO, "Spawned thread $i as pid $pid");
86 sleep(common_config('queue', 'spawndelay'));
89 $this->log(LOG_INFO, "Waiting for children to complete.");
90 while (count($children) > 0) {
92 $pid = pcntl_wait($status);
94 $i = array_search($pid, $children);
96 $this->log(LOG_ERR, "Ignoring exit of unrecognized child pid $pid");
99 if (pcntl_wifexited($status)) {
100 $exitCode = pcntl_wexitstatus($status);
101 $info = "status $exitCode";
102 } else if (pcntl_wifsignaled($status)) {
103 $exitCode = self::EXIT_ERR;
104 $signal = pcntl_wtermsig($status);
105 $info = "signal $signal";
107 unset($children[$i]);
109 if ($this->shouldRespawn($exitCode)) {
110 $this->log(LOG_INFO, "Thread $i pid $pid exited with $info; respawing.");
114 $this->log(LOG_ERROR, "Couldn't fork to respawn thread $i; aborting thread.\n");
115 } else if ($pid == 0) {
116 $this->initAndRunChild($i);
118 $this->log(LOG_INFO, "Respawned thread $i as pid $pid");
119 $children[$i] = $pid;
121 sleep(common_config('queue', 'spawndelay'));
123 $this->log(LOG_INFO, "Thread $i pid $pid exited with status $exitCode; closing out thread.");
127 $this->log(LOG_INFO, "All child processes complete.");
132 * Determine whether to respawn an exited subprocess based on its exit code.
133 * Otherwise we'll respawn all exits by default.
135 * @param int $exitCode
136 * @return boolean true to respawn
138 protected function shouldRespawn($exitCode)
140 if ($exitCode == self::EXIT_SHUTDOWN) {
141 // Thread requested a clean shutdown.
144 // Otherwise we should always respawn!
150 * Initialize things for a fresh thread, call runThread(), and
151 * exit at completion with appropriate return value.
153 protected function initAndRunChild($thread)
155 $this->set_id($this->get_id() . "." . $thread);
157 $exitCode = $this->runThread();
162 * Reconnect to the database for each child process,
163 * or they'll get very confused trying to use the
166 protected function resetDb()
168 // @fixme do we need to explicitly open the db too
169 // or is this implied?
170 global $_DB_DATAOBJECT;
171 unset($_DB_DATAOBJECT['CONNECTIONS']);
173 // Reconnect main memcached, or threads will stomp on
174 // each other and corrupt their requests.
175 $cache = common_memcache();
180 // Also reconnect memcached for status_network table.
181 if (!empty(Status_network::$cache)) {
182 Status_network::$cache->close();
183 Status_network::$cache = null;
187 function log($level, $msg)
189 common_log($level, get_class($this) . ' ('. $this->get_id() .'): '.$msg);
194 return strtolower(get_class($this).'.'.$this->get_id());