This extension seeks to provide an implementation of the Actor model in PHP. Due to the size and complexity of this project, I am making an early-stage release of it, in hopes that it will attract the help of other developers.
Relevant reading:
Quick details:
- Uses an N:M threading model
- Cooperative scheduling of actors
- Actors have an initial cost of ~500 bytes
Requirements:
- A ZTS version of PHP 7.2 (master branch is currently unsupported)
- An x86-64 Unix-based OS
- The Pthread library
Major goals:
- Stabilise things (ongoing)
- Implement supervision trees (in progress)
- Implement remote actors (to do)
- Implement internal non-blocking APIs (to do)
How you can help:
- Resolve any of the open issues of this repo
- Open new issues for: bugs, general feedback, support for a platform, ideas discussion, etc
This extension has been tested on OS X (Yosemite and Sierra), Centos 7 (64bit), and Ubuntu 14.04 and 16.04 (64bit).
Each actorised application will have its own actor system. This actor system is responsible for managing the actors within its system, along with configuring it.
Each actor has a single entry point: Actor::receive(). This method will be automatically invoked upon actor creation, where the actor will then be able to begin handling messages from its mailbox. Once the Actor::receive() method has finished executing, the actor will be destroyed. This means that in order to keep an actor alive, the Actor::receive() method needs to keep executing. This can be achieved by invoking the Actor::receiveBlock() method (perhaps in tandem with looping) to voluntarily interrupt the actor, placing it in an idle state where it will wait for new messages to arrive.
To shut down an actor system, the ActorSystem::shutdown() static method must be invoked (it can be called from anywhere in the application).
The following script demonstrates the basic flow of execution when using an actor:
<?phpusephactor\{ActorSystem, Actor, ActorRef};
$actorSystem = newActorSystem();
class Test extends Actor
{
// internal actor stateprivate$str;
publicfunction__construct(string$param)
{
$this->str = $param;
// send a message to itself$this->send('actor name', "{$param}{$param}");
}
// automatically invoked (at an arbitrary time after __construct)publicfunctionreceive()
{
var_dump($this->str); // string(5) "arg 1"$message = $this->receiveBlock(); // wait here for a new messagevar_dump($message); // string(11) "arg 1 arg 1"
ActorSystem::shutdown(); // shut down the actor system// end of receive() method - the actor will be destroyed (asynchronously)
}
}
// spawn the new actor - executes the actor's __construct() and receive() methodsnewActorRef(Test::class, ['arg 1'], 'actor name');<?phpnamespacePhactor;
finalclass ActorSystem
{
publicfunction__construct(int$threadCount = $coreCount + 10); // 10 additional threadspublicstaticfunctionshutdown(void) : void;
publicfunctionblock(void) : void; // should be ignored (used internally)
}
finalclass ActorRef
{
private$ref;
private$name = '';
publicfunction__construct(string$actorClassName[, array$ctorArgs[, string$actorName]]);
// for debugging only - do not use this for the sender (since it will be treated as the actor name)publicfunction getRef(void) : string;
publicfunction getName(void) : string;
// for retrieving a reference to an actor objectpublicstatic function fromActor(Actor $actor) : ActorRef;
}
abstractclass Actor
{
// the public API and entry point of an actorpublicabstractfunctionreceive(void) : void;
// send a message (asynchronously) to an actorprotectedfinalfunctionsend(ActorRef$toActorRef, mixed$message) : void;
protectedfinalfunctionsend(string$toActorName, mixed$message) : void;
// wait for a message (cooperatively yields the actor)protectedfinalfunctionreceiveBlock(void) : mixed;
}
finalclass Supervisor
{
// supervision strategypublicconstONE_FOR_ONE = 0;
publicfunction__construct(ActorRef|string$supervisor[, int$strategy = Supervisor::ONE_FOR_ONE]);
// Add a pre-existing actor to the group of supervised workerspublicfunction addWorker(ActorRef|string$worker) : void;
// Creates a new actor, links it to the supervisor, and return its actor reference.// This should be used if the actor's constructor may throw an exceptionpublicfunction newWorker(string$actorClass[, array$ctorArgs[, string$actorName]]) : ActorRef;
}All messages and actor constructor arguments (the second parameter of ActorRef::__construct and Supervisor::newWorker) will be serialised.
An anonymous actor sending a message to itself:
<?phpusephactor\{ActorSystem, Actor, ActorRef};
$actorSystem = newActorSystem();
class Test extends Actor
{
publicfunction__construct()
{
// since this is an anonymous actor, its reference must be fetched in// order to send a message to itself$actorRef = ActorRef::fromActor($this);
$this->send($actorRef, 123);
}
publicfunctionreceive()
{
var_dump($this->receiveBlock()); // int(123)
ActorSystem::shutdown();
}
}
// spawn a new anonymous actornewActorRef(Test::class);An actor sending messages to itself in a synchronous execution style:
<?phpusephactor\{ActorSystem, Actor, ActorRef};
$actorSystem = newActorSystem();
class Test extends Actor
{
private$messages = ['something 1', 'something 2', 'something 3', 'shutdown'];
publicfunction__construct(string$s)
{
$this->send('testing', $s);
}
publicfunctionreceive()
{
$i = 0;
do {
// block here, waiting for a new message to be received$message = $this->receiveBlock();
var_dump($message);
if ($message === 'shutdown') {
break;
}
// send a message to itself again$this->send('testing', $this->messages[$i++]);
} while (true);
// shut down the actor system - without this, the PHP process will not stop
ActorSystem::shutdown();
}
}
// spawn the new actornewActorRef(Test::class, ['something 0'], 'testing');Using a supervisor with the ONE_FOR_ONE supervision strategy to handle the crashing of an actor (by automatically restarting it):
<?phpusephactor\{ActorSystem, Actor, ActorRef, Supervisor};
$as = newActorSystem(1);
// the supervisorclass A extends Actor
{
publicfunctionreceive(){$this->receiveBlock();}
}
// the workerclass B extends Actor
{
privatestatic$i = 0;
publicfunction__construct()
{
if (self::$i === 0) {
++self::$i; // make it crash once herevar_dump('Crashing B in __construct()');
thrownewexception();
}
}
publicfunctionreceive()
{
if (self::$i === 1) {
++self::$i; // make it crash once herevar_dump('Crashing B in receive()');
thrownewexception();
}
$this->send('b', 1); // send itself a message to resume after the following interrupt$this->receiveBlock(); // interrupt and wait for the int(1) message to arriveif (self::$i === 2) {
++self::$i; // make it crash once herevar_dump('Crashing B in receive() again');
thrownewexception();
}
var_dump('Made it!');
ActorSystem::shutdown();
}
}
$a = newActorRef(A::class, [], 'a');
$s = newSupervisor($a);
$b = $s->newWorker(B::class, [], 'b');Note: whilst static variables have been used in the above example (to crash an actor a controlled number of times), static state is very unpredictable and should be avoided in such concurrent applications.
Whenever an actor is restarted, a new actor object will be created (so the __construct and receive methods will both be invoked again). The restarted actor will have the same reference and (if provided) the same name.
Because the actor could crash inside of its constructor, we created the new actor using the Supervisor::newWorker() method. This enables for an actor to be automatically restarted if it failed to construct itself (currently, the restart limit is hard coded to 5 - this will become configurable in future).