123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230 |
- <?php
- 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;
- }
-
- public abstract function runThread();
-
- function run()
- {
- $this->initPipes();
- $children = array();
- for ($i = 1; $i <= $this->threads; $i++) {
- $pid = pcntl_fork();
- if ($pid < 0) {
- $this->log(LOG_ERR, "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_ERR, "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;
- }
-
- 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_ERR, "Couldn't create inter-process sockets");
- exit(1);
- }
- }
-
- public function processManager()
- {
- return new ProcessManager($this->parentReader);
- }
-
- protected function shouldRespawn($exitCode)
- {
- if ($exitCode == self::EXIT_SHUTDOWN) {
-
- return false;
- } else {
-
- return true;
- }
- }
-
- protected function initAndRunChild($thread)
- {
-
- fclose($this->parentWriter);
- $this->set_id($this->get_id() . "." . $thread);
- $this->resetDb();
- $exitCode = $this->runThread();
- exit($exitCode);
- }
-
- protected function resetDb()
- {
-
-
- global $_DB_DATAOBJECT;
- unset($_DB_DATAOBJECT['CONNECTIONS']);
-
-
- $cache = Cache::instance();
- if ($cache) {
- $cache->reconnect();
- }
-
- 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());
- }
- }
|