Exception "ProcessException: Received 0 of 5 expected bytes" when reading from stdout out of time

[PhantomArt]

I've been experimenting with Amp\Process to better understand how it works. I launched several processes, connecting them stdout -> stdin, simulated various situations in order to understand where errors could occur and how to handle them correctly, including so that there were no zombie processes left. I have encountered one problem, the nature of which is not clear to me.

Important: I am using the Windows 10 operating system and what I am working on will run on the same operating system.

I will give some synthetic examples. Most of them work correctly, but one works unexpectedly. In all examples, I will use the ImageMagick command to convert a PNG image to a JPEG.

Example 1 (works well):

\Amp\Process\Process::start([
  'magick',
  'image.png', // input file
  '-quality', '100',
  'JPEG:image.jpg' // output file
]);

\Revolt\EventLoop::run();

Now I modify this example to pass the input image to stdin.

Example 2 (works well):

$content = \file_get_contents('image.png');

$process = \Amp\Process\Process::start([
  'magick',
  '-', // input stdin
  '-quality', '100',
  'JPEG:image.jpg' // output file
]);

$process->getStdin()->write($content);
$process->getStdin()->end();

\Revolt\EventLoop::run();

Next, I change the example so that the output image is sent to stdout.

Example 3 (works well):

$content = \file_get_contents('image.png');

$process = \Amp\Process\Process::start([
  'magick',
  '-', // input stdin
  '-quality', '100',
  'JPEG:-' // output stdout
]);

$process->getStdin()->write($content);
$process->getStdin()->end();

$content = \Amp\ByteStream\buffer($process->getStdout());
\file_put_contents('image.jpg', $content);

\Revolt\EventLoop::run();

Now I remove the code that reads from stdout. I expect the script to hang forever on the \Revolt\EventLoop::run(); call. I realize this is stupid code, but it demonstrates the problem.

Example 4 (not working correctly):

$content = \file_get_contents('image.png');

$process = \Amp\Process\Process::start([
  'magick',
  '-', // input stdin
  '-quality', '100',
  'JPEG:-' // output stdout
]);

$process->getStdin()->write($content);
$process->getStdin()->end();

\Revolt\EventLoop::run();

The script does hang at the line \Revolt\EventLoop::run();, but every time after exactly 120 seconds, I get an exception:

Uncaught Amp\Process\ProcessException: Received 0 of 5 expected bytes in C:\test\vendor\amphp\process\src\Internal\Windows\SocketConnector.php:245

The exception text is always identical, it reports the expected 5 bytes.

Then I additionally remove the code that sends the input image to stdin.

Example 5 (works well):

\Amp\Process\Process::start([
  'magick',
  '-', // input stdin
  '-quality', '100',
  'JPEG:-' // output stdout
]);

\Revolt\EventLoop::run();

In this case, the script hangs indefinitely, as expected (perhaps not indefinitely, but certainly much longer than 120 seconds).

Then I try again to specify the input file as the command argument.

Example 6 (works well):

\Amp\Process\Process::start([
  'magick',
  'image.png', // input file
  '-quality', '100',
  'JPEG:-' // output stdout
]);

\Revolt\EventLoop::run();

In this example, the script still hangs forever, as expected.

In each of the examples I tried replacing \Revolt\EventLoop::run(); with $process->join();, this did not affect the results in any way.

Below I provide the full stack trace of the exception from example 4, but with $process->join();:

PHP Fatal error:  Uncaught Amp\Process\ProcessException: Received 0 of 5 expected bytes in C:\test\vendor\amphp\process\src\Internal\Windows\SocketConnector.php:245
Stack trace:
#0 C:\test\vendor\amphp\process\src\Internal\Windows\SocketConnector.php(219): Amp\Process\Internal\Windows\SocketConnector->read(Object(Amp\ByteStream\ReadableResourceStream), 5)
#1 C:\test\vendor\amphp\process\src\Internal\Windows\SocketConnector.php(102): Amp\Process\Internal\Windows\SocketConnector->readExitCode(Object(Amp\ByteStream\ReadableResourceStream))
#2 C:\test\vendor\amphp\amp\src\functions.php(34): Amp\Process\Internal\Windows\SocketConnector->Amp\Process\Internal\Windows\{closure}()
#3 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\AbstractDriver.php(425): Amp\{closure}(NULL, NULL, Array)
#4 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\AbstractDriver.php(616): Revolt\EventLoop\Internal\AbstractDriver->invokeMicrotasks()
#5 [internal function]: Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#6 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\DriverSuspension.php(64): Fiber->resume(NULL)
#7 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\AbstractDriver.php(425): Revolt\EventLoop\Internal\DriverSuspension::Revolt\EventLoop\Internal\{closure}()
#8 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\AbstractDriver.php(616): Revolt\EventLoop\Internal\AbstractDriver->invokeMicrotasks()
#9 [internal function]: Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#10 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\AbstractDriver.php(498): Fiber->start()
#11 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\AbstractDriver.php(553): Revolt\EventLoop\Internal\AbstractDriver->invokeCallbacks()
#12 [internal function]: Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#13 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\AbstractDriver.php(94): Fiber->resume()
#14 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\DriverSuspension.php(117): Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#15 C:\test\vendor\amphp\amp\src\Future.php(251): Revolt\EventLoop\Internal\DriverSuspension->suspend()
#16 C:\test\vendor\amphp\process\src\Internal\Windows\WindowsRunner.php(128): Amp\Future->await(NULL)
#17 C:\test\vendor\amphp\process\src\Process.php(123): Amp\Process\Internal\Windows\WindowsRunner->join(Object(Amp\Process\Internal\Windows\WindowsHandle), NULL)
#18 C:\test\test.php(19): Amp\Process\Process->join()
#19 {main}

Next Amp\Future\UnhandledFutureError: Unhandled future: Amp\Process\ProcessException: "Received 0 of 5 expected bytes"; Await the Future with Future::await() before the future is destroyed or use Future::ignore() to suppress this exception. The future has been created at #0 C:\test\vendor\amphp\amp\src\functions.php:40 Amp\Internal\FutureState->__construct()
#1 C:\test\vendor\amphp\process\src\Internal\Windows\SocketConnector.php:100 Amp\async()
#2 C:\test\vendor\amphp\process\src\Internal\Windows\WindowsRunner.php:101 Amp\Process\Internal\Windows\SocketConnector->connectPipes()
#3 C:\test\vendor\amphp\process\src\Process.php:80 Amp\Process\Internal\Windows\WindowsRunner->start()
#4 C:\test\test.php:9 Amp\Process\Process::start() in C:\test\vendor\amphp\amp\src\Internal\FutureState.php:53
Stack trace:
#0 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\AbstractDriver.php(425): Amp\Internal\FutureState->__destruct()
#1 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\AbstractDriver.php(616): Revolt\EventLoop\Internal\AbstractDriver->invokeMicrotasks()
#2 [internal function]: Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#3 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\DriverSuspension.php(64): Fiber->resume(NULL)
#4 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\AbstractDriver.php(425): Revolt\EventLoop\Internal\DriverSuspension::Revolt\EventLoop\Internal\{closure}()
#5 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\AbstractDriver.php(616): Revolt\EventLoop\Internal\AbstractDriver->invokeMicrotasks()
#6 [internal function]: Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#7 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\AbstractDriver.php(498): Fiber->start()
#8 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\AbstractDriver.php(553): Revolt\EventLoop\Internal\AbstractDriver->invokeCallbacks()
#9 [internal function]: Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#10 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\AbstractDriver.php(94): Fiber->resume()
#11 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\DriverSuspension.php(117): Revolt\EventLoop\Internal\AbstractDriver->Revolt\EventLoop\Internal\{closure}()
#12 C:\test\vendor\amphp\amp\src\Future.php(251): Revolt\EventLoop\Internal\DriverSuspension->suspend()
#13 C:\test\vendor\amphp\process\src\Internal\Windows\WindowsRunner.php(128): Amp\Future->await(NULL)
#14 C:\test\vendor\amphp\process\src\Process.php(123): Amp\Process\Internal\Windows\WindowsRunner->join(Object(Amp\Process\Internal\Windows\WindowsHandle), NULL)
#15 C:\test\test.php(19): Amp\Process\Process->join()
#16 {main}

Next Revolt\EventLoop\UncaughtThrowable: Uncaught Amp\Future\UnhandledFutureError thrown in event loop callback Amp\Internal\FutureState::Amp\Internal\{closure} defined in C:\test\vendor\amphp\amp\src\Internal\FutureState.php:54; use Revolt\EventLoop::setErrorHandler() to gracefully handle such exceptions: Unhandled future: Amp\Process\ProcessException: "Received 0 of 5 expected bytes"; Await the Future with Future::await() before the future is destroyed or use Future::ignore() to suppress this exception. The future has been created at #0 C:\test\vendor\amphp\amp\src\functions.php:40 Amp\Internal\FutureState->__construct()
#1 C:\test\vendor\amphp\process\src\Internal\Windows\SocketConnector.php:100 Amp\async()
#2 C:\test\vendor\amphp\process\src\Internal\Windows\WindowsRunner.php:101 Amp\Process\Internal\Windows\SocketConnector->connectPipes()
#3 C:\test\vendor\amphp\process\src\Process.php:80 Amp\Process\Internal\Windows\WindowsRunner->start()
#4 C:\test\test.php:9 Amp\Process\Process::start() in C:\test\vendor\revolt\event-loop\src\EventLoop\UncaughtThrowable.php:13
Stack trace:
#0 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\AbstractDriver.php(400): Revolt\EventLoop\UncaughtThrowable::throwingCallback(Object(Closure), Object(Amp\Future\UnhandledFutureError))
#1 C:\test\vendor\revolt\event-loop\src\EventLoop\Internal\DriverSuspension.php(127): Revolt\EventLoop\Internal\AbstractDriver::Revolt\EventLoop\Internal\{closure}()
#2 C:\test\vendor\amphp\amp\src\Future.php(251): Revolt\EventLoop\Internal\DriverSuspension->suspend()
#3 C:\test\vendor\amphp\process\src\Internal\Windows\WindowsRunner.php(128): Amp\Future->await(NULL)
#4 C:\test\vendor\amphp\process\src\Process.php(123): Amp\Process\Internal\Windows\WindowsRunner->join(Object(Amp\Process\Internal\Windows\WindowsHandle), NULL)
#5 C:\test\test.php(19): Amp\Process\Process->join()
#6 {main}
  thrown in C:\test\vendor\revolt\event-loop\src\EventLoop\UncaughtThrowable.php on line 13

[PhantomArt]

I have found that different programs can behave differently. For example, the jpegoptim utility simply terminates execution even if I don't read stdout output. And what's more interesting is that I can read this output even after the process has completed.

$content = \file_get_contents('image.jpg');

$process = \Amp\Process\Process::start([
  'jpegoptim',
  '--strip-all',
  '--max=87',
  '-',
]);

$process->getStdin()->write($content);
$process->getStdin()->end();

$process->join();

$content = \Amp\ByteStream\buffer($process->getStdout());
\file_put_contents('image2.jpg', $content);

How to understand this behavior?

But at least in this example there is no data loss or exception. But what should I do in the case of ImageMagick if my application is quite complex and has a lot of asynchronous processes? For example, I want to be able to read data from stdout not instantly, but after an indefinite time (more than 120 seconds).

[PhantomArt]

I continued my research and came to the conclusion that there are two problems here at once. Below I will consider the situation with the ImageMagick call and the exception after 120 seconds.

First of all, I decided to try to reproduce this situation using proc_open().

$content = \file_get_contents('image.jpg');

$process = \proc_open(
  [
    'magick', '-',
    '-quality', '100',
    'JPEG:-',
  ], [
    0 => ['pipe', 'r'],
    1 => ['pipe', 'w'],
    2 => \STDOUT], // process stderr to output
  $pipes
);

\fwrite($pipes[0], $content);
\fclose($pipes[0]);

echo proc_close($process) . "\r\n";

I wanted to understand how this code would behave, whether the process would hang at the proc_close() call stage forever or whether it would exit after 120 seconds.

But it turned out that the process terminates instantly with code 1. The stderr output looks like this:

magick: Output file write error --- out of disk space? `-' @ error/jpeg.c/JPEGErrorHandler/348.
magick: unable to write file `-' @ error/constitute.c/WriteImage/1441.

So I decided instead of calling proc_close() to make a loop polling proc_get_status() to see if the status of the process was changing in any way over time, especially after 120 seconds.

$content = \file_get_contents('image.jpg');

$process = \proc_open(
  [
    'magick', '-',
    '-quality', '100',
    'JPEG:-',
  ], [
    0 => ['pipe', 'r'],
    1 => ['pipe', 'w'],
    2 => \STDOUT], // process stderr to output
  $pipes
);

\fwrite($pipes[0], $content);
\fclose($pipes[0]);

for ($i = 0; $i < 150; $i++) {
  echo $i . ': ' . json_encode(proc_get_status($process)) . "\r\n";
  sleep(1);
}

But it turned out that the status does not change in any way even after 150 seconds (I didn’t wait any longer):

149: {"command":"\"magick\" \"-\" \"-quality\" \"100\" \"JPEG:-\"","pid":25248,"running":true,"signaled":false,"stopped":false,"exitcode":-1,"termsig":0,"stopsig":0}

In general, I still don’t understand why the exception occurs after exactly 120 seconds when calling ImageMagick using \Amp\Process. But I still came to some conclusions.

Problem 1.

The exception occurs in Amp\Process\Internal\Windows\SocketConnector when trying to read 5 bytes from ProcessWrapper.exe (if I understand correctly).

private function readExitCode(ReadableResourceStream $stream): int
{
    $packet = \unpack('Csignal/Ncode', $this->read($stream, 5));

    if ($packet['signal'] !== SignalCode::EXIT_CODE) {
        throw new HandshakeException(HandshakeStatus::SIGNAL_UNEXPECTED);
    }

    return (int) $packet['code'];
}

This is a completely legitimate exception that can arise for various reasons. Although in my case this problem is 100% repeatable, it can be attributed to a feature of ImageMagick’s behavior (even if this is not confirmed by the experiment with proc_open()). I can take this into account and work with this limitation. But in that case, I want to move on to the second problem, which seems to me more serious.

Problem 2.

If this exception only occurred when calling $process->join(), then I wouldn't call it a problem. But the point is that this exception occurs in absolutely any random asynchronous context. For example, if instead of calling $process->join() I would use \Amp\delay().

for ($i = 0; $i < 150; $i++) {
  echo $i . "\r\n";
  \Amp\delay(1);
}

On the 120th iteration, \Amp\delay() throws an exception related to \Amp\Process. It seems to me that this should not be the case. This is not a legitimate context for this exception.

Please give me some comment on this so that I can move on. Is this normal behavior and should I be prepared to catch random exceptions from any async function? Or is this erroneous behavior that shouldn't exist?

[kelunik]

Hi @PhantomArt, this definitely doesn't sound like expected behavior. How do you know the exception is thrown there in the 120th iteration?

[PhantomArt]

Because I run a loop with a counter and a delay of 1 second inside, and the exception occurs after 120 iterations of 1 second.

But even before that, I took a different approach. I didn't mention this to keep the code snippets more concise. Essentially all my tests are wrapped in try-finally:

try {
  
  $content = \file_get_contents('image.jpg');
  
  $process = \Amp\Process\Process::start([
    'magick', '-',
    '-quality', '100',
    'JPEG:-',
  ]);
  
  $process->getStdin()->write($content);
  $process->getStdin()->end();
  
  $time = \Amp\now();
  
  echo $process->join() . "\r\n";
  
} finally {
  echo (\Amp\now() - $time) . " s\r\n";
}

Each time you start, the exception occurs after exactly 120 seconds (well, of course, with some fractions of milliseconds).

[kelunik]

I totally get that 120 seconds thing, which is another problem we need to look at, but my question was rather: You said the exception is thrown in that delay loop, are you sure about that? Can you catch it there? I'm asking because things are concurrent, so the exception being printed while that loop is running doesn't mean it's thrown into that loop.

[PhantomArt]

Yes, apparently it is. Here's an example:

$content = \file_get_contents('image.jpg');

$process = \Amp\Process\Process::start([
  'magick', '-',
  '-quality', '100',
  'JPEG:-',
]);

$process->getStdin()->write($content);
$process->getStdin()->end();

$time = \Amp\now();

try {
  \Amp\delay(150);
} catch (Throwable $ex) {
  echo "I caught the exception!\r\n\r\n";
  echo $ex->getMessage() . "\r\n\r\n";
}

echo (\Amp\now() - $time) . " s\r\n\r\n";

After 120 seconds I see in the console:

I caught the exception!

Uncaught Amp\Future\UnhandledFutureError thrown in event loop callback Amp\Internal\FutureState::Amp\Internal\{closure} defined in C:\test\vendor\amphp\amp\src\Internal\FutureState.php:54; use Revolt\EventLoop::setErrorHandler() to gracefully handle such exceptions: Unhandled future: Amp\Process\ProcessException: "Received 0 of 5 expected bytes"; Await the Future with Future::await() before the future is destroyed or use Future::ignore() to suppress this exception. The future has been created at #0 C:\test\vendor\amphp\amp\src\functions.php:40 Amp\Internal\FutureState->__construct()
#1 C:\test\vendor\amphp\process\src\Internal\Windows\SocketConnector.php:100 Amp\async()
#2 C:\test\vendor\amphp\process\src\Internal\Windows\WindowsRunner.php:101 Amp\Process\Internal\Windows\SocketConnector->connectPipes()
#3 C:\test\vendor\amphp\process\src\Process.php:80 Amp\Process\Internal\Windows\WindowsRunner->start()
#4 C:\test\test.php:10 Amp\Process\Process::start()

120.04010777600342 s

The exception message looks strange, it contains a stack trace, but apparently this is intended. You should know better about this than I do.

[kelunik]

Thanks! That trace is quite helpful. There's an async block that doesn't handle all exceptions that can occur. Instead of this exception being lost and silent, which makes things quite hard to debug, we throw such exceptions in an event loop callback. If there's an error handler set in the event loop, e.g. to log them, everything can continue as normal. If there's no error handler, we interrupt the suspension in {main}, which happens to be your delay(...) call here, to ensure the error doesn't go unnoticed. I've fixed this in 1ef7ac4.

[PhantomArt]

Great! Tell me how to install an exception handler that occurs in the event loop? I don't see this in the documentation.

Do I understand correctly that if such an exception occurs, then even if it is intercepted using such a handler, a Future is formed in my application, which I can never wait for using await?

[PhantomArt]

I found. I set up an event loop exception handler and tested what would happen with calls to $process->join() and \Revolt\EventLoop::run();.

\Revolt\EventLoop::setErrorHandler(function(\Throwable $ex) {
  echo "EventLoop Error!\r\n";
  echo get_class($ex) . ":\r\n";
  echo $ex->getMessage() . "\r\n\r\n";
});

$content = \file_get_contents('image.jpg');

$process = \Amp\Process\Process::start([
  'magick', '-',
  '-quality', '100',
  'JPEG:-',
]);

$process->getStdin()->write($content);
$process->getStdin()->end();

$time = \Amp\now();

try {
  \Amp\delay(130);
} catch (\Throwable $ex) {
  echo get_class($ex) . ":\r\n";
  echo $ex->getMessage() . "\r\n\r\n";
}

echo (\Amp\now() - $time) . " s\r\n\r\n";

try {
  echo "Process exit code: " . $process->join() . "\r\n";
} catch (\Throwable $ex) {
  echo get_class($ex) . ":\r\n";
  echo $ex->getMessage() . "\r\n\r\n";
}

echo (\Amp\now() - $time) . " s\r\n\r\n";

try {
  \Revolt\EventLoop::run();
} catch (\Throwable $ex) {
  echo get_class($ex) . ":\r\n";
  echo $ex->getMessage() . "\r\n\r\n";
}

echo (\Amp\now() - $time) . " s\r\n\r\n";

The console output looks like this:

EventLoop Error!
Amp\Future\UnhandledFutureError:
Unhandled future: Amp\Process\ProcessException: "Received 0 of 5 expected bytes"; Await the Future with Future::await() before the future is destroyed or use Future::ignore() to suppress this exception. The future has been created at #0 C:\test\vendor\amphp\amp\src\functions.php:40 Amp\Internal\FutureState->__construct()
#1 C:\test\vendor\amphp\process\src\Internal\Windows\SocketConnector.php:100 Amp\async()
#2 C:\test\vendor\amphp\process\src\Internal\Windows\WindowsRunner.php:101 Amp\Process\Internal\Windows\SocketConnector->connectPipes()
#3 C:\test\vendor\amphp\process\src\Process.php:80 Amp\Process\Internal\Windows\WindowsRunner->start()
#4 C:\test\test.php:15 Amp\Process\Process::start()

130.00098737402004 s

Error:
Event loop terminated without resuming the current suspension (the cause is either a fiber deadlock, or an incorrectly unreferenced/canceled watcher):

130.00159527899814 s

130.00179804000072 s

The call to $process->join() also fails with an exception, and while the exception message doesn't tell me what happened, at least I can reliably catch it and handle it.

Now I'm ready for any surprises that happen in the event loop. Maybe this message will be useful to someone. Thank you.

[kelunik]

Yeah, exactly like that! You still get an exception, because there definitely seems to be a bug. It hangs indefinitely on Linux, but I can reproduce the 120 second "timeout" on Windows.