<?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);
	}
}

?>