Worker::waitPayload virtually disables custom posix signal handling

[l-x]

At the moment the \Spiral\RoadRunner\Worker::waitPayload is a blocking stream operation which virtually disables asynchronous signal handling in my worker scripts.

We have the absolute necessity to maintain a TCP connection to an internal service using manual heartbeat requests, regardless of whether the worker loop is currently waiting for tasks or not. We currently achieve this via pcntl_alarm callbacks, but this does not seem to be possible in Roadrunner at the moment.

Is there any way to call custom code at regular (configurable) intervals while Worker::waitPayload is waiting for data frames?

[rustatian]

Hey @l-x 👋🏻
Might be it would be better to write a small custom plugin for RR and build it with Velox?

[rustatian]

@wolfy-j / @roxblnfk WDYT guys, do we have the capacity to implement/plan this feature in future?

[wolfy-j]

Hi, while we have a ticket we do not plan to take on this feature any time soon (maybe never). It is possible to make non blocking workers - like this example for RR1 - https://github.com/phpinnacle/amridge

But the scope of work is too large while it does not provide any substantial benefits for most of the users.

I can only advice you:

1. Make your own worker by forking ours. If your PR might work without any BC we might even merge it back.
2. Describe the core issue outside of the scope of the worker, maybe some TCP connection pool on RR end will be better for resource sharing and healthchecks.

[l-x]

Just to clarify, my problem is not the Worker blocking my application but the relay blocking the whole PHP VM while waiting for data. I am able to solve this with a slightly modified StreamRelay.

[l-x]

To whom it may concern, we solved the issue with the following simple wrapper for the PHP StreamRelay:

<?php

declare(strict_types=1);

namespace Derp;

use Spiral\Goridge\Frame;
use Spiral\Goridge\RelayInterface;
use Spiral\Goridge\StreamRelay;

use function stream_select;

class StreamRelayWrapper implements RelayInterface
{
    /**
     * @param resource $readStream
     * @param StreamRelay $relay
     */
    public function __construct(
        private $readStream,
        private readonly StreamRelay $relay,
    ) {
    }

    private function waitForData(): void
    {
        $null = null;

        while (true) {
            $readStreams = [$this->readStream];
            
            if (
                @stream_select(
                    $readStreams,
                    $null,
                    $null,
                    0,
                    200_000
                )
            ) {
                break;
            }
        }
    }

    public function waitFrame(): Frame
    {
        $this->waitForData();

        return $this->relay->waitFrame();
    }

    public function send(Frame $frame): void
    {
        $this->relay->send($frame);
    }

    public function hasFrame(): bool
    {
        return $this->relay->hasFrame();
    }
}

$relay = new StreamRelayWrapper(STDIN, new StreamRelay(STDIN, STDOUT));

[rustatian]

I think you won't need to use this wrapper anymore when I implement these additional features for the AMQP plugin, am I right?

[l-x]

Yeah, that's true :D