<?phpdefine('LINE_BREAK',"<br>");define('SERVER_LOAD_MAX',5.0);define('SERVER_MAX_THREADS',300);define('DEBUG',1);/** * * This is generic OS Threading class for PHP. Basically allows you to add processes to the system. * So rather then doing the thread creation inside the php process this actually starts a new * process in the OS for each. * * There's definately drawbacks to this in that you have the overhead of running php for each. However * vs. fork() I've found it to be a cleaner and faster solution. * * For the best use this should be used in conjunction with a: * * 1. data management class to lock records so you don't end up with colisions. * 2. que class to que jobs to be run. * */class Threader{ private $load; private $argvs; private $fileloc; /** * Tries to run a new process * * @return bool - 1=success,0=to high of a load * */ public function startNewProcess() { $this->getCurrentLoad(); if($this->load < SERVER_LOAD_MAX) { $cmd = $this->buildThreadCommand(); if($cmd) { echo $cmd.LINE_BREAK; if(!DEBUG) $this->cleanExec($cmd); return 1; } } } /** * Sets the location of the file to be ran by the thread * i.e. /home/user/files/myfile.php * * @param string $fileloc */ public function setFileLoc($fileloc) { $this->fileloc = $fileloc; } /** * Gets the value in the file location var * * @return string */ public function getFileLoc(){ return $this->fileloc; } /** * Sets the argv values takes a array as input * with the argv positions as the array keys, 0,1,2,3 * * @param array $argvs */ public function setArgvs($argvs) { $this->argvs = $argvs; } /** * returns currently set argv values * * @return array */ public function getArgvs(){ return $this->argvs; } /** * returns the number of process running. Can take a file location * or by defualt returns all php files * * @param string $fileloc * @return int */ public function getNumProcessesRunning($fileloc = 'php') { return count($this->getProcesses($fileloc)); } /** * returns all the processes running. Can take a file location * or by default returns all php processes. * * @param string $fileloc * @return array - of process strings */ public function getProcesses($fileloc = 'php') { $ts = array(); $count = 0; $cmd = "ps aux | grep $fileloc"; $lines = $this->cleanExec($cmd); foreach($lines as $line) { if(!stripos($line,'grep')) { $regex = "(\S+)"; preg_match_all($regex,$line,$matches); $ts[$count]['process'] = $matches[0][1]; $ts[$count]['file'] = $matches[0][12]; $count++; } } return $ts; } /** * kills all the php processes * * @return bool - 1=succes,0=failure */ public function killAllPhpProcesses() { $cmd = 'killall -9 php'; $this->cleanExec($cmd); return 1; } /** * kill specific process * * @param int $id * @return bool */ public function killProcesses($ids) { foreach($ids as $id) { $this->killProcess($id); } } public function killProcess($id) { if($id) { $cmd = 'kill $id'; $this->cleanExec($cmd); return 1; } } /** * get the current load of the system * * @return array * $load['1min'] * $load['5min'] * $load['15min'] */ public function getCurrentLoad() { $cmd = "uptime"; $output = exec($cmd); $sp = explode("average: ",$output); $sp = explode(", ",$sp[1]); $load['1min'] = trim($sp[0]); $load['5min'] = trim($sp[1]); $load['15min'] = trim($sp[2]); $this->load = $load['1min']; return $load; } ///////////// INTERNAL ////////////////////// private function buildThreadCommand() { if(isset($this->fileloc)) { $cmd = "php -q $this->fileloc "; foreach($this->argvs as $argv) { if($argv != null) { $cmd .= "$argv "; } } return $cmd; } } private function cleanExec($cmd) { exec($cmd,$output); return $output; }}//////// TESTING /////////////*// pull job from que db or api call to make the server do some work$jobId = '324834';$fileloc = '/somefile';$argvs = array('option1','option2');$t = new Threader();$t->setFileLoc($fileloc);$t->setArgvs($argvs);$res = $t->startNewProcess();if($res){// update flag on jobecho "Job: $jobId - Started".LINE_BREAK;}echo $t->getNumProcessesRunning();echo LINE_BREAK;var_dump($t->getProcesses());echo LINE_BREAK;echo var_dump($t->getCurrentLoad());*/?>