+
−<?php
+
−
+
−/**
+
− * Multi-threading (well sort of) tools
+
− * 
+
− * Greyhound - real web management for Amarok
+
− * Copyright (C) 2008 Dan Fuhry
+
− *
+
− * This program is Free Software; you can redistribute and/or modify it under the terms of the GNU General Public License
+
− * as published by the Free Software Foundation; either version 2 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 General Public License for details.
+
− */
+
−
+
−require_once('json.php');
+
−
+
−/**
+
− * Global signal handler for SIGCHLD.
+
− */
+
−
+
−function Threader_SigChld()
+
−{
+
−  global $threader_instances;
+
−  foreach ( $threader_instances as &$mt )
+
−  {
+
−    if ( is_object($mt) )
+
−    {
+
−      $mt->event_sigchld();
+
−    }
+
−  }
+
−}
+
−
+
−/**
+
− * Global signal handler for SIGUSR2.
+
− */
+
−
+
−function Threader_SigUsr2()
+
−{
+
−  global $threader_instances, $threader_notick;
+
−  if ( @$threader_notick )
+
−    return;
+
−  foreach ( $threader_instances as &$mt )
+
−  {
+
−    if ( is_object($mt) )
+
−    {
+
−      $mt->event_sigusr2();
+
−    }
+
−  }
+
−}
+
−
+
−/**
+
− * List of Threader instances. Needed for global handling of signals.
+
− * @var array
+
− */
+
−
+
−global $threader_instances;
+
−$threader_instances = array();
+
−
+
−/**
+
− * Tools for emulating multi-threaded operation in PHP scripts.
+
− * @package Amarok
+
− * @subpackage WebControl
+
− * @author Dan Fuhry
+
− * @license GNU General Public License <http://www.gnu.org/licenses/old-licenses/gpl-2.0.html>
+
− */
+
−
+
−class Threader
+
−{
+
−  
+
−  /**
+
−   * Return value of fork() if the process is a child.
+
−   * @const int
+
−   */
+
−  
+
−  const FORK_CHILD = -1;
+
−  
+
−  /**
+
−   * Set to true if this is a child process. No exceptions.
+
−   * @var bool
+
−   * @access private
+
−   */
+
−  
+
−  private $is_child = false;
+
−  
+
−  /**
+
−   * Sockets for inter-process communication.
+
−   * @var array
+
−   * @access private
+
−   */
+
−  
+
−  protected $ipc_sockets = array();
+
−  
+
−  /**
+
−   * Socket for communication with the parent. Obviously only used after calling fork().
+
−   * @var resource
+
−   * @access private
+
−   */
+
−  
+
−  protected $parent_sock = false;
+
−  
+
−  /**
+
−   * Services_JSON instance.
+
−   * @var object
+
−   * @access private
+
−   */
+
−  
+
−  protected $json = false;
+
−  
+
−  /**
+
−   * PID of the parent process.
+
−   * @var int
+
−   * @access private
+
−   */
+
−  
+
−  protected $parent_pid = 1;
+
−  
+
−  /**
+
−   * List of actions for IPC events.
+
−   * @var array
+
−   * @access private
+
−   */
+
−  
+
−  protected $ipc_actions = array();
+
−  
+
−  /**
+
−   * Constructor. Sets up signal handlers. Nothing to see here, move along.
+
−   */
+
−  
+
−  public function __construct()
+
−  {
+
−    global $threader_instances;
+
−    
+
−    if ( function_exists('pcntl_signal') )
+
−    {
+
−      declare(ticks=1);
+
−      
+
−      $threader_instances[] =& $this;
+
−      
+
−      pcntl_signal(SIGCHLD, 'Threader_SigChld');
+
−      pcntl_signal(SIGUSR2, 'Threader_SigUsr2');
+
−    }
+
−    
+
−    $this->json = new Services_JSON(SERVICES_JSON_LOOSE_TYPE);
+
−    $this->parent_pid = getmypid();
+
−  }
+
−  
+
−  /**
+
−   * Forks the current process. See your system's fork(2) man page for details.
+
−   * @return int FORK_CHILD if child process, PID of child if parent process. Returns false on failure.
+
−   */
+
−  
+
−  public function fork()
+
−  {
+
−    // create our new sockets for IPC
+
−    $socket_pair = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP);
+
−    // fork (emoticon of the day: --<E)
+
−    $fork_result = pcntl_fork();
+
−    if ( $fork_result == -1 )
+
−    {
+
−      // fork failed.
+
−      return false;
+
−    }
+
−    else if ( $fork_result )
+
−    {
+
−      // we are the parent - register the child
+
−      fclose($socket_pair[0]);
+
−      $this->ipc_sockets[$fork_result] = $socket_pair[1];
+
−      return $fork_result;
+
−    }
+
−    else
+
−    {
+
−      // we are the child.
+
−      fclose($socket_pair[1]);
+
−      $this->parent_sock = $socket_pair[0];
+
−      $this->is_child = true;
+
−      return self::FORK_CHILD;
+
−    }
+
−  }
+
−  
+
−  /**
+
−   * Are we a child?
+
−   * @return bool
+
−   */
+
−  
+
−  public function is_child()
+
−  {
+
−    return $this->is_child;
+
−  }
+
−  
+
−  /**
+
−   * Register an action so that when it is fired over IPC, a custom function can be called.
+
−   * @param string Action
+
−   * @param callback Function to call
+
−   * @return true on success, false on failure
+
−   */
+
−  
+
−  function ipc_register($action, $callback)
+
−  {
+
−    if ( !is_string($action) || empty($action) || !is_callable($callback) )
+
−    {
+
−      return false;
+
−    }
+
−    $this->ipc_actions[$action] = $callback;
+
−    return true;
+
−  }
+
−  
+
−  /**
+
−   * Send through an IPC event. If this is a child, it only notifies the parent; if we're the parent, all children are notified. 
+
−   * @param array Data to be sent through. This must be an associative array containing an "action" key at minimum. If this a key "propagate" set to true, a parent that receives this will propagate the message to all children.
+
−   * @return null
+
−   */
+
−  
+
−  function ipc_send($data)
+
−  {
+
−    if ( !isset($data['action']) )
+
−    {
+
−      return false;
+
−    }
+
−    $data = $this->json->encode($data);
+
−    if ( $this->is_child() )
+
−    {
+
−      fwrite($this->parent_sock, "$data\n");
+
−      // signal the parent that we've got an event
+
−      posix_kill($this->parent_pid, SIGUSR2);
+
−    }
+
−    else
+
−    {
+
−      // signal each child
+
−      foreach ( $this->ipc_sockets as $pid => $socket )
+
−      {
+
−        fwrite($socket, "$data\n");
+
−        posix_kill($pid, SIGUSR2);
+
−      }
+
−    }
+
−    return null;
+
−  }
+
−  
+
−  /**
+
−   * Handler for SIGCHLD events.
+
−   * @access private
+
−   */
+
−  
+
−  function event_sigchld()
+
−  {
+
−    // this should never happen to children.
+
−    if ( $this->is_child() )
+
−    {
+
−      return null;
+
−    }
+
−    
+
−    // wait for child to exit.
+
−    pcntl_wait($status);
+
−    // for each child PID, kill with signal 0 (effectively, test if process is alive)
+
−    // if posix_kill fails, it's dead so remove it from the list.
+
−    foreach ( $this->ipc_sockets as $pid => $socket )
+
−    {
+
−      if ( !@posix_kill($pid, 0) )
+
−      {
+
−        // signal failed.
+
−        fclose($socket);
+
−        unset($this->ipc_sockets[$pid]);
+
−      }
+
−    }
+
−  }
+
−  
+
−  /**
+
−   * Handler for IPC events.
+
−   * @access private
+
−   */
+
−  
+
−  function event_sigusr2()
+
−  {
+
−    if ( $this->is_child() )
+
−    {
+
−      // this is easy - the parent sent the signal.
+
−      @stream_set_blocking($this->parent_sock, 0);
+
−      $command = rtrim(fgets($this->parent_sock, 102400), "\n");
+
−    }
+
−    else
+
−    {
+
−      // since we can't find which PID sent the signal, set the timeout to a very small amount
+
−      // of time and try to read; if we get something, awesome.
+
−      foreach ( $this->ipc_sockets as $pid => $socket )
+
−      {
+
−        @stream_set_blocking($socket, 0);
+
−        $command = rtrim(@fgets($socket, 102400), "\n");
+
−        if ( !empty($command) )
+
−        {
+
−          break;
+
−        }
+
−      }
+
−    }
+
−    if ( empty($command) )
+
−    {
+
−      // hmm, got a sigusr2 without an incoming command. oh well, ignore.
+
−      return null;
+
−    }
+
−    $command = $this->json->decode($command);
+
−    if ( !isset($command['action']) )
+
−    {
+
−      // no action = no way to figure out how to handle this.
+
−      return null;
+
−    }
+
−    if ( !isset($this->ipc_actions[$command['action']]) )
+
−    {
+
−      // action not registered
+
−      return null;
+
−    }
+
−    // should we propagate this event?
+
−    if ( !$this->is_child() && isset($command['propagate']) && $command['propagate'] === true )
+
−    {
+
−      $this->ipc_send($command);
+
−    }
+
−    // we're good
+
−    call_user_func($this->ipc_actions[$command['action']], $command, $this);
+
−  }
+
−  
+
−  /**
+
−   * Kills all child processes.
+
−   * @access public
+
−   */
+
−  
+
−  public function kill_all_children()
+
−  {
+
−    foreach ( $this->ipc_sockets as $pid => $socket )
+
−    {
+
−      $socklen = count($this->ipc_sockets);
+
−      posix_kill($pid, SIGTERM);
+
−      // wait until we are conscious of this child's death
+
−      while ( count($this->ipc_sockets) >= $socklen )
+
−      {
+
−        usleep(20000);
+
−      }
+
−    }
+
−  }
+
−  
+
−}
+
−
+
−?>