Andrew Dorokhov
Docs and background:
The process finishes when the last thread finishes.
Any application objects meant for multithreaded use should extend Threaded.
Installation
1. PHP built with ZTS
Make sure PHP was compiled with ZTS (Zend Thread Safety). This is a compile-time setting and cannot be changed later. Check with:
php -v
Example output:
PHP 7.2.2 (cli) (built: Feb 28 2018 10:47:41) ( ZTS )
Copyright (c) 1997-2018 The PHP Group
Zend Engine v3.2.0, Copyright (c) 1998-2018 Zend Technologies
ZTS must be present.
On Ubuntu you can use a third-party repository for ZTS builds:
sudo add-apt-repository ppa:ondrej/php-zts
sudo apt-get update
2. Install the pthreads extension
Install the pthreads PHP extension the same way you install other extensions.
Threaded
Threaded encapsulates data and work that participate in multithreading.
It is often used as a job object: it has no start() of its own in typical usage, and instances are passed to workers. For starting a dedicated thread of execution, prefer Thread, which extends Threaded.
Threaded implements:
ArrayAccessTraversableCountable
ArrayAccess
Because Threaded implements ArrayAccess, you can treat the object like an array:
class FirstThread extends \Thread
{
public function run()
{
// ...
}
}
$first = new FirstThread;
$first[] = 'foo';
$first[] = 'bar';
$first[] = 'baz';
var_dump($first);
Extra helpers:
$first->merge(['hello']); // append elements
$first->pop(); // remove and return the last element
$first->shift(); // remove and return the first element
$first->chunk($size);
Countable
$first = new FirstThread;
$first[] = 'foo';
$first[] = 'bar';
$first[] = 'baz';
echo count($first);
Traversable
You can use foreach, including inside the thread:
class FirstThread extends \Thread
{
public function run()
{
foreach ($this as $value) {
echo $value;
echo "\n";
}
}
}
$first = new FirstThread;
$first[] = 'foo';
$first[] = 'bar';
$first[] = 'baz';
$first->start();
Thread
Usually used to run work in a new thread. Extend \Thread and implement run():
class First extends \Thread
{
public function run()
{
foreach (range(1, 10) as $x) {
echo $x;
echo "\n";
}
}
}
$thread = new First;
$thread->start();
echo "Started!";
- Extend
\Thread. - Put the thread body in
run(). - Call
start()to run that body in a new thread.
Another pattern — do work in the parent while the child runs, then join and read results:
class ChildThread extends Thread
{
public $data;
public function run()
{
/* Do some work */
$this->data = 'result';
}
}
$thread = new ChildThread();
if ($thread->start()) {
/*
* Do some work here while the child thread runs.
*/
$thread->join();
// Safe to read $thread->data after join
}
run() starts in a separate thread once Thread::start() is called from the context that created the thread. You can only start and join a thread from that same creation context.
$thread->start();
$thread->join(); // wait until the thread finishes
Methods
Start a thread (inheritance flags are optional):
$first->start();
// $first->start(PTHREADS_INHERIT_ALL); // default
// $first->start(PTHREADS_INHERIT_NONE);
// $first->start(PTHREADS_INHERIT_INI); // inherit INI entries
Check whether the thread is still running:
$first->isRunning();
Check whether the thread ended with a fatal error (a critical error in a thread does not tear down the main process):
$first->isTerminated();
Keep a reference in the parent
Keep a live reference to the thread object in the parent after creation:
$searches = ['cats', 'dogs', 'birds'];
foreach ($searches as &$search) {
$search = new SearchGoogle($search);
$search->start();
}
Worker
A Worker is a dedicated thread that runs stacked jobs in order:
<?php
class MyWork extends Threaded
{
private $name;
public function __construct($name)
{
$this->name = $name;
}
public function run()
{
echo $this->name . ":";
for ($i = 0; $i < 10; $i++) {
echo ".";
sleep(1);
}
}
}
$worker = new Worker;
for ($i = 0; $i < 5; $i++) {
$worker->stack(new MyWork("test$i"));
}
$worker->start();
Synchronization
Threads share one address space, so they can exchange data via object properties (or, alternatively, a shared static holder: one thread writes, another reads).
Without safe patterns you get concurrent-access bugs.
Unsafe sharing
class W extends Thread
{
public $data;
public function run()
{
for ($i = 0; $i < 10; $i++) {
echo ".\n";
if ($this->data) {
echo "We got: $this->data\n";
$this->data = null;
}
sleep(1);
}
}
}
$w = new W;
$w->start();
sleep(3);
$w->data = "foo";
sleep(2);
$w->data = "bar";
This may appear to work, but concurrent reads and writes without synchronization are not reliable.
join() — wait for another thread
The simplest sync: wait until a thread finishes, then read its data.
class First extends \Thread
{
// ...
}
$thread = new First;
$thread->start();
echo "Started!";
echo "Joining!";
$thread->join();
- The current context attaches to the given thread.
- It blocks until that thread completes.
- That is the simplest form of synchronization: one thread waits for another.
synchronized(), wait(), and notify()
synchronized() runs a block so access to shared state is coordinated and you avoid unsafe concurrent access to a shared resource.
class First extends \Thread
{
public $signal;
public function run()
{
$thread = $this;
$this->synchronized(
function () use ($thread) {
if (empty($thread->signal)) {
$thread->wait();
}
}
);
}
}
$first = new First;
$first->start();
Here signal is meant to be set from another thread. With only the code above, the thread waits forever.
If you set the signal from outside:
$first = new First;
$first->start();
sleep(1);
$first->signal = "test";
the thread can finish after about a second once it observes the change (depending on how wait/notify is used).
wait()makes the thread wait for a notification from another thread.notify()sends that notification.
Example with explicit wait/notify:
class My extends Thread
{
public function run()
{
/** cause this thread to wait **/
$this->synchronized(function ($thread) {
if (!$thread->done) {
$thread->wait();
}
}, $this);
}
}
$my = new My();
$my->start();
/** send notification to the waiting thread **/
$my->synchronized(function ($thread) {
$thread->done = true;
$thread->notify();
}, $my);
var_dump($my->join());