Executing functions in parallel

Question

I have a function that needs to go over around 20K rows from an array, and apply an external script to each. This is a slow process, as PHP is waiting for the script to be executed before continuing with the next row.

In order to make this process faster I was thinking on running the function in different parts, at the same time. So, for example, rows 0 to 2000 as one function, 2001 to 4000 on another one, and so on. How can I do this in a neat way? I could make different cron jobs, one for each function with different params: myFunction(0, 2000), then another cron job with myFunction(2001, 4000), etc. but that doesn't seem too clean. What's a good way of doing this?

Answer 1

Not sure if a solution for your situation but you can redirect the output of system calls to a file, thus PHP will not wait until the program is finished. Although this may result in overloading your server.

http://www.php.net/manual/en/function.exec.php - If a program is started with this function, in order for it to continue running in the background, the output of the program must be redirected to a file or another output stream. Failing to do so will cause PHP to hang until the execution of the program ends.

Answer 2

If you'd like to execute parallel tasks in PHP, I would consider using Gearman. Another approach would be to use pcntl_fork(), but I'd prefer actual workers when it's task based.

Answer 3

Have a look at pcntl_fork. This allows you to spawn child processes which can then do the separate work that you need.

Answer 4

you can use "PTHREADS"

very easy to install and works great on windows

download from here -> http://windows.php.net/downloads/pecl/releases/pthreads/2.0.4/

Extract the zip file and then

• move the file 'php_pthreads.dll' to php\ext\ directory.

• move the file 'pthreadVC2.dll' to php\ directory.

then add this line in your 'php.ini' file:

extension=php_pthreads.dll

save the file.

you just done :-)

now lets see example of how to use it:

class ChildThread extends Thread {
    public $data;

    public function run() {
        /* Do some expensive work */

        $this->data = 'result of expensive work';
    }
}

$thread = new ChildThread();

if ($thread->start()) {     
    /*
     * Do some expensive work, while already doing other
     * work in the child thread.
     */

    // wait until thread is finished
    $thread->join();

    // we can now even access $thread->data
}

for more information about PTHREADS read php docs here:

PHP DOCS PTHREADS

• if you'r using WAMP like me, then you should add 'pthreadVC2.dll' into \wamp\bin\apache\ApacheX.X.X\bin and also edit the 'php.ini' file (same path) and add the same line as before

  extension=php_pthreads.dll

GOOD LUCK!

Answer 5

The only waiting time you suffer is between getting the data and processing the data. Processing the data is actually completely blocking anyway (you just simply have to wait for it). You will not likely gain any benefits past increasing the number of processes to the number of cores that you have. Basically I think this means the number of processes is small so scheduling the execution of 2-8 processes doesn't sound that hideous. If you are worried about not being able to process data while retrieving data, you could in theory get your data from the database in small blocks, and then distribute the processing load between a few processes, one for each core.

I think I align more with the forking child processes approach for actually running the processing threads. There is a brilliant demonstration in the comments on the pcntl_fork doc page showing an implementation of a job daemon class

http://php.net/manual/en/function.pcntl-fork.php

<?php 
declare(ticks=1); 
//A very basic job daemon that you can extend to your needs. 
class JobDaemon{ 

    public $maxProcesses = 25; 
    protected $jobsStarted = 0; 
    protected $currentJobs = array(); 
    protected $signalQueue=array();   
    protected $parentPID; 

    public function __construct(){ 
        echo "constructed \n"; 
        $this->parentPID = getmypid(); 
        pcntl_signal(SIGCHLD, array($this, "childSignalHandler")); 
    } 

    /** 
    * Run the Daemon 
    */ 
    public function run(){ 
        echo "Running \n"; 
        for($i=0; $i<10000; $i++){ 
            $jobID = rand(0,10000000000000); 

            while(count($this->currentJobs) >= $this->maxProcesses){ 
               echo "Maximum children allowed, waiting...\n"; 
               sleep(1); 
            } 

            $launched = $this->launchJob($jobID); 
        } 

        //Wait for child processes to finish before exiting here 
        while(count($this->currentJobs)){ 
            echo "Waiting for current jobs to finish... \n"; 
            sleep(1); 
        } 
    } 

    /** 
    * Launch a job from the job queue 
    */ 
    protected function launchJob($jobID){ 
        $pid = pcntl_fork(); 
        if($pid == -1){ 
            //Problem launching the job 
            error_log('Could not launch new job, exiting'); 
            return false; 
        } 
        else if ($pid){ 
            // Parent process 
            // Sometimes you can receive a signal to the childSignalHandler function before this code executes if 
            // the child script executes quickly enough! 
            // 
            $this->currentJobs[$pid] = $jobID; 

            // In the event that a signal for this pid was caught before we get here, it will be in our signalQueue array 
            // So let's go ahead and process it now as if we'd just received the signal 
            if(isset($this->signalQueue[$pid])){ 
                echo "found $pid in the signal queue, processing it now \n"; 
                $this->childSignalHandler(SIGCHLD, $pid, $this->signalQueue[$pid]); 
                unset($this->signalQueue[$pid]); 
            } 
        } 
        else{ 
            //Forked child, do your deeds.... 
            $exitStatus = 0; //Error code if you need to or whatever 
            echo "Doing something fun in pid ".getmypid()."\n"; 
            exit($exitStatus); 
        } 
        return true; 
    } 

    public function childSignalHandler($signo, $pid=null, $status=null){ 

        //If no pid is provided, that means we're getting the signal from the system.  Let's figure out 
        //which child process ended 
        if(!$pid){ 
            $pid = pcntl_waitpid(-1, $status, WNOHANG); 
        } 

        //Make sure we get all of the exited children 
        while($pid > 0){ 
            if($pid && isset($this->currentJobs[$pid])){ 
                $exitCode = pcntl_wexitstatus($status); 
                if($exitCode != 0){ 
                    echo "$pid exited with status ".$exitCode."\n"; 
                } 
                unset($this->currentJobs[$pid]); 
            } 
            else if($pid){ 
                //Oh no, our job has finished before this parent process could even note that it had been launched! 
                //Let's make note of it and handle it when the parent process is ready for it 
                echo "..... Adding $pid to the signal queue ..... \n"; 
                $this->signalQueue[$pid] = $status; 
            } 
            $pid = pcntl_waitpid(-1, $status, WNOHANG); 
        } 
        return true; 
    } 
}