amphp / cluster Goto Github PK

View Code? Open in Web Editor NEW

60.0 7.0 9.0 312 KB

Building multi-core network applications with PHP.

License: MIT License

PHP 100.00%

amphp cluster multi-processing sockets php async multi-core cluster-watcher parallel

cluster's People

Contributors

Stargazers

Watchers

Forkers

paulyd210 andreybolonin hectormonroy006 lyhiving vl25 shaogx stephenhonor azjezz meslubi2021

cluster's Issues

When will amphp/cluster be available for AMPHP 3?

Hi,

is there are roadmap on when amphp/cluster is available for AMPHP3 or what things are open?

Best Regards
mschop

How to bench this cluster?

wrk -t4 -c400 -d10s tcp://127.0.0.1:1337

But worker accept the requests

Compatibility with unix:// sockets?

When listening on unix sockets startup fails with

Worker 4 failed: Amp\Parallel\Sync\ContextPanicError: Uncaught Amp\Socket\SocketException in child process or thread with message "Could not create server tcp:///var/www/project/tmp/php.sock: [Error: #0] Failed to parse address "/var/www/project/tmp/php.sock"" and code "0"; use Amp\Parallel\Sync\ContextPanicError::getOriginalTrace() for the stack trace in the child process or thread in /var/www/project/vendor/amphp/parallel/lib/Sync/ExitFailure.php:46
Stack trace:
#0 /var/www/project/vendor/amphp/parallel/lib/Sync/ExitFailure.php(39): Amp\Parallel\Sync\ExitFailure->createException()
#1 /var/www/project/vendor/amphp/parallel/lib/Context/Process.php(232): Amp\Parallel\Sync\ExitFailure->getResult()
#2 [internal function]: Amp\Parallel\Context\Process->Amp\Parallel\Context\{closure}()
#3 /var/www/project/vendor/amphp/amp/lib/Coroutine.php(118): Generator->send()
#4 /var/www/project/vendor/amphp/amp/lib/Internal/Placeholder.php(149): Amp\Coroutine->Amp\{closure}()
#5 /var/www/project/vendor/amphp/amp/lib/Coroutine.php(123): Amp\Coroutine->resolve()
#6 /var/www/project/vendor/amphp/amp/lib/Internal/Placeholder.php(149): Amp\Coroutine->Amp\{closure}()
#7 /var/www/project/vendor/amphp/amp/lib/Deferred.php(53): Amp\Promise@anonymous->resolve()
#8 /var/www/project/vendor/amphp/byte-stream/lib/ResourceInputStream.php(101): Amp\Deferred->resolve()
#9 /var/www/project/vendor/amphp/amp/lib/Loop/NativeDriver.php(327): Amp\ByteStream\ResourceInputStream::Amp\ByteStream\{closure}()
#10 /var/www/project/vendor/amphp/amp/lib/Loop/NativeDriver.php(127): Amp\Loop\NativeDriver->selectStreams()
#11 /var/www/project/vendor/amphp/amp/lib/Loop/Driver.php(138): Amp\Loop\NativeDriver->dispatch()
#12 /var/www/project/vendor/amphp/amp/lib/Loop/Driver.php(72): Amp\Loop\Driver->tick()
#13 /var/www/project/vendor/amphp/amp/lib/Loop.php(95): Amp\Loop\Driver->run()
#14 /var/www/project/vendor/amphp/cluster/bin/cluster(250): Amp\Loop::run()
#15 /var/www/project/vendor/bin/cluster(117): include('...')
#16 {main} [] []

Stacktrace of exception thrown in workers is not logged

When an exception is thrown in worker, only the message and code of exception is logged.

A stacktrace is logged, but it is the stacktrace of Process and its state, not the actual stacktrace of exception.

Example:

[2021-02-02T14:15:37.836279+00:00] cluster-86033.error: Worker 2 failed: Amp\Parallel\Sync\ContextPanicError: Uncaught Error in child process or thread with message "Call to a member function get() on null" and code "0"; use Amp\Parallel\Sync\ContextPanicError::getOriginalTrace() for the stack trace in the child process or thread in /<project-path>/vendor/amphp/parallel/lib/Sync/ExitFailure.php:46

This exception causes my workers to die (not a bug with Cluster, bug from my code). But logged error does not help at all. I had to to the classical dump&die starting from worker script to find out where was the error.

Is this expected behavior? Am I missing something?

My worker code is pretty much like HTTP server example from this repo.

I'm willing to look into it, if this is an issue.

Ratchet and WAMP Server

Hi, any ideas how to use this with Ratchet for Wamp server in cluster arch?

SCM_RIGHTS issue for Windows in TransferSocket class

It seems that socket passing algorithm is not possible for the Windows platform, as Windows does not support constants like SCM_RIGHTS. If I understand correctly, is there another approach that needs to be used here? Or perhaps it's not necessary to pass the socket in this way at all?

OS: Windows 11
PHP: PHP 8.3.4 (cli)

Log:
The transfer socket threw an exception: Undefined constant "SCM_RIGHTS"::C:\work\ct\cluster\vendor\amphp\cluster\src\Internal\StreamResourceReceivePipe.php:72
Failed sending request to bind server socket: Sending on the channel failed. Did the context die?::C:\work\ct\cluster\vendor\amphp\cluster\src\ServerSocketPipeFactory.php:65
The transfer socket threw an exception: Undefined constant "SCM_RIGHTS"::C:\work\ct\cluster\vendor\amphp\cluster\src\Internal\StreamResourceReceivePipe.php:72
The transfer socket threw an exception: Undefined constant "SCM_RIGHTS"::C:\work\ct\cluster\vendor\amphp\cluster\src\Internal\StreamResourceReceivePipe.php:72

ClusterLogHandler cannot accept all log records

ClusterLogHandler cannot accept all log records.
e.g. in \Amp\Http\Server\Driver\Internal\AbstractHttpDriver::handleInternalServerError
you have

$this->logger->error(
  "...",
  [
    ...
    'method' => $request,
    ...
  ]
);

Then ClusterLogHandler cannot send such error context via channel because $request have closures in nested properties.
Maybe need to add some kind of filter there? e.g. skip unserializable items?

I can contribute on fix for that.
Just let me know how do you think such problem should be handled.

Thanks.

Lib Event error: event_base_loop can run on each event_base at once

Install library pecl install event
Options:

Enable internal debugging in Event [no] :
Enable sockets support in Event [yes] :
libevent installation prefix [/usr] :
Include libevent's pthreads library and enable thread safety support in Event [no] :
Include libevent protocol-specific functionality support including HTTP, DNS, and RPC [yes] : no
Include libevent OpenSSL support [yes] : no
PHP Namespace for all Event classes [no] :
openssl installation prefix [no] :

% pecl list
Installed packages, channel pecl.php.net:
=========================================
Package   Version State
event     3.0.2r1 stable
imagick   3.4.4   stable
memcached 3.1.5   stable
raphf     2.0.1   stable
xdebug    2.8.0   stable

% ./bin/cluster -w 1 examples/hello-world.php
PHP Warning:  EventBase::loop(): event_base_loop: reentrant invocation.  Only one event_base_loop can run on each event_base at once. in /project/vendor/amphp/amp/lib/Loop/EventDriver.php on line 255
PHP Stack trace:
PHP   1. {main}() /project/vendor/amphp/parallel/lib/Context/Internal/process-runner.php:0
PHP   2. Amp\Parallel\Context\Internal\{closure:/project/vendor/amphp/parallel/lib/Context/Internal/process-runner.php:40-123}() /project/vendor/amphp/parallel/lib/Context/Internal/process-runner.php:123
PHP   3. Amp\Promise\wait() /project/vendor/amphp/parallel/lib/Context/Internal/process-runner.php:105
PHP   4. Amp\Loop::run() /project/vendor/amphp/amp/lib/functions.php:229
PHP   5. Amp\Loop\EventDriver->run() /project/vendor/amphp/amp/lib/Loop.php:95
PHP   6. Amp\Loop\EventDriver->run() /project/vendor/amphp/amp/lib/Loop/EventDriver.php:206
PHP   7. Amp\Loop\EventDriver->tick() /project/vendor/amphp/amp/lib/Loop/Driver.php:72
PHP   8. Amp\Loop\EventDriver->dispatch() /project/vendor/amphp/amp/lib/Loop/Driver.php:138
PHP   9. EventBase->loop() /project/vendor/amphp/amp/lib/Loop/EventDriver.php:255
PHP  10. Amp\Loop\EventDriver->Amp\Loop\{closure:/project/vendor/amphp/amp/lib/Loop/EventDriver.php:59-79}() /project/vendor/amphp/amp/lib/Loop/EventDriver.php:255
PHP  11. Amp\Socket\DnsConnector::Amp\Socket\{closure:/project/vendor/amphp/socket/src/DnsConnector.php:84-86}() /project/vendor/amphp/amp/lib/Loop/EventDriver.php:63
PHP  12. Amp\Deferred->resolve() /project/vendor/amphp/socket/src/DnsConnector.php:85
PHP  13. {anonymous-class:/project/vendor/amphp/amp/lib/Deferred.php:22-27}->resolve() /project/vendor/amphp/amp/lib/Deferred.php:52
PHP  14. Amp\Promise\{closure:/project/vendor/amphp/amp/lib/functions.php:279-284}() /project/vendor/amphp/amp/lib/Internal/Placeholder.php:149
PHP  15. Amp\Deferred->resolve() /project/vendor/amphp/amp/lib/functions.php:282
PHP  16. {anonymous-class:/project/vendor/amphp/amp/lib/Deferred.php:22-27}->resolve() /project/vendor/amphp/amp/lib/Deferred.php:52
PHP  17. Amp\Internal\PrivatePromise->onResolve() /project/vendor/amphp/amp/lib/Internal/Placeholder.php:143
PHP  18. {anonymous-class:/project/vendor/amphp/amp/lib/Deferred.php:22-27}->onResolve() /project/vendor/amphp/amp/lib/Internal/PrivatePromise.php:23
PHP  19. Amp\Coroutine->Amp\{closure:/project/vendor/amphp/amp/lib/Coroutine.php:91-148}() /project/vendor/amphp/amp/lib/Internal/Placeholder.php:46
PHP  20. Amp\Coroutine->resolve() /project/vendor/amphp/amp/lib/Coroutine.php:123
PHP  21. Amp\Coroutine->Amp\{closure:/project/vendor/amphp/amp/lib/Coroutine.php:91-148}() /project/vendor/amphp/amp/lib/Internal/Placeholder.php:149
PHP  22. Generator->send() /project/vendor/amphp/amp/lib/Coroutine.php:118
PHP  23. Amp\Cluster\Internal\{closure:/project/vendor/amphp/cluster/src/Internal/cluster-runner.php:13-63}() /project/vendor/amphp/amp/lib/Coroutine.php:118
PHP  24. Amp\Cluster\Internal\{closure:/project/vendor/amphp/cluster/src/Internal/cluster-runner.php:57-60}() /project/vendor/amphp/cluster/src/Internal/cluster-runner.php:60
PHP  25. require() /project/vendor/amphp/cluster/src/Internal/cluster-runner.php:59
PHP  26. Amp\Loop::run() /project/vendor/amphp/cluster/examples/hello-world.php:49
PHP  27. Amp\Loop\EventDriver->run() /project/vendor/amphp/amp/lib/Loop.php:95
PHP  28. Amp\Loop\EventDriver->run() /project/vendor/amphp/amp/lib/Loop/EventDriver.php:206
PHP  29. Amp\Loop\EventDriver->tick() /project/vendor/amphp/amp/lib/Loop/Driver.php:72
PHP  30. Amp\Loop\EventDriver->dispatch() /project/vendor/amphp/amp/lib/Loop/Driver.php:138
PHP  31. EventBase->loop() /project/vendor/amphp/amp/lib/Loop/EventDriver.php:255

On start with php ./examples/hello-world.php works correctly.

Restart Workers

We have to implement worker restarts and graceful reloading, preferably with zero downtime.

Error in cluster

I see this in my logs regularly:

api_1             | [2020-05-31T22:02:33.704741-04:00] worker-17.DEBUG: Accept 10.121.1.2:41080 on 10.121.1.3:1337 #413 [] []
api_1             | [2020-05-31T22:02:33.708023-04:00] worker-17.DEBUG: GET http://local.pl.com:1337/secret/ HTTP/1.1 @ 10.121.1.2:41080 [] []
api_1             | [2020-05-31T22:02:34.612730-04:00] cluster-6.ERROR: STDERR from PID 17: src/unix/core.c:877: uv__io_stop: Assertion `loop->watchers[w->fd] == w' failed.

I'm using lib-uv in docker from php:7.4-cli image. When I disable the uv-extension the problem is gone.

For reference, here is my Dockerfile:

FROM php:7.4-cli

RUN apt-get update \
    && apt-get install libuv1-dev libpq-dev -y \
    && pecl install uv-beta raphf \
    && docker-php-ext-enable uv raphf \
    # we need to install pq in extra step, because raphf needs to be there already
    && pecl install pq \
    && docker-php-ext-enable pq

RUN echo date.timezone = "America/New_York" >  /usr/local/etc/php/conf.d/timezone.ini \
    && echo expose_php = Off > /usr/local/etc/php/conf.d/expose_php.ini

ADD . /var/www

WORKDIR /var/www

ENTRYPOINT ./vendor/bin/cluster ./bin/server.php -- --port 1337

configurable cluster worker shutdown timeout

currently, the context cluster worker has a hardcoded 5 seconds timeout on shutdown, i believe this aspect of the cluster should be configurable.

In the case of neu, the worker dispatches events when shutting down 1 2 3, allowing services to do clean up, which might take more than 5 secodns

Missing resume in StreamResourceSendPipe ?

Hello,

I think there is missing suspension resume in
\Amp\Cluster\Internal\StreamResourceSendPipe::__construct
in onWritable callback success path
after
if (!$transferSocket->sendSocket($export, $data)) {
...
}

Not sure how to reproduce it because in \Amp\Cluster\Internal\StreamResourceSendPipe::send
$this->transferSocket->sendSocket does not return false for simple testcase
(using parent.php, child.php from your 'Transferring Client Sockets' example and client.php that connect to exposed socket)
But if it return false suspension won't be resumed.

Thanks,
Sergey

Error running cluster: The provided socket has already been closed

tested on latest commit: 9dd11ec

I have ext-sockets installed on PHP 8.1.12 (Alpine Linux 3.15 - docker)

php bin/node.php - runs the webserver successfully

php vendor/bin/cluster bin/node.php gives me this output (original is a bit longer, I just put here the important bits):

[2022-11-01T19:37:37.286526+00:00] cluster-2426.error: Worker 111 (PID 2652) died unexpectedly: The process stopped responding, potentially due to a fatal error or calling exit, restarting... {"id":111,"pid":2652} []
^C[2022-11-01T19:37:37.289365+00:00] cluster-2426.info: Stopping cluster due to received signal: 2 [] []
[2022-11-01T19:37:37.295714+00:00] cluster-2426.info: Worker 3 terminated cleanly {"id":3,"pid":2435} []
[2022-11-01T19:37:37.295854+00:00] cluster-2426.info: Worker 113 terminated cleanly {"id":113,"pid":2654} []
[2022-11-01T19:37:37.295951+00:00] cluster-2426.info: Worker 114 terminated cleanly {"id":114,"pid":2656} []
[2022-11-01T19:37:37.296109+00:00] cluster-2426.info: Worker 115 terminated cleanly {"id":115,"pid":2658} []
[2022-11-01T19:37:37.296226+00:00] cluster-2426.info: Worker 112 terminated cleanly {"id":112,"pid":2648} []
[2022-11-01T19:37:37.296351+00:00] cluster-2426.info: Worker 116 terminated cleanly {"id":116,"pid":2660} []
[2022-11-01T19:37:37.303222+00:00] cluster-2426.info: Started worker with ID 117 {"id":117,"pid":2662} []
[2022-11-01T19:37:37.303406+00:00] cluster-2426.info: Worker 117 terminated cleanly {"id":117,"pid":2662} []

Fatal error: Uncaught Amp\Socket\SocketException: The provided socket has already been closed in /var/www/vendor/amphp/cluster/src/Internal/TransferSocket.php:24
Stack trace:
#0 /var/www/vendor/amphp/cluster/src/StreamResourceSendPipe.php(26): Amp\Cluster\Internal\TransferSocket->__construct(Object(Amp\Socket\ResourceSocket))
#1 /var/www/vendor/amphp/cluster/src/ClusterSocketServerProvider.php(46): Amp\Cluster\StreamResourceSendPipe->__construct(Object(Amp\Socket\ResourceSocket), Object(Amp\Serialization\NativeSerializer))
#2 /var/www/vendor/amphp/cluster/src/Watcher.php(148): Amp\Cluster\ClusterSocketServerProvider->provideFor(Object(Amp\Socket\ResourceSocket))
#3 /var/www/vendor/amphp/amp/src/functions.php(34): Amp\Cluster\Watcher->Amp\Cluster\{closure}()
#4 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/AbstractDriver.php(422): Amp\{closure}(NULL, NULL, Array)
#5 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/AbstractDriver.php(612): Revolt\EventLoop\Internal\AbstractDriver->invokeMicrotasks()
#6 [internal function]: Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#7 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/AbstractDriver.php(494): Fiber->resume()
#8 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/AbstractDriver.php(549): Revolt\EventLoop\Internal\AbstractDriver->invokeCallbacks()
#9 [internal function]: Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#10 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/AbstractDriver.php(83): Fiber->resume()
#11 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/DriverSuspension.php(95): Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#12 /var/www/vendor/amphp/amp/src/Future.php(247): Revolt\EventLoop\Internal\DriverSuspension->suspend()
#13 /var/www/vendor/amphp/cluster/src/Watcher.php(216): Amp\Future->await()
#14 /var/www/vendor/amphp/cluster/bin/cluster(254): Amp\Cluster\Watcher->join()
#15 /var/www/vendor/bin/cluster(120): include('/var/www/vendor...')
#16 {main}

Next Amp\Future\UnhandledFutureError: Unhandled future: Amp\Socket\SocketException: "The provided socket has already been closed"; Await the Future with Future::await() before the future is destroyed or use Future::ignore() to suppress this exception in /var/www/vendor/amphp/amp/src/Internal/FutureState.php:38
Stack trace:
#0 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/AbstractDriver.php(422): Amp\Internal\FutureState->__destruct()
#1 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/AbstractDriver.php(612): Revolt\EventLoop\Internal\AbstractDriver->invokeMicrotasks()
#2 [internal function]: Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#3 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/AbstractDriver.php(422): Fiber->throw(Object(Amp\Socket\SocketException))
#4 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/AbstractDriver.php(612): Revolt\EventLoop\Internal\AbstractDriver->invokeMicrotasks()
#5 [internal function]: Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#6 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/AbstractDriver.php(422): Fiber->resume(NULL)
#7 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/AbstractDriver.php(612): Revolt\EventLoop\Internal\AbstractDriver->invokeMicrotasks()
#8 [internal function]: Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#9 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/AbstractDriver.php(494): Fiber->resume()
#10 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/AbstractDriver.php(549): Revolt\EventLoop\Internal\AbstractDriver->invokeCallbacks()
#11 [internal function]: Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#12 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/AbstractDriver.php(83): Fiber->resume()
#13 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/DriverSuspension.php(95): Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#14 /var/www/vendor/amphp/amp/src/Future.php(247): Revolt\EventLoop\Internal\DriverSuspension->suspend()
#15 /var/www/vendor/amphp/cluster/src/Watcher.php(216): Amp\Future->await()
#16 /var/www/vendor/amphp/cluster/bin/cluster(254): Amp\Cluster\Watcher->join()
#17 /var/www/vendor/bin/cluster(120): include('/var/www/vendor...')
#18 {main}

Next Revolt\EventLoop\UncaughtThrowable: Uncaught Amp\Future\UnhandledFutureError thrown in event loop callback Amp\Internal\FutureState::Amp\Internal\{closure} defined in /var/www/vendor/amphp/amp/src/Internal/FutureState.php:39; use Revolt\EventLoop::setErrorHandler() to gracefully handle such exceptions: Unhandled future: Amp\Socket\SocketException: "The provided socket has already been closed"; Await the Future with Future::await() before the future is destroyed or use Future::ignore() to suppress this exception in /var/www/vendor/revolt/event-loop/src/EventLoop/UncaughtThrowable.php:11
Stack trace:
#0 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/AbstractDriver.php(397): Revolt\EventLoop\UncaughtThrowable::throwingCallback(Object(Closure), Object(Amp\Future\UnhandledFutureError))
#1 /var/www/vendor/revolt/event-loop/src/EventLoop/Internal/DriverSuspension.php(100): Revolt\EventLoop\Internal\AbstractDriver::Revolt\EventLoop\Internal\{closure}()
#2 /var/www/vendor/amphp/amp/src/Future.php(247): Revolt\EventLoop\Internal\DriverSuspension->suspend()
#3 /var/www/vendor/amphp/cluster/src/Watcher.php(216): Amp\Future->await()
#4 /var/www/vendor/amphp/cluster/bin/cluster(254): Amp\Cluster\Watcher->join()
#5 /var/www/vendor/bin/cluster(120): include('/var/www/vendor...')
#6 {main}
  thrown in /var/www/vendor/revolt/event-loop/src/EventLoop/UncaughtThrowable.php on line 11

Windows Support

Currently we have no guards around Loop::onSignal(), which will throw on Windows.