<?php
 
if ( ! class_exists( 'Daemon' ) ) {
        require_once( OWA_INCLUDE_DIR.'Daemon.class.php' );
}
 
if ( ! class_exists( 'CronParser.php' ) ) {
        require_once(OWA_INCLUDE_DIR.'CronParser.php');
}
 
class owa_daemon extends Daemon {
        
        var $pids = array();
        var $params = array();
        var $max_workers = 5;
        var $job_scheduling_interval = 30;
        var $eq;
        var $workerCountByJob = array();
        var $lastExecutionTimeByJob = array();
        var $jobsByPid = array();
        var $defaultMaxWorkersPerJob = 3;
        var $jobs;
        
        function __construct() {
                
                $this->params = $this->getArgs();
                
                if (isset($this->params['interval'])) {
                        $this->job_scheduling_interval = $this->params['interval'];
                }
                
                if (isset($this->params['max_workers'])) {
                        $this->max_workers = $this->params['max_workers'];
                }
                
                if (isset($this->params['pid_file_location'])) {
                        $this->pidFileLocation = $this->params['pid_file_location'];
                }
                
                if (isset($this->params['uid'])) {
                        $this->userID = $this->params['uid'];
                }
                
                if (isset($this->params['gid'])) {
                        $this->groupID = $this->params['gid'];
                }
 
                if (isset($this->params['pid_file_location'])) {
                        $this->pidFileLocation = $this->params['pid_file_location'];
                }
                
                $s = owa_coreAPI::serviceSingleton();
                $this->jobs = $s->getMap('backgound_jobs');
                
                $this->eq = owa_coreAPI::getEventDispatch();
                
                return parent::__construct();
        }
        
        function getArgs() {
                
                $params = array();
                // get params from the command line args
                // $argv is a php super global variable
                global $argv;
                for ( $i=1; $i < count( $argv ); $i++ ) {
                        $it = split("=",$argv[$i]);
                        $params[$it[0]] = $it[1];
                }
                
                return $params;
        }
 
        function _logMessage($msg, $status = DLOG_NOTICE) {
                
                if ($status & DLOG_TO_CONSOLE) {
                echo $msg."\n";
        }
        
                owa_coreAPI::notice("Daemon: $msg");
        }
        
        function isWorkerAvailable() {
                
                $active_workers = count( $this->pids );
                $available_workers = $this->max_workers - $active_workers;
                if ( $available_workers >= 1 ) {
                        return true;
                } else {
                        return false;
                }
        }
        
        function isAnotherWorkerAllowed($job_name, $job_max_workers = '') {
                
                if ( ! $job_max_workers ) {
                        $job_max_workers = $this->defaultMaxWorkersPerJob;
                }
                
                if ( array_key_exists($job_name, $this->workerCountByJob ) ) {
                        if ( $this->workerCountByJob[$job_name] < $job_max_workers) {
                                owa_coreAPI::debug(sprintf(
                                                "New worker processes is allowed for job: %s. %d of %d processes are active.", 
                                                $job_name, 
                                                $this->workerCountByJob[$job_name], $job_max_workers 
                                ));
                                return true;
                        } else {
                                owa_coreAPI::debug(sprintf(
                                                "New worker processes not allowed for job: %s. %d of %d processes are active.", 
                                                $job_name, 
                                                $this->workerCountByJob[$job_name], $job_max_workers 
                                ));
                                return false;
                        }
                } else {
                        owa_coreAPI::debug(sprintf(
                                        "New worker processes is allowed for job: %s. %d of %d processes are active.", 
                                        $job_name, 
                                        $this->workerCountByJob[$job_name], $job_max_workers 
                        ));
                        return true;
                }       
        }
        
        function isTimeForJob($cron_tab, $last_execution_time) {
                
                $cron = new CronParser();
                $cron->calcLastRan($cron_tab);
                $last_due = $cron->getLastRanUnix();
                
                if ($last_due > $last_execution_time) {
                        return true;
                } else {
                        return false;
                }
        }
        
        function getLastExecutionTime($job_name) {
                
                if ( array_key_exists( $job_name, $this->lastExecutionTimeByJob ) ) {
                        return $this->lastExecutionTimeByJob[$job_name];
                } else {
                        return 0;
                }
        }
        
        /**
         * This function is happening in a while loop
         */
        function _doTask() {
                                
                if ( $this->isWorkerAvailable() ) {
                        
                        $jobs = $this->jobs;
                        
                        if ( $jobs ) {
                                $i = 0;
                                //for ($i = 0; $i < $available_workers; $i++) {
                                foreach ($jobs as $k => $job) {
                                        
                                        if ( $this->isAnotherWorkerAllowed( $job['name'], $job['max_processes'] ) && 
                                                 $this->isTimeForJob( $job['cron_tab'], $this->getLastExecutionTime( $job['name'] ) ) ) {
                                                // fork a new child
                                                $pid = pcntl_fork();
                                                if ( ! $pid ) {
                                                        // this part is executed in the child
                                                        owa_coreAPI::debug( 'New child process executing job ' . print_r( $job, true ) );
                                                        pcntl_exec( OWA_DIR.'cli.php', $job['cmd'] ); // takes an array of arguments
                                                        exit();
                                                } elseif ($pid == -1) {
                                                        // happens when something goes wrong and fork fails (handle errors here)
                                                        owa_coreAPI::debug( 'Could not fork new child' );
                                                } else {
                                                        // this part is executed in the parent
                                                        // We add pids to a global array, so that when we get a kill signal
                                                        // we tell the kids to flush and exit.
                                                        if ( array_key_exists( $k, $this->workerCountByJob ) ) {
                                                                $this->workerCountByJob[$k]++;
                                                        } else {
                                                                $this->workerCountByJob[$k] = 1;
                                                                $this->lastExecutionTimeByJob[$k] = time();
                                                                $this->jobsByPid[$pid] = $k;
                                                        }
                                                        
                                                        $this->pids[] = $pid;   
                                                }
                                        }                                                                       
                                }
                        }
                }
 
                // Collect any children which have exited on their own. pcntl_waitpid will
                // return the PID that exited or 0 or ERROR
                // WNOHANG means we won't sit here waiting if there's not a child ready
                // for us to reap immediately
                // -1 means any child
                $dead_and_gone = pcntl_waitpid( -1, $status, WNOHANG );
                
                while( $dead_and_gone > 0 ) {
                        // Remove the gone pid from the array
                        unset( $this->pids[array_search( $dead_and_gone, $this->pids )] );
                        $past_job = $this->jobsByPid[$dead_and_gone];
                        // decrement worker count
                        --$this->workerCountByJob[$past_job];
                        unset($this->jobsByPid[$dead_and_gone]);
                
                        // Look for another one
                        $dead_and_gone = pcntl_waitpid( -1, $status, WNOHANG);
                }
                
                owa_coreAPI::debug(sprintf(
                                "Daemon Statistics -- pidsByJob: %s, workerCountByJob: %s, lastExecutionTimeByJob: %s",
                                print_r( $this->pidsByJob, true),
                                print_r( $this->workerCountByJob, true),
                                print_r( $this->lastExecutiontimeByJob, true)
                ));
                
                // Sleep for some interval
                sleep($this->job_scheduling_interval);
        }
}
 
?>