common_memcache() => Cache::instance()
[statusnet-biz:statusnet-biz.git] / 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