[RESOLVED] how to fork off a time dependent process with PHP

[sfullman]

I have a procedure I need to run which basically is like this:

foreach($records as $record){
   //contact another website's API via curl..

   //record the information in the database

}

The real problem here is that it takes time to get the reply from the other server - sometimes as much as 60 seconds, which they can't help me with

SO, I started looking into proc_open and some approach to multithreading, and so far I have managed to lock up my server several times and need a reboot from hostgator.. ugh

so my thought was to do as follows. First, here's the code I found on the web:

class Thread {
# from http://www.alternateinterior.com/2007/05/multi-threading-strategies-in-php.html
var $pref ; // process reference
var $pipes; // stdio
var $buffer; // output buffer

/* private */ function Thread() {
		$this->pref = 0;
		$this->buffer = "";
		$this->pipes = (array)NULL;
}
/* public static */ function Create ($file) {

	$t = new Thread;
	$descriptor = array (
		0 => array ("pipe", "r"), 
		1 => array ("pipe", "w"), 
		2 => array ("pipe", "w")
	);
	prn('begin');
	$t->pref = proc_open ("php -f $file ", $descriptor, $t->pipes);
	print_r($t->pref);

	stream_set_blocking ($t->pipes[1], 0);
	exit;
	return $t;
}

/* public instance */ function isActive () {
		$this->buffer .= $this->listen();
		$f = stream_get_meta_data ($this->pipes[1]);
		return !$f["eof"];
}
/* public instance */ function close () {

	$r = proc_close ($this->pref);
	$this->pref = NULL;
	return $r;
}
/* public instance */ function tell ($thought) {
		fwrite ($this->pipes[0], $thought);

}
/* public instance */ function listen () {
		$buffer = $this->buffer;
		$this->buffer = "";
		while ($r = fgets ($this->pipes[1], 1024)) {
				$buffer .= $r;

	}
	return $buffer;
}
/* public instance */ function getError () {
		$buffer = "";
		while ($r = fgets ($this->pipes[2], 1024)) {

			$buffer .= $r;
	}
	return $buffer;
}
}

and here's my brilliant idea of how to use it (except of course for the fact it doesn't work 🙂

foreach($records as $idx=>$record){
	prn('[p] '. date('H:i:s').' processing record '. $idx);
	//use your imagination here, we have to do a time intensive process with the record - done in myfile.php
	$threadCount=15; //number of processes allowed to run
	$processed=false; //we do not have a thread yet
	for($i=1; $i<=$threadCount; $i++){
		if(!$threads[$i]){
			//run the call on this node
			$processed=true;
			prn('	starting new thread on '.$i);
			$threads[$i]=Thread::create('contactAPI.php');
	exit;
			break;
		}else if($threads[$i]->isActive()){
			//we cannot use this thread index, it is busy
			prn('	thread '.$i.' busy');
			continue;
		}else{
			//close the thread and run the new process
			$threads[$i]->close();
			$processed=true;
			prn('	reusing thread '.$i);
			$threads[$i]=Thread::create('contactAPI.php');
			break;
		}
	}
	if(!$processed){
		//handle overflow somehow, or notify of error etc.
		prn( 'unable to process this record index '.$idx);
	}
}

I mean, that looks simple, and I think the logic is sound but calling Thread::create seems to create process that hang around long after, AND the file I am calling never does anything - I've tried to ahve it email me when it's called etc. but no dice.

The bottom line is this:
1. I need to be able to somehow call a process with some parameters, which may take a while to run, and have that process run while I loop on to the next process
2. I need to know whether these processes are active so I can avoid setting a million processes in motion and locking the system up.

I'm searching for more but with the server going down it's hard to test this out 🙁

Samuel

[bradgrafelman]

Something you need to do "periodically" sounds like you just need to setup a cronjob; have you looked into this instead?

[sfullman]

OK Brad, not periodically at all, what i need to do is rip through a million records, and send the data off via an API. HOWEVER:

1. the api may take 6 or more seconds, sometimes 30
2. php is an inline process so, I must wait till each of those are done to go on to the next record
3. that would take, approximately, forever (though forever is much longer actually)

I'm actually working on this now, and will add an article and link to for the community. It's, in my opinion, really cool 🙂

[sfullman]

OK, I am happy enough with this to post it for the community. This shoulld help a lot of people trying to do this

steps:
1. YOU MUST HAVE PHP5 to run this, since proc_get_status() is a php5 function.
2. Create a file test.email.php and make sure it is executable:

#!/usr/local/bin/php -q
<?php
ob_start();
require('config.php');

parse_str($argv[1],$a);
ob_start();
print_r($a);
$out=ob_get_contents();
ob_end_clean();
extract($a);

echo "start\n";
//do something time-intensive
mail($developerEmail, "test $test, record $record, thread $thread",$out,$fromHdrBugs);
sleep(5);
echo "stop\n";
ob_end_clean();

?>

3. Create another file, which you will call in the browser to see how it works:

   session_start();
   $_SESSION['special']['test']++;
   
   register_shutdown_function('iamdone');
   function iamdone(){
   	echo 'i am done';
   }
   $descriptorspec = array(
   	0 => array("pipe","r"),
   	1 => array("pipe","w"),
   	2 => array("file","./error.log","a")
   );
   
   // define current working directory where files would be stored
   $cwd = './' ;
   
   echo '<pre>';
   
   $recordCount=30;
   $threads=4;
   $failsafeBreak=4;
   
   echo 'test #'.$_SESSION['special']['test']."\n";
   for($record=1; $record<=$recordCount; $record++){
   	echo "-- processing record $record\n";
   
   # no longer needed: $processed=false;
   
   $failsafe=0;
   while(true){
   	$failsafe++;
   	for($i=1; $i<=$threads; $i++){
   		if(is_resource($processes[$i])){
   			$a=proc_get_status($processes[$i]);
   			if($a['running']){
   				//cannot use this node
   				echo "	thread $i busy\n";
   				continue;
   			}else{
   				//create the thread
   				echo "	running record $record (r) on thread $i\n";
   				proc_close($processes[$i]);
   				$processes[$i]=proc_open('./test.email.php "test='.$_SESSION['special']['test'].'&record='.$record.'&thread='.$i.'"', $descriptorspec, $pipes, $cwd);
   				$processed=true;
   				break(2);
   			}
   		}else{
   			//run on this node
   			echo "	running record $record on thread $i\n";
   			$processes[$i]=proc_open('./test.email.php "test='.$_SESSION['special']['test'].'&record='.$record.'&thread='.$i.'"', $descriptorspec, $pipes, $cwd);
   			$processed=true;
   			break(2);
   		}
   	}
   	echo "all threads busy, waiting..\n";
   	sleep(2);
   	if($failsafe >= $failsafeBreak){
   		echo "failsafe count exceeded, no thread available to process record $record\n";
   		break;
   	}
   	echo "resuming with pass ".($failsafe+1)."..\n";
   }
   
   # no longer needed: if(!$processed)	echo "no thread available to process record $record\n";
   }
   

   in a nutshell, this calls a process in a 2nd file that may take some time, but also allows you to control how many threads are running. you would set the $failsafeBreak value higher than 4 in practice. The loop will "wait" until there is an available thread.
   
   This coding should be helpful on any looping process that needs to talk across a network.
   
   Samuel

[sfullman]

I tested it using cURL on an API that took 30 seconds sometime per thread - I upped the number of threads to 20. Getting 50 records, the script had to pause for 2 seconds a total of 18 times. So the process was "DONE" in about a minute with all the sub-processes finishing up. Cumulatively, the average time was about 20 seconds, so what would have previusly taken 50*20=1000 seconds to process was done in about a minute. Not bad, thanks to proc_open()

If anyone out there pulls strings, will be glad to write an article about this one.

Samuel