532e486a93
Keeps stray daemon subprocesses from floating around when we kill the parents via a signal! Accomplished by opening a bidirectional pipe in the parent process; the children close out the writer end and keep the reader in their open sockets list. When the parent dies, the children see that the socket's been closed out and can perform an orderly shutdown.
230 lines
7.3 KiB
PHP
230 lines
7.3 KiB
PHP
<?php
|
|
/*
|
|
* StatusNet - the distributed open-source microblogging tool
|
|
* Copyright (C) 2010, StatusNet, Inc.
|
|
*
|
|
* This program is free software: you can redistribute it and/or modify
|
|
* it under the terms of the GNU Affero General Public License as published by
|
|
* the Free Software Foundation, either version 3 of the License, or
|
|
* (at your option) any later version.
|
|
*
|
|
* This program is distributed in the hope that it will be useful,
|
|
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
* GNU Affero General Public License for more details.
|
|
*
|
|
* You should have received a copy of the GNU Affero General Public License
|
|
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
*/
|
|
|
|
/**
|
|
* Base class for daemon that can launch one or more processing threads,
|
|
* respawning them if they exit.
|
|
*
|
|
* This is mainly intended for indefinite workloads such as monitoring
|
|
* a queue or maintaining an IM channel.
|
|
*
|
|
* Child classes should implement the
|
|
*
|
|
* We can then pass individual items through the QueueHandler subclasses
|
|
* they belong to. We additionally can handle queues for multiple sites.
|
|
*
|
|
* @package QueueHandler
|
|
* @author Brion Vibber <brion@status.net>
|
|
*/
|
|
abstract class SpawningDaemon extends Daemon
|
|
{
|
|
protected $threads=1;
|
|
|
|
const EXIT_OK = 0;
|
|
const EXIT_ERR = 1;
|
|
const EXIT_SHUTDOWN = 100;
|
|
const EXIT_RESTART = 101;
|
|
|
|
function __construct($id=null, $daemonize=true, $threads=1)
|
|
{
|
|
parent::__construct($daemonize);
|
|
|
|
if ($id) {
|
|
$this->set_id($id);
|
|
}
|
|
$this->threads = $threads;
|
|
}
|
|
|
|
/**
|
|
* Perform some actual work!
|
|
*
|
|
* @return int exit code; use self::EXIT_SHUTDOWN to request not to respawn.
|
|
*/
|
|
public abstract function runThread();
|
|
|
|
/**
|
|
* Spawn one or more background processes and let them start running.
|
|
* Each individual process will execute whatever's in the runThread()
|
|
* method, which should be overridden.
|
|
*
|
|
* Child processes will be automatically respawned when they exit.
|
|
*
|
|
* @todo possibly allow for not respawning on "normal" exits...
|
|
* though ParallelizingDaemon is probably better for workloads
|
|
* that have forseeable endpoints.
|
|
*/
|
|
function run()
|
|
{
|
|
$this->initPipes();
|
|
|
|
$children = array();
|
|
for ($i = 1; $i <= $this->threads; $i++) {
|
|
$pid = pcntl_fork();
|
|
if ($pid < 0) {
|
|
$this->log(LOG_ERROR, "Couldn't fork for thread $i; aborting\n");
|
|
exit(1);
|
|
} else if ($pid == 0) {
|
|
$this->initAndRunChild($i);
|
|
} else {
|
|
$this->log(LOG_INFO, "Spawned thread $i as pid $pid");
|
|
$children[$i] = $pid;
|
|
}
|
|
sleep(common_config('queue', 'spawndelay'));
|
|
}
|
|
|
|
$this->log(LOG_INFO, "Waiting for children to complete.");
|
|
while (count($children) > 0) {
|
|
$status = null;
|
|
$pid = pcntl_wait($status);
|
|
if ($pid > 0) {
|
|
$i = array_search($pid, $children);
|
|
if ($i === false) {
|
|
$this->log(LOG_ERR, "Ignoring exit of unrecognized child pid $pid");
|
|
continue;
|
|
}
|
|
if (pcntl_wifexited($status)) {
|
|
$exitCode = pcntl_wexitstatus($status);
|
|
$info = "status $exitCode";
|
|
} else if (pcntl_wifsignaled($status)) {
|
|
$exitCode = self::EXIT_ERR;
|
|
$signal = pcntl_wtermsig($status);
|
|
$info = "signal $signal";
|
|
}
|
|
unset($children[$i]);
|
|
|
|
if ($this->shouldRespawn($exitCode)) {
|
|
$this->log(LOG_INFO, "Thread $i pid $pid exited with $info; respawing.");
|
|
|
|
$pid = pcntl_fork();
|
|
if ($pid < 0) {
|
|
$this->log(LOG_ERROR, "Couldn't fork to respawn thread $i; aborting thread.\n");
|
|
} else if ($pid == 0) {
|
|
$this->initAndRunChild($i);
|
|
} else {
|
|
$this->log(LOG_INFO, "Respawned thread $i as pid $pid");
|
|
$children[$i] = $pid;
|
|
}
|
|
sleep(common_config('queue', 'spawndelay'));
|
|
} else {
|
|
$this->log(LOG_INFO, "Thread $i pid $pid exited with status $exitCode; closing out thread.");
|
|
}
|
|
}
|
|
}
|
|
$this->log(LOG_INFO, "All child processes complete.");
|
|
return true;
|
|
}
|
|
|
|
/**
|
|
* Create an IPC socket pair which child processes can use to detect
|
|
* if the parent process has been killed.
|
|
*/
|
|
function initPipes()
|
|
{
|
|
$sockets = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, 0);
|
|
if ($sockets) {
|
|
$this->parentWriter = $sockets[0];
|
|
$this->parentReader = $sockets[1];
|
|
} else {
|
|
$this->log(LOG_ERROR, "Couldn't create inter-process sockets");
|
|
exit(1);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Build an IOManager that simply ensures that we have a connection
|
|
* to the parent process open. If it breaks, the child process will
|
|
* die.
|
|
*
|
|
* @return ProcessManager
|
|
*/
|
|
public function processManager()
|
|
{
|
|
return new ProcessManager($this->parentReader);
|
|
}
|
|
|
|
/**
|
|
* Determine whether to respawn an exited subprocess based on its exit code.
|
|
* Otherwise we'll respawn all exits by default.
|
|
*
|
|
* @param int $exitCode
|
|
* @return boolean true to respawn
|
|
*/
|
|
protected function shouldRespawn($exitCode)
|
|
{
|
|
if ($exitCode == self::EXIT_SHUTDOWN) {
|
|
// Thread requested a clean shutdown.
|
|
return false;
|
|
} else {
|
|
// Otherwise we should always respawn!
|
|
return true;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Initialize things for a fresh thread, call runThread(), and
|
|
* exit at completion with appropriate return value.
|
|
*/
|
|
protected function initAndRunChild($thread)
|
|
{
|
|
// Close the writer end of our parent<->children pipe.
|
|
fclose($this->parentWriter);
|
|
$this->set_id($this->get_id() . "." . $thread);
|
|
$this->resetDb();
|
|
$exitCode = $this->runThread();
|
|
exit($exitCode);
|
|
}
|
|
|
|
/**
|
|
* Reconnect to the database for each child process,
|
|
* or they'll get very confused trying to use the
|
|
* same socket.
|
|
*/
|
|
protected function resetDb()
|
|
{
|
|
// @fixme do we need to explicitly open the db too
|
|
// or is this implied?
|
|
global $_DB_DATAOBJECT;
|
|
unset($_DB_DATAOBJECT['CONNECTIONS']);
|
|
|
|
// Reconnect main memcached, or threads will stomp on
|
|
// each other and corrupt their requests.
|
|
$cache = common_memcache();
|
|
if ($cache) {
|
|
$cache->reconnect();
|
|
}
|
|
|
|
// Also reconnect memcached for status_network table.
|
|
if (!empty(Status_network::$cache)) {
|
|
Status_network::$cache->close();
|
|
Status_network::$cache = null;
|
|
}
|
|
}
|
|
|
|
function log($level, $msg)
|
|
{
|
|
common_log($level, get_class($this) . ' ('. $this->get_id() .'): '.$msg);
|
|
}
|
|
|
|
function name()
|
|
{
|
|
return strtolower(get_class($this).'.'.$this->get_id());
|
|
}
|
|
}
|
|
|