joopschilder/php-async
joopschilder/php-async
1
0
Fork 0
Code Issues 2 Pull Requests Releases 2 Wiki Activity

Add brackets, update README, refactored naming

Browse Source
This commit is contained in:
Joop Schilder 2019-12-08 01:10:25 +01:00
parent f29d1a0d7a
commit eef63af0b0
8 changed files with 210 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -1,6 +1,5 @@
/bin/ /bin/
/vendor/ /vendor/
composer.lock
/.idea/ /.idea/
/.phpstorm/ /.phpstorm/
/.vscode/ /.vscode/

116
README.md
View File

@ -1,48 +1,66 @@
# joopschilder/php-async # `php-async`
## Introduction
This package provides functions to run callables asynchronously in PHP. Return values are shared via System-V shared memory.<br/>
To use this package, you'll need PHP >= 7.0.0 with `ext-sysvshm` and `ext-pcntl`.<br/>
You should consider the state of this package to be <i>experimental</i>.<br/><br/>
<b>Note:</b> This package should not be used in a CGI environment.
The key that is used to access the block of shared memory is created based on the inode information of one of the source files.
This means that, whenever multiple instances (processes) from the same project source are created, they will try to use the same block of memory and collisions will occur.
I might swap the `ftok()` call for a random string generator somewhere down the road.<br/><br/>
<b>Note:</b> It is possible (but discouraged) to change the amount of available shared memory.
If you wish to do so, it's as simple as calling either `Runtime::_setSharedMemorySizeMB(<amount of MB>);` or `Runtime::_setSharedMemorySizeB(<amount of bytes>);`.<br/>
If you want to use 32MB for example, call `Runtime::_setSharedMemorySizeMB(32);`.<br/>
Be sure to make this call before using any of the asynchronous functionalities.
## Installation ## Installation
This package is available on <a href="https://packagist.org/packages/joopschilder/php-async">Packagist</a>
and can be installed using <a href="https://getcomposer.org/">Composer</a>:<br/> This package is available on [Packagist](https://packagist.org/packages/joopschilder/php-async) and can be installed using [Composer](https://getcomposer.org/):
```bash ```bash
$ composer require joopschilder/async-php $ composer require joopschilder/async-php
``` ```
It's also possible to manually add it to your `composer.json`: It's also possible to manually add it to your `composer.json`:
```json ```json
{ {
"require": { "require": {
"joopschilder/php-async": "dev-master" "joopschilder/php-async": "~1.0"
} }
} }
``` ```
## Usage ## What is this?
#### Functions
The library exposes three functions in the global namespace that provide indirect access to the class `Asynchronous`:
* `async(callable $function, ...$parameters)` to run something asynchronously, giving back a `Promise`;
* `async_wait_all()` to wait for all currently running jobs to finish;
* `async_reap_zombies()` to clean up any zombie processes during runtime if any exist;
#### Promises This package provides functions to run callables asynchronously in PHP. Return values are shared via System-V shared memory.
Whenever you call `async(...)`, a `Promise` instance is returned.<br/> To use this package, you need PHP >= 7 with `ext-sysvshm` and `ext-pcntl`.
You should consider the state of this package to be __highly experimental__.
___Note:__ This package should not be used in a CGI environment._
The key that is used to access the block of shared memory is created based on the inode information of one of the source files.
This means that, whenever multiple instances (processes) from the same project source are created, they will try to use the same block of memory and collisions will occur.
I might swap the `ftok()` call for a multi-instance supporting mechanism later on (feel free to do so yourself).
___Note:__ It is possible (but discouraged) to change the amount of available shared memory._
If you wish to do so, it's as simple as calling either `Runtime::setSharedMemorySizeMB(<amount of MB>)` or `Runtime::setSharedMemorySizeB(<amount of bytes>)`.
If you want to use 32MB for example, call `Runtime::setSharedMemorySizeMB(32)`.<br/>
Be sure to make this call before using any of the asynchronous functionalities.
## What is this not?
This is, as you probably guessed by now, not intended for use in a production environment.
I'm not saying you _can't_, I'm just saying you _shouldn't_.
The code is _not_ unit tested. It has been documented throughout though, so feel free to take a look.
## Usage
### Functions
The library exposes three functions in the global namespace that provide indirect access to the `Asynchronous` class:
* `async(callable $function, ...$parameters)` to run something asynchronously, returning a `Promise`;
* `async_wait_all()` to wait for all currently running jobs to finish;
* `async_cleanup()` to clean up any zombie processes during runtime if any exist;
### A `Promise`, you say?
`async(...)` returns an instance of `JoopSchilder\Asynchronous\Promise`.
A `Promise` is considered to be resolved when the function it belongs to returned a value or finished execution. A `Promise` is considered to be resolved when the function it belongs to returned a value or finished execution.
To block execution until a promise is resolved, simply call the `resolve()` method on the promise. To block execution until a promise is resolved, simply call the `resolve()` method on the promise.
It's possible to check whether the promise has been resolved in a non-blocking way by calling the `isResolved()` method.<br/> It's possible to check whether the promise has been resolved in a non-blocking way by calling the `isResolved()` method.
You can actually return anything that is serializable in PHP: objects, arrays, strings, you name it. You can actually return anything that is serializable in PHP: objects, arrays, strings, you name it.
```php
<?php
require_once __DIR__ . '/vendor/autoload.php';
```php
$promise = async(function() { $promise = async(function() {
sleep(random_int(1, 5)); sleep(random_int(1, 5));
return getmypid(); return getmypid();
@ -53,16 +71,15 @@ $promise = async(function() {
$promise->resolve(); $promise->resolve();
$pid = $promise->getValue(); $pid = $promise->getValue();
``` ```
The shutdown handler and destructors should take care of the cleanup.<br/> The shutdown handler and destructors should take care of the cleanup.
#### Asynchronous curl requests
... though you should probably look into curl multi handles for this: <a href="http://php.net/manual/en/function.curl-multi-init.php">curl_multi_init()</a>. ### Asynchronous curl requests
The only reason `curl` is used here is to provide an intuitive example.
If you really wanted to perform concurrent http requests you should look into either [`curl_multi_init`](http://php.net/manual/en/function.curl-multi-init.php) or just use [Guzzle](http://docs.guzzlephp.org/en/stable/).
```php ```php
<?php $job = function(string $url) {
require_once __DIR__ . '/vendor/autoload.php';
// Create the body for the process...
$process = function(string $url) {
$handle = curl_init($url); $handle = curl_init($url);
curl_setopt($handle, CURLOPT_FOLLOWLOCATION, 1); curl_setopt($handle, CURLOPT_FOLLOWLOCATION, 1);
curl_setopt($handle, CURLOPT_RETURNTRANSFER, 1); curl_setopt($handle, CURLOPT_RETURNTRANSFER, 1);
@ -71,30 +88,27 @@ $process = function(string $url) {
file_put_contents(uniqid('download_'), $response); file_put_contents(uniqid('download_'), $response);
}; };
// Define some urls we want to download... $urls = ['example.com', 'example2.com', 'some.other.domain'];
$urls = [ foreach($urls as $url) {
'example.com', async($job, $url);
'example2.com', }
'some.other.domain'
];
// And there we go.
foreach($urls as $url)
async($process, $url);
``` ```
That's all there is to it. That's all there is to it.
## Tips ## Tips
If you're on a UNIX system, you can make use of the tools `ipcs` and `ipcrm` to monitor and manage the shared memory blocks.<br/>
To track what's happening in real time, I like to use:<br/> If you're using a UNIX system, you can make use of the tools `ipcs` and `ipcrm` to monitor and manage System V shared memory blocks.
To see what's happening, you can use:
```bash ```bash
$ watch -n 1 "ipcs -m --human && ipcs -m -p && ipcs -m -t && ipcs -m -u" $ watch -n 1 "ipcs -m --human && ipcs -m -p && ipcs -m -t && ipcs -m -u"
``` ```
<br/>To clean all 'unused' shared memory blocks (they might remain resident in RAM if your program terminated unexpectedly):<br/>
To clean all 'unused' shared memory blocks (they might stay in RAM if your program terminated unexpectedly), run
```bash ```bash
$ ipcrm -a $ ipcrm -a
``` ```
## What's next? ## What's next?
- Improving stability Who really knows?
- Add an explaining diagram

39
composer.json
View File

@ -1,19 +1,26 @@
{ {
"name": "joopschilder/php-async", "name": "joopschilder/php-async",
"authors": [ "license": "MIT",
{ "authors": [
"name": "Joop Schilder", {
"email": "j.n.m.schilder@st.hanze.nl" "name": "Joop Schilder",
} "email": "jnmschilder@protonmail.com",
], "homepage": "https://joopschilder.nl/",
"autoload": { "role": "Developer"
"files": ["src/functions.php"],
"psr-4": {
"JoopSchilder\\Asynchronous\\": ["src"]
}
},
"require": {
"ext-pcntl": "*",
"ext-sysvshm": "*"
} }
],
"autoload": {
"files": [
"lib/functions.php"
],
"psr-4": {
"JoopSchilder\\Asynchronous\\": [
"src"
]
}
},
"require": {
"ext-pcntl": "*",
"ext-sysvshm": "*"
}
} }

20
composer.lock generated Normal file
View File

@ -0,0 +1,20 @@
{
"_readme": [
"This file locks the dependencies of your project to a known state",
"Read more about it at https://getcomposer.org/doc/01-basic-usage.md#installing-dependencies",
"This file is @generated automatically"
],
"content-hash": "b0b3e06a09fa46405754b75a4beba98c",
"packages": [],
"packages-dev": [],
"aliases": [],
"minimum-stability": "stable",
"stability-flags": [],
"prefer-stable": false,
"prefer-lowest": false,
"platform": {
"ext-pcntl": "*",
"ext-sysvshm": "*"
},
"platform-dev": []
}

6
src/functions.php → lib/functions.php
View File

@ -16,14 +16,14 @@ if (!function_exists('async')) {
} }
if (!function_exists('async_reap_zombies')) { if (!function_exists('async_cleanup')) {
/** /**
* *
*/ */
function async_reap_zombies() function async_cleanup()
{ {
Asynchronous::reap(); Asynchronous::cleanup();
} }
} }

137
src/Asynchronous.php
View File

@ -2,8 +2,11 @@
namespace JoopSchilder\Asynchronous; namespace JoopSchilder\Asynchronous;
use Throwable;
/** /**
* Class Asynchronous * Class Asynchronous
* @package JoopSchilder\Asynchronous
* Responsible for management of child processes and shared memory. * Responsible for management of child processes and shared memory.
*/ */
class Asynchronous class Asynchronous
@ -19,11 +22,38 @@ class Asynchronous
/** /**
* @param callable $function * Asynchronous constructor.
* @param mixed ...$parameters
* @return Promise|null;
*/ */
public static function run(callable $function, ...$parameters) private function __construct()
{
/*
* The reason we do this is for when the shm block
* already exists. We attach, remove, detach and reattach
* to ensure a clean state.
*/
$this->attachToSharedMemory();
}
/**
*
*/
public function __destruct()
{
if (Runtime::isChild()) {
return;
}
self::getInstance()->freeSharedMemoryBlock();
}
/**
* @param callable $function
* @param mixed ...$parameters
* @return Promise|null
*/
public static function run(callable $function, ...$parameters): ?Promise
{ {
/* /*
* Prepare for fork * Prepare for fork
@ -31,16 +61,10 @@ class Asynchronous
$instance = self::getInstance(); $instance = self::getInstance();
$promiseKey = Promise::generatePromiseKey(); $promiseKey = Promise::generatePromiseKey();
/*
* Fork the parent
*/
$pid = pcntl_fork(); $pid = pcntl_fork();
if (-1 === $pid) {
/*
* The fork failed. Instead of returning a promise, we return null.
*/
if ($pid == -1)
return null; return null;
}
/* /*
* Parent process. We keep track of the PID of the child process * Parent process. We keep track of the PID of the child process
@ -66,25 +90,25 @@ class Asynchronous
* the Promise to be able to resolve. * the Promise to be able to resolve.
*/ */
Runtime::markAsChild(); Runtime::markAsChild();
$instance->_attachToShm(); $instance->attachToSharedMemory();
try { try {
$response = call_user_func($function, ...$parameters); $response = call_user_func($function, ...$parameters);
shm_put_var($instance->shm, $promiseKey, $response ?? Promise::RESPONSE_NONE); shm_put_var($instance->shm, $promiseKey, $response ?? Promise::RESPONSE_NONE);
exit(0); exit(0);
} catch (Throwable $throwable) {
} catch (\Throwable $throwable) {
shm_put_var($instance->shm, $promiseKey, Promise::RESPONSE_ERROR); shm_put_var($instance->shm, $promiseKey, Promise::RESPONSE_ERROR);
exit(1); exit(1);
} }
} }
/** /**
* *
*/ */
public static function reap() public static function cleanup(): void
{ {
/* /*
* Iterate over all child process PIDs and check * Iterate over all child process PIDs and check
@ -93,58 +117,35 @@ class Asynchronous
$instance = self::getInstance(); $instance = self::getInstance();
foreach ($instance->children as $index => $pid) { foreach ($instance->children as $index => $pid) {
$response = pcntl_waitpid($pid, $status, WNOHANG); $response = pcntl_waitpid($pid, $status, WNOHANG);
if ($response === $pid) if ($response === $pid) {
unset($instance->children[$index]); unset($instance->children[$index]);
}
} }
} }
/**
*
*/
public static function waitForChildren()
{
self::getInstance()->_awaitChildren();
}
/** /**
* *
*/ */
public static function removeShmBlock() public static function waitForChildren(): void
{ {
self::getInstance()->_removeShmBlock(); self::getInstance()->awaitChildProcesses();
} }
/** /**
* @return int *
*/ */
public static function childCount() public static function removeShmBlock(): void
{ {
return count(self::getInstance()->children); self::getInstance()->freeSharedMemoryBlock();
} }
/*
* Private methods below
*/
/**
* Asynchronous constructor.
*/
private function __construct()
{
/*
* The reason we do this is for when the shm block
* already exists. We attach, remove, detach and reattach
* to ensure a clean state.
*/
$this->_attachToShm(); // Attach
}
/** /**
* @return $this * @return $this
*/ */
private function _awaitChildren() private function awaitChildProcesses(): self
{ {
/* /*
* Wait for the children to terminate * Wait for the children to terminate
@ -157,14 +158,12 @@ class Asynchronous
return $this; return $this;
} }
/** /**
* @return $this * @return $this
*/ */
private function _removeShmBlock() private function freeSharedMemoryBlock(): self
{ {
/*
* Detach from the shared memory block
*/
if (is_resource($this->shm)) { if (is_resource($this->shm)) {
shm_remove($this->shm); shm_remove($this->shm);
shm_detach($this->shm); shm_detach($this->shm);
@ -177,7 +176,7 @@ class Asynchronous
/** /**
* @return $this * @return $this
*/ */
private function _attachToShm() private function attachToSharedMemory(): self
{ {
$this->shm = shm_attach(Runtime::getSharedMemoryKey(), Runtime::getSharedMemorySize()); $this->shm = shm_attach(Runtime::getSharedMemoryKey(), Runtime::getSharedMemorySize());
@ -188,14 +187,9 @@ class Asynchronous
/** /**
* @return Asynchronous * @return Asynchronous
*/ */
private static function getInstance() private static function getInstance(): self
{ {
if (is_null(self::$instance)) { if (is_null(self::$instance)) {
/*
* This is executed once during runtime;
* when a functionality from this class
* is used for the first time.
*/
self::$instance = new static(); self::$instance = new static();
self::registerHandlers(); self::registerHandlers();
} }
@ -207,32 +201,17 @@ class Asynchronous
/** /**
* *
*/ */
private static function registerHandlers() private static function registerHandlers(): void
{ {
$instance = self::getInstance(); $instance = self::getInstance();
/*
* The shutdown handler
*/
register_shutdown_function(function () use (&$instance) { register_shutdown_function(function () use (&$instance) {
if (Runtime::isChild()) if (Runtime::isChild()) {
return; return;
}
$instance->_awaitChildren(); $instance->awaitChildProcesses();
$instance->_removeShmBlock(); $instance->freeSharedMemoryBlock();
}); });
} }
/**
*
*/
public function __destruct()
{
if (Runtime::isChild())
return;
$instance = self::getInstance();
$instance->_removeShmBlock();
}
} }

34
src/Promise.php
View File

@ -2,6 +2,10 @@
namespace JoopSchilder\Asynchronous; namespace JoopSchilder\Asynchronous;
/**
* Class Promise
* @package JoopSchilder\Asynchronous
*/
class Promise class Promise
{ {
/* /*
@ -43,8 +47,9 @@ class Promise
* shm keys). * shm keys).
*/ */
self::$generatedKey++; self::$generatedKey++;
if (self::$generatedKey > 99999999) if (self::$generatedKey > 99999999) {
self::$generatedKey = 0; self::$generatedKey = 0;
}
return $promiseKey; return $promiseKey;
} }
@ -65,7 +70,7 @@ class Promise
/** /**
* @return bool * @return bool
*/ */
public function isResolved() public function isResolved(): bool
{ {
$this->attempt(); $this->attempt();
@ -76,7 +81,7 @@ class Promise
/** /**
* @return bool * @return bool
*/ */
public function isVoid() public function isVoid(): bool
{ {
return $this->getValue() === self::RESPONSE_NONE; return $this->getValue() === self::RESPONSE_NONE;
} }
@ -85,7 +90,7 @@ class Promise
/** /**
* @return bool * @return bool
*/ */
public function isError() public function isError(): bool
{ {
return $this->getValue() === self::RESPONSE_ERROR; return $this->getValue() === self::RESPONSE_ERROR;
} }
@ -103,14 +108,15 @@ class Promise
/** /**
* @return $this * @return $this
*/ */
public function resolve() public function resolve(): self
{ {
/* /*
* Actually block execution until a value is written to * Actually block execution until a value is written to
* the expected memory location of this Promise. * the expected memory location of this Promise.
*/ */
while (!$this->isResolved()) while (!$this->isResolved()) {
usleep(1000); usleep(50);
}
return $this; return $this;
} }
@ -128,24 +134,28 @@ class Promise
* garbage collector has noticed that there are no more * garbage collector has noticed that there are no more
* references to this Promise instance. * references to this Promise instance.
*/ */
if (Runtime::isChild()) if (Runtime::isChild()) {
return; return;
}
if (shm_has_var($this->shm, $this->key)) if (shm_has_var($this->shm, $this->key)) {
shm_remove_var($this->shm, $this->key); shm_remove_var($this->shm, $this->key);
}
shm_detach($this->shm); shm_detach($this->shm);
} }
/** /**
* @return $this * @return $this
*/ */
private function attempt() private function attempt(): self
{ {
if (shm_has_var($this->shm, $this->key)) if (shm_has_var($this->shm, $this->key)) {
$this->value = shm_get_var($this->shm, $this->key); $this->value = shm_get_var($this->shm, $this->key);
}
return $this; return $this;
} }
} }

34
src/Runtime.php
View File

@ -1,6 +1,5 @@
<?php <?php
namespace JoopSchilder\Asynchronous; namespace JoopSchilder\Asynchronous;
/** /**
@ -34,15 +33,10 @@ class Runtime
} }
/*
* Public
*/
/** /**
* @return int * @return int
*/ */
public static function getSharedMemorySize() public static function getSharedMemorySize(): int
{ {
return self::$sharedMemSize; return self::$sharedMemSize;
} }
@ -51,14 +45,14 @@ class Runtime
/** /**
* @return int * @return int
*/ */
public static function getSharedMemoryKey() public static function getSharedMemoryKey(): int
{ {
/* /*
* Use the filename as an identifier to create the * Use the filename as an identifier to create the System V IPC key.
* System V IPC key.
*/ */
if (is_null(self::$sharedMemKey)) if (is_null(self::$sharedMemKey)) {
self::$sharedMemKey = ftok(__FILE__, 't'); self::$sharedMemKey = ftok(__FILE__, 't');
}
return self::$sharedMemKey; return self::$sharedMemKey;
} }
@ -67,7 +61,7 @@ class Runtime
/** /**
* @return bool * @return bool
*/ */
public static function isChild() public static function isChild(): bool
{ {
return !self::$inParentRuntime; return !self::$inParentRuntime;
} }
@ -78,30 +72,30 @@ class Runtime
* To be used by internal classes. * To be used by internal classes.
*/ */
/** /**
* @param int $size_mb * @param int $sizeMegaBytes
*/ */
public static function _setSharedMemorySizeMB(int $size_mb) public static function setSharedMemorySizeMB(int $sizeMegaBytes): void
{ {
self::$sharedMemSize = abs($size_mb) * (1024 ** 2); self::$sharedMemSize = abs($sizeMegaBytes) * (1024 ** 2);
} }
/** /**
* @param int $size_b * @param int $sizeBytes
*/ */
public static function _setSharedMemorySizeB(int $size_b) public static function setSharedMemorySizeB(int $sizeBytes): void
{ {
self::$sharedMemSize = $size_b; self::$sharedMemSize = $sizeBytes;
} }
/** /**
* *
*/ */
public static function markAsChild() public static function markAsChild(): void
{ {
self::$inParentRuntime = false; self::$inParentRuntime = false;
} }
} }