File: include/class.MTDaemon.php

<?php

/**
 * MultiThreaded Daemon (MTD)
 * 
 * Copyright (c) 2007, Benoit Perroud
 * 
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or
 * without modification, are permitted provided that the following
 * conditions are met: Redistributions of source code must retain the
 * above copyright notice, this list of conditions and the following
 * disclaimer. Redistributions in binary form must reproduce the above
 * copyright notice, this list of conditions and the following disclaimer
 * in the documentation and/or other materials provided with the
 * distribution.
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * @package     MTD
 * @author      Benoit Perroud <ben@migtechnology.ch>
 * @copyright   2007 Benoit Perroud
 * @license     http://www.gnu.org/copyleft/gpl.html
 * @version     $Id: class.MTDaemon.php 8 2007-10-19 09:16:00Z killerwhile $
 *
 * See http://code.google.com/p/phpmultithreadeddaemon/ 
 * and http://phpmultithreaddaemon.blogspot.com/ for more information
 *
 */

require_once dirname(__FILE__) . '/class.MTLog.php';

abstract class MTDaemon 
{

    /*
     * Configuration vaiables
     */

    // max concurrent threads
    protected $max_threads = 4; // should be implemented with sem_get('name', $max_aquire) but this can't be dynamically updated as this var.

    // sleep time when no job
    protected $idle_sleep_time = 5; 

    /*
     * Internal constants
     */
    const _INDEX_DATA = 0;    // shared data goes here.
    const _INDEX_THREADS = 1; // how many threads are working
    const _INDEX_SLOTS = 2;   // which slot do the current thread use ?

    /*
     * Internal variables
     */
    protected $shm;                         // = ftok(__FILE__, 'g');
    protected $shared_data;                 // = shm_attach($this->shm);

    protected $mutex;                       // lock critical path, used in lock() and unlock()
    protected $mutex_main_process;          // lock main process only. Children can continue to run
    protected $mutex_children_processes;    // lock children processes only. Main can continue to run

    protected $main_thread_pid;

    /*
     * Constructor
     * 
     * @params $threads : number of concurrent threads, default 4
     * @params $idelsleeptime : time to sleep when no job ready (getNext return null), in seconds, default 5 
     */
    public function __construct($threads = null, $idlesleeptime = null)
    {
        if ($threads) $this->max_threads = $threads;
        if ($idlesleeptime) $this->idle_sleep_time = $idlesleeptime;
        $this->main_thread_pid = posix_getpid();
    }

    /*
     * Hook called just before the main loop.
     * 
     * Remark : cleanup code goes here.
     */
    protected function _prerun()
    {
        MTLog::getInstance()->info('Starting daemon with ' . $this->max_threads . ' slots');

        $this->shm                      = ftok(__FILE__, 'g'); // global shm
        $this->shared_data              = shm_attach($this->shm);

        $this->mutex                    = sem_get($this->shm);
        $this->mutex_main_process       = sem_get(ftok(__FILE__, 'm'));
        $this->mutex_children_processes = sem_get(ftok(__FILE__, 'c'));

        shm_put_var($this->shared_data, self::_INDEX_DATA, array());

        $this->setThreads(0);

        $slots = array();
        for ($i = 0; $i < $this->max_threads; $i++) {
            $slots[] = false;
        }
        shm_put_var($this->shared_data, self::_INDEX_SLOTS, $slots);
    }

    /*
     * Hook called just after the main loop
     */
    protected function _postrun()
    {
        MTLog::getInstance()->info('Stopping daemon. ');

        shm_remove($this->shared_data);
        sem_remove($this->mutex);
        sem_remove($this->mutex_process);
    }

    /*
     * Main loop, request next job using getNext() and execute run($job) in a separate thread
     * _prerun and _postrun hooks are called before and after the main loop -> usefull for cleanup and so on.
     */
    public function handle()
    {

        $this->run = true;

        $this->_prerun();

        while ($this->run) {

            /* 
             * Terminating all child, to not let some zombie leaking the memory.
             */

            MTLog::getInstance()->debug2('-- Next iteration ');

            $this->lock();

            // HACK : avoid zombie and free earlier the memory
            do {
                $res = pcntl_wait($status, WNOHANG);
                MTLog::getInstance()->debug2('$res = pcntl_wait($status, WNOHANG); called with $res = ' . $res);
                if ($res > 0) MTLog::getInstance()->debug('(finishing child with pid ' . $res . ')');
            } while ($res > 0);

            /*
             * Loop until a slot frees 
             */
            while (!$this->hasFreeSlot()) {
                $this->unlock();
                MTLog::getInstance()->debug('No more free slot, waiting');
                $res = pcntl_wait($status); // wait until a child ends up
                MTLog::getInstance()->debug2('$res = pcntl_wait($status); called with $res = ' . $res);
                if ($res > 0) {
                    MTLog::getInstance()->debug('Finishing child with pid ' . $res);
                } else {
                    MTLog::getInstance()->error('Outch1, this souldn\'t happen. Verify your implementation ...');
                    $this->run = false;
                    continue;
                }
                $this->lock();
            }

            $slot = $this->requestSlot();
            $this->incThreads();

            $this->unlock();

            if ($slot === null) {
                var_dump(shm_get_var($this->shared_data, self::_INDEX_DATA));
                var_dump(shm_get_var($this->shared_data, self::_INDEX_THREADS));
                var_dump(shm_get_var($this->shared_data, self::_INDEX_SLOTS));

                MTLog::getInstance()->error('Outch2, this souldn\'t happen. Verify your implementation ...');
                $this->run = false;
                continue;
            }
            
            /*
             * Request next action to handle
             */
            $next = $this->getNext($slot);
            
            /*
             * If no job
             */
            if (!$next) {

                MTLog::getInstance()->debug('No job, sleeping at most ' . $this->idle_sleep_time . ' sec ... ');

// TODO : waiting for signal pushed into a queue when inserting a new job.

                $this->lock();
                $this->releaseSlot($slot);
                $this->decThreads();
                $this->unlock();

                sleep($this->idle_sleep_time);

                continue;
                
            } else { 

                $pid = pcntl_fork();

                if ($pid == -1) {
                        MTLog::getInstance()->error('[fork] Duplication impossible');
                        $this->run = false;
                        continue;
                } else if ($pid) {

                        usleep(10); // HACK : give the hand to the child -> a simple way to better handle zombies

                        continue;
                } else {

                    MTLog::getInstance()->debug('Executing thread #' . posix_getpid() . ' in slot ' . number_format($slot));

                    $res = $this->run($next, $slot);

                    $this->lock();
                    $this->releaseSlot($slot);
                    $this->decThreads();
                    $this->unlock();

                    exit($res);

                }
            }
        }

        $this->_postrun();

    }

    /*
     * Request data of the next element to run in a thread
     * 
     * return null or false if no job currently
     */
    abstract public function getNext($slot);

    /*
     * Process the element fetched by getNext in a new thread
     * 
     * return the exiting status of the thread
     */
    abstract public function run($next, $slot);

    /*
     *
     */
    protected function lock()
    {
        MTLog::getInstance()->debug2('[lock] lock');
        $res = sem_acquire($this->mutex);
        if (!$res) exit(-1);
    }

    /*
     *
     */
    protected function unlock()
    {
        MTLog::getInstance()->debug2('[lock] unlock');
        $res = sem_release($this->mutex);
        if (!$res) exit(-1);
    }

    /*
     *
     */
    protected function lockMain()
    {
        MTLog::getInstance()->debug2('[lock] lock main process');
        $res = sem_acquire($this->mutex_main_process);
        if (!$res) exit(-1);
    }

    /*
     *
     */
    protected function unlockMain()
    {
        MTLog::getInstance()->debug2('[lock] unlock main process');
        $res = sem_release($this->mutex_main_process);
        if (!$res) exit(-1);
    }

    /*
     *
     */
    protected function lockChildren()
    {
        MTLog::getInstance()->debug2('[lock] lock children processes');
        $res = sem_acquire($this->mutex_children_processes);
        if (!$res) exit(-1);
    }

    /*
     *
     */
    protected function unlockChildren()
    {
        MTLog::getInstance()->debug2('[lock] unlock children processes');
        $res = sem_release($this->mutex_children_processes);
        if (!$res) exit(-1);
    }

    /*
     * Get a shared var based on hash.
     *
     * Return null if the var doesn't exist.
     */
    protected function getVar($name, $lock = false)
    {
        if ($lock) $this->lock();
        $vars = shm_get_var($this->shared_data, self::_INDEX_DATA);
        $value = (isset($vars[$name])) ? $vars[$name] : null;
        if ($lock) $this->unlock();
        return $value;
    }

    /*
     * Set a shared var.
     * 
     * Remark : the var should be serialized.
     */
    protected function setVar($name, $value, $lock = false)
    {
        if ($lock) $this->lock();
        $vars = shm_get_var($this->shared_data, self::_INDEX_DATA);
        $vars[$name] = $value;
        $res = shm_put_var($this->shared_data, self::_INDEX_DATA, $vars);
        if ($lock) $this->unlock();
        return $res;
    }

    /*
     * Get the number of running threads
     */
    protected function getThreads($lock = false)
    {
        if ($lock) $this->lock();
        $res = shm_get_var($this->shared_data, self::_INDEX_THREADS);
        if ($lock) $this->unlock();
        return $res;
    }

    /*
     * Set the number of running threads
     */    
    protected function setThreads($threads, $lock = false)
    {
        if ($lock) $this->lock();
        $res = shm_put_var($this->shared_data, self::_INDEX_THREADS, $threads);
        if ($lock) $this->unlock();
        return $res;
    }

    /*
     * Increment the number of running threads
     */
    protected function incThreads($lock = false)
    {
        if ($lock) $this->lock();
        $threads = $this->getThreads();
        $res = shm_put_var($this->shared_data, self::_INDEX_THREADS, $threads + 1);
        MTLog::getInstance()->debug('incThreads, $threads = ' . ($threads + 1));
        if ($lock) $this->unlock();
        return $res;
    }

    /*
     * Decrement the number of running threads
     */
    protected function decThreads($lock = false)
    {
        if ($lock) $this->lock();
        $threads = $this->getThreads();
        $res = shm_put_var($this->shared_data, self::_INDEX_THREADS, $threads - 1);
        MTLog::getInstance()->debug('decThreads, $threads = ' . ($threads - 1));
        if ($lock) $this->unlock();
        return $res;
    }

    /*
     * Return true if any slot is free
     */
    protected function hasFreeSlot()
    {
        $threads = $this->getThreads();
        $res = ($threads < $this->max_threads) ? true : false;
        MTLog::getInstance()->debug('Has free slot ? => #running threads = ' . $threads);
        return $res;
    }

    /*
     * Assign a free slot
     *
     * Return null if no free slot is available
     */
    protected function requestSlot($lock = false)
    {
        MTLog::getInstance()->debug('Requesting slot ... ');
        $slot = null;
        if ($lock) $this->lock();
        $slots = shm_get_var($this->shared_data, self::_INDEX_SLOTS);
        for ($i = 0; $i < $this->max_threads; $i++) {
            if (!isset($slots[$i])) {
                $slots[$i] = true;
                $slot = $i;
                break;
            } else {
                if ($slots[$i] == false) {
                    $slots[$i] = true;
                    $slot = $i;
                    break;
                }
            }
        }
        shm_put_var($this->shared_data, self::_INDEX_SLOTS, $slots);
        if ($lock) $this->unlock();
        if (is_null($slot)) {
            MTLog::getInstance()->debug('no free slots !!');
        } else {
            MTLog::getInstance()->debug('slot ' . $slot . ' found.');
        }
        return $slot;
    }

    /*
     * Release given slot
     */
    protected function releaseSlot($slot, $lock = false) {
        if ($lock) $this->lock();
        $slots = shm_get_var($this->shared_data, self::_INDEX_SLOTS);
        $slots[$slot] = false;
        shm_put_var($this->shared_data, self::_INDEX_SLOTS, $slots);
        if ($lock) $this->unlock();
        MTLog::getInstance()->debug('Releasing slot ' . $slot);
        return true;
    }

}