A non-blocking stream abstraction for PHP based on Amp.

I encountered hanging feof resulting in StreamException("The stream was closed by the peer"). This happens when doing requests in Kubernetes to Facebook.

Omitting the feof check "fixes" it.

A workaround (and better code) would be to not check on @feof before the write but setting an error handler for the fwrite and cathing warnings and converting them to stream exceptions.

Example:

                    $eh = set_error_handler([$this, 'onSocketWriteError']);
                    
                    if ($chunkSize) {
                        $written = \fwrite($stream, $data, $chunkSize);
                    } else {
                        $written = \fwrite($stream, $data);
                    }

                    set_error_handler($eh);

Since a4739c8a6dd96d772cfe49a95d5dfda8c7417595, following hangs with no output

<?php

use Amp\ByteStream\ResourceInputStream;
use Amp\ByteStream\ResourceOutputStream;
use Amp\Loop;

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

$writeCoRoutine2 = \Amp\asyncCoroutine(function() {
    $middleReadStream = new ResourceInputStream(fopen('middle', 'r'));

    while (true) {
        echo yield $middleReadStream->read();
    }
});

Loop::run(function () use ($writeCoRoutine2) {
    \Amp\ByteStream\pipe(
        new ResourceInputStream(fopen('/home/g9735/Downloads/somefile.avi', 'r')),
        new ResourceOutputStream(fopen('middle', 'w'))
    );

    $writeCoRoutine2();
});

as soon as read() in $writeCoRoutine2 is executed, event loop will never return to pipe

Fixes #51

I tried hard to write a test for simulating such write failure, but any trick I tried, stream_select will not work with those. Another option would be to use unqualified fwrite call, but I don't think that would be approved.

https://github.com/amphp/byte-stream/blob/6d6c89f58c213e600e2ab5e3b1678fb61333eeb7/lib/ResourceOutputStream.php#L70

I currently debugged an issue in IpcLogger of aerys. It hang with 100% CPU as the socket was not detected as dead. In my setup @fwrite does NOT populate error_get_last for some reason. Furthermore it also was not populated if the registered error handler returned 'true'.

The only really reliable solution (in IpcLogger) was like that:

        $eh = set_error_handler([$this, 'onDeadIpcSock']);
        $bytes = fwrite($this->ipcSock, $this->writeBuffer);
        set_error_handler($eh);

If it affects that code then I am pretty sure it also affects the byte-stream code, right? Not sure if this is a bug in PHP because in other places, the @ / error_reporting(0) DID populate the error_get_last

Any other idea to catch fwrite errors in a better way?

I'm dealing with situation that following line https://github.com/amphp/byte-stream/blob/6bbfcb6f47e92577e739586ba0c87e867be70a23/lib/ResourceOutputStream.php#L95 throws exception, but doesn't expose information what was written/what is left to write, which makes it very difficult to handle such situation in case of incomplete chunk write. My aim is handling gracefully such exception and re-writing this data to different stream.

I suggest to create custom exception which carries $data it was not succeeded to write.

I need to depend on ResourceOutputStream, because I rely on methods missing from interface. And I need to mock it for my tests. Hence I am proposing to drop hard final and replace it with annotation based final, which is same strategy Symfony follows. This gives you best from both worlds - you are still not obligated to keep an eye for inheritance based BC issues, but now people can mock it

The assertion "Trying to write on a previously fclose()'d resource. Do NOT manually fclose() resources the loop still has a reference to." happens to me even if I have not a single fclose(), stream_XXX() or socket_XXX() call in my code.

My client and server thin wrappers around Server::listen() and connect() are running with Docker containers on Windows. So it might be related to docker networking internals, breaking sockets under some circumstances.

It happens occassionaly. I spent a whole day to find steps to reproduce it with no luck.

Anyway, since the case does really happen sometimes in runtime, I think it would be better to handle it and prevent the infinite loop.

Fixes #51

Test script below. It assumes you execute it in different mount than your /tmp/. Feel free to modify and use it to copy some actual files instead of str_repeat and change it so it does not fill up both disks. Ideally run from tmpfs.

<?php

use Amp\ByteStream\CannotWriteChunkException;
use Amp\ByteStream\ResourceOutputStream;
use Amp\Loop;

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

Loop::run(function () {
    $chunkSize = null;
    $out = new ResourceOutputStream(\fopen('foo', 'wb'), $chunkSize);
    $fallback = new ResourceOutputStream(\fopen('/tmp/bar', 'wb'), $chunkSize);
    $str = str_repeat('a', 8192);
    $written = 0;

    try {
	    while (($size = yield $out->write($str)) !== null) {
	        $written += $size;
	    }
    } catch (CannotWriteChunkException $exception) {
        $written += $exception->getFailingPositionInChunk();
        $written += yield $fallback->write(substr($str, $exception->getFailingPositionInChunk()));
        $out = $fallback;
	}

	echo 'written '.$written.PHP_EOL.'file sizes: '.(filesize('foo') + filesize('/tmp/bar'));
});

I don't see a good way how to write useful phpunit test for this, gave a lot of tries. Closest I could come up with is

public function testCannotWriteChunkException()
{
    $server = \stream_socket_server('localhost:8111', $errno, $errstr, STREAM_SERVER_BIND | STREAM_SERVER_LISTEN);

    $watcherId = Loop::onReadable($server, static function () use ($server, &$watcherId) {
        fread(\stream_socket_accept($server, 0), 1);
        Loop::cancel($watcherId);
    });

    $stream = new ResourceOutputStream(stream_socket_client('localhost:8111'));

    self::expectException(CannotWriteChunkException::class);
    wait($stream->write(str_repeat('-', 8 ** 7)));
}

But even that works only on MacOS and Windows, not Linux

kelunik@kelunik ❯ ~/GitHub/amphp/artax ❯ 10:24:18 ❯ master
$ php examples/1-get-request.php
HTTP/1.1 200 OK

{
  "user-agent": "Mozilla/5.0 (compatible; Artax)"
}

PHP Warning:  feof(): supplied resource is not a valid stream resource in /home/kelunik/GitHub/amphp/artax/vendor/amphp/byte-stream/lib/ResourceInputStream.php on line 60

I guess we can just suppress that one?

I see this in my logs:

Amp\ByteStream\StreamException Failed to write to stream after multiple attempts; fwrite(): send of 7302 bytes failed with errno=32 Broken pipe

This happens like every few seconds, I can try to provide a reproducible script, but that may take some time (currently on a business trip).

But I thought maybe you have an idea what to check.

php://temp and php://memory streams result in 100% CPU usage. It's probably due to them not being actual streams that can be used with stream_select, so there should probably be a test for that.

https://gist.github.com/umbri/2a3e558d487b5e6756768547e660f95a

<?php

use Amp\ByteStream\ResourceInputStream;
use Amp\ByteStream\ResourceOutputStream;
use Amp\Loop;

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

Loop::run(function () {
    $memory = fopen("php://temp", "r+");

    $stdin = new ResourceInputStream($memory);
    $stdout = new ResourceOutputStream($memory);

    Loop::delay(1000, function () use ($stdout) {
        echo "write" . PHP_EOL;

        $stdout->write("Hello, World!");
    });

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

I use Ridge V2 for RabbitMQ. When I upgrade the package from version 12 to version v2.0.0-beta.14, message processing on the production server is greatly reduced (see point 1 in the figure). When I return version v2.0.0-beta.12 (see point 2 in the figure), the processing speed returns to the required value.

When trying to do a simple HTTP request with HttpClient, I invariable receive a DnsException thrown by Amp\Dns\Rfc1035StubResolver:197, which is part of a MultiReasonException containing two such exceptions, whose previous exception is an AssertionError from byte-stream\lib\ResourceInputStream.php:135 where an assertion containing the following message fails:

Trying to read from a previously fclose()'d resource. Do NOT manually fclose() resources the loop still has a reference to.

Libraries

amphp/amp                             v2.5.2
amphp/beanstalk                       v0.3.2
amphp/byte-stream                     v1.8.1
amphp/cache                           v1.4.0
amphp/dns                             v1.2.3
amphp/hpack                           v3.1.0
amphp/http                            v1.6.3
amphp/http-client                     v4.5.5
amphp/http-client-cookies             v1.1.0
amphp/parallel                        v1.4.0
amphp/parser                          v1.0.0
amphp/postgres                        v1.3.3
amphp/process                         v1.1.1
amphp/serialization                   v1.0.0
amphp/socket                          v1.1.3
amphp/sql                             v1.0.1
amphp/sql-common                      v1.1.2
amphp/sync                            v1.4.0
amphp/uri                             v0.1.4
amphp/windows-registry                v0.3.3

Hi. I think it's would be great to have possibility to read chunks with specified length since commonly dealing with binary protocols looks like

$lengthBytes = fread($fd, 4); // read Int32 Length
$length = unpack('Llength', $lengthBytes)['length'];
$packet = fread($fd, $length);

Not sure about possibility to make InputStream behave like public function read(?int $length = null): Promise;, but maybe some buffering class similar to LineReader would be useful, for example:

class LengthReader
{
    public function __construct(InputStream $inputStream);

    public function readLine(int $length): Promise;
}