PHP Multithreading using pthreads extension

It is possible to solve complex problems splitting tasks in multiple threads. I was interested to solve one particular problem that seems to me that is very typical. Let me describe it. There is a pool of tasks that need to be executed as fast as possible.

So let's get started. To do this, create a data provider object of MyDataProvider (Threaded) class. There will be only one object and it is the same for all threads.

/**
 * Provider data flows
 */
class MyDataProvider extends Threaded
{
    /**
     * @var int How many items in our imaginary database
     */
    private $total = 2000000;

    /**
     * @var int How many items were processed
     */
    private $processed = 0;

    /**
     * We go to the next item and return it
     * 
     * @return mixed
     */
    public function getNext()
    {
        if ($this->processed === $this->total) {
            return null;
        }

        $this->processed++;

        return $this->processed;
    }
}

For each task we will use the MyWorker (Worker) class, which will be stored on the parent process.

/**
 * MyWorker here is used to share the provider between instances MyWork.
 */
class MyWorker extends Worker
{
    /**
     * @var MyDataProvider
     */
    private $provider;

    /**
     * @param MyDataProvider $provider
     */
    public function __construct(MyDataProvider $provider)
    {
        $this->provider = $provider;
    }

    /**
     * Called when sending in the Pool
     */
    public function run()
    {
        // In this example, we do not need to do anything here
    }

    /**
     * Returns the provider
     * 
     * @return MyDataProvider
     */
    public function getProvider()
    {
        return $this->provider;
    }
}

The processing of each pool task, lets suppose it is a kind of time consuming operation, is our bottleneck. So we start a multi-threaded task using MyWork (Threaded) class.

/**
 * MyWork is a task that can be executed in parallel
 */
class MyWork extends Threaded
{

    public function run()
    {
        do {
            $value = null;

            $provider = $this->worker->getProvider();

            // Sync obtaining data
            $provider->synchronized( function( $provider ) use (&$value) {
               $value = $provider->getNext();
            }, $provider);

            if ($value === null) {
                continue;
            }

            // A certain consuming operation
            $count = 100;
            for ($j = 1; $j <= $count; $j++) {
                sqrt($j+$value) + sin($value/$j) + cos($value);
            }
        }
        while ($value !== null);
    }

}

Note that to pick up the data from the service provider you need to use the synchronized() function. Otherwise, there is the possibility that the data be processed more than one time, or you may miss some data.

Now we make it all work using the Pool object.

require_once 'MyWorker.php';
require_once 'MyWork.php';
require_once 'MyDataProvider.php';

$threads = 8;

// Create provider. This service may for example read some data
// from a file or from the database
$provider = new MyDataProvider();

// Create a pool of workers
$pool = new Pool($threads, 'MyWorker', [$provider]);

$start = microtime(true);

// In this case flows are balanced. 
// Therefore, there is good to create as many threads as processes in our pool.
$workers = $threads;
for ($i = 0; $i < $workers; $i++) {
    $pool->submit(new MyWork());
}

$pool->shutdown();

printf("Done for %.2f seconds" . PHP_EOL, microtime(true) - $start);

This is a rather elegant solution in my opinion.

That's all! Well, that is almost everything you need to do.

Actually, there is something that might disappoint the inquisitive reader. This does not work on a standard PHP, compiled with the default options. To use pthreads multi-threading support it is necessary to compile PHP with the ZTS (Zend Thread Safety) option.

PHP Settings

sudo add-apt-repository ppa:ondrej/php-zts
sudo apt update
sudo apt-get install php7.0-zts php7.0-zts-dev

git clone https://github.com/krakjoe/pthreads.git
./configure
make -j8
sudo make install
echo "extension=pthreads.so" > /etc/pthreads.ini
sudo cp pthreads.ini /etc/php/7.0-zts/cli/conf.d/pthreads.ini

$ lscpu
CPU(s):                8
Threads per core:      2
Cores per socket:      4
Model name:            Intel(R) Core(TM) i7-4700HQ CPU @ 2.40GHz