InvalidWatcherError in ResourceOutputStream

[kelunik]

I'm getting the following exception while having a look at amphp/socket#32:

PHP Fatal error:  Uncaught Amp\Loop\InvalidWatcherError: Cannot enable an invalid watcher identifier: 'byb' in /home/kelunik/GitHub/amphp/socket/vendor/amphp/amp/lib/Loop/Driver.php:369
Stack trace:
#0 /home/kelunik/GitHub/amphp/socket/vendor/amphp/amp/lib/Loop.php(210): Amp\Loop\Driver->enable('byb')
#1 /home/kelunik/GitHub/amphp/socket/vendor/amphp/byte-stream/lib/ResourceOutputStream.php(188): Amp\Loop::enable('byb')
#2 /home/kelunik/GitHub/amphp/socket/vendor/amphp/byte-stream/lib/ResourceOutputStream.php(130): Amp\ByteStream\ResourceOutputStream->send('test', false)
#3 /home/kelunik/GitHub/amphp/socket/lib/Socket.php(67): Amp\ByteStream\ResourceOutputStream->write('test')
#4 /home/kelunik/GitHub/amphp/socket/test.php(35): Amp\Socket\Socket->write('test')
#5 /home/kelunik/GitHub/amphp/socket/vendor/amphp/amp/lib/Loop/NativeDriver.php(101): {closure}('byc', NULL)
#6 /home/kelunik/GitHub/amphp/socket/vendor/amphp/amp/lib/Loop/Driver.php(130): Amp\Loop\NativeDriver->dispatch(true)
#7 /home/kelunik/GitHub/amphp/socket/vendor/a in /home/kelunik/GitHub/amphp/socket/vendor/amphp/amp/lib/Loop/Driver.php on line 369

server.php

<?php

require __DIR__ . "/vendor/autoload.php";

use Amp\Socket\ServerSocket;
use function Amp\asyncCoroutine;

\Amp\Loop::run(function () {
    \Amp\Loop::repeat(5 * 1000, function () {
        echo "current RAM: " . round((memory_get_usage() / 1024 / 1024), 2) . "MB, pick RAM: " . round((memory_get_peak_usage() / 1024 / 1024), 2) . "MB\n";
    });

    $clientHandler = asyncCoroutine(function (ServerSocket $client) {
        echo "client connected" . PHP_EOL;

        while (($chunk = yield $client->read()) !== null) {
            echo "client data -> " . strlen($chunk) . PHP_EOL;
        }

        echo "client disonnected" . PHP_EOL;
    });

    $server = \Amp\Socket\listen("127.0.0.1:12001");

    while ($client = yield $server->accept()) {
        // $clientHandler($client);
    }
});

client.php

<?php

require __DIR__ . "/vendor/autoload.php";

\Amp\Loop::run(function () {
    echo "Driver -> " . get_class(\Amp\Loop::get()) . "\n";

    \Amp\Loop::repeat(5 * 1000, function () {
        echo "current RAM: " . round((memory_get_usage() / 1024 / 1024), 2) . "MB, pick RAM: " . round((memory_get_peak_usage() / 1024 / 1024), 2) . "MB\n";
    });

    for ($i = 0; $i < 1000; $i++) {
        client($i);
    }
});

/**
 * @param $i
 *
 * @return \Amp\Promise<\Amp\Socket\ClientSocket>
 */
function client($i)
{
    return \Amp\call(function () use ($i) {
        /**
         * @var $client \Amp\Socket\ClientSocket
         */
        $client = yield \Amp\Socket\connect("127.0.0.1:12001");

        \Amp\Loop::repeat(1000, function () use ($client, $i) {
            $data = "test";

            echo "write $i data -> " . strlen($data) . PHP_EOL;

            $client->write($data);
        });

        return $client;
    });
}

[ghost]

confirmed, i get the same

[bwoebi]

There are two flawed cases as far as I see...

We do throw an exception inside send() after:

• close() has been called
• end() has been called and the last message has been successfully sent

We do not throw an exception inside send() after:

• end() has been called, but the last message has not been sent yet => should cause a watcher on a closed resource (Bad. May corrupt event loop as well or (typically) trigger an assertion failure: $written !== false).
• fwrite() returns 0 while being the the first call to it within that onWritable loop. The promise is failed and the watcher cancelled. => Will trigger an InvalidWatcherError (this bug)

What is the expected behavior in the last two cases?

For the first case I think we can just throw \Error for trying to write to a self-end()ed stream.

For the latter case we should immediately return new Failure(new StreamException("Failed to write to socket: read watcher already closed due to previous write failure")) (not throw!).