Php-amqplib: Broken Pipe error, when using pcntl_fork

[henriqueoliveira]

Hello,

I m using pcntl_fork to generate consumers on demand, but the i got this errors, on the third fork, my application is working, before i get this message, i ve notice that the script hangs on Closing connection.

Fatal error: Uncaught exception 'PhpAmqpLib\Exception\AMQPRuntimeException' with message 'Broken pipe or closed connection' in /vagrant/vendor/videlalvaro/php-amqplib/PhpAmqpLib/Wire/IO/StreamIO.php:174

Thanks

[roihq]

@henriqueoliveira @videlalvaro This should be fixed with PR 195

[henriqueoliveira]

Thanks @roihq and @videlalvaro

Awesome project by the way.

[o10g]

No, actually PR 195 doesn't fix the problem.

The problem is that child process disconnects the RabbitMQ and the parrent, which uses the same connection, now can't reconnect, as it's failed on closeChannels() as they are already closed.

[dangson]

@o10g Did you ever find a solution to this problem? I'm having the same issue.

[o10g]

@dangson , yes, I found the solution after I looked at the code more deeply:

$connection = new AMQPLazyConnection(
    $conf->rabbitmq->host,
    $conf->rabbitmq->port,
    $conf->rabbitmq->user,
    $conf->rabbitmq->password,
    $conf->rabbitmq->vhost
);
$connection->set_close_on_destruct(false);

The last line solves the problem.

[michaelklishin]

Assuming this can be closed then. @o10g if you find a way this can be reflected in the docs, please submit a PR, would be great to make this easier to discover (I don't think we can enforce this behaviour for all users).

[dangson]

@o10g, doesn't that leave a bunch of open connections since they're not closed when the objects are destructed?

[cblanquera]

@dangson I just checked it closes it.

without $connection->set_close_on_destruct(false);

$ ps -A | grep php
  391 ??         0:22.29 /usr/local/opt/php70/sbin/php-fpm --fpm-config /usr/local/etc/php/7.0/php-fpm.conf
  419 ??         0:00.00 /usr/local/opt/php70/sbin/php-fpm --fpm-config /usr/local/etc/php/7.0/php-fpm.conf
  420 ??         0:00.00 /usr/local/opt/php70/sbin/php-fpm --fpm-config /usr/local/etc/php/7.0/php-fpm.conf
43291 ttys002    0:00.07 php work.php
43292 ttys002    0:00.00 php work.php
43293 ttys002    0:00.00 php work.php
43294 ttys002    0:00.00 php work.php
43295 ttys002    0:00.00 php work.php
43296 ttys002    0:00.00 php work.php
43300 ttys002    0:00.02 php work.php
43301 ttys002    0:00.00 php work.php
43302 ttys002    0:00.00 php work.php
43303 ttys002    0:00.00 php work.php
43304 ttys002    0:00.00 php work.php
43305 ttys002    0:00.00 php work.php
43313 ttys002    0:00.00 php work.php
43314 ttys002    0:00.00 php work.php
43315 ttys002    0:00.00 php work.php
43318 ttys002    0:00.00 php work.php
43319 ttys002    0:00.00 php work.php
43323 ttys002    0:00.00 php work.php
43324 ttys002    0:00.00 php work.php
43325 ttys002    0:00.00 php work.php
43326 ttys002    0:00.00 php work.php
43327 ttys002    0:00.00 php work.php
43337 ttys004    0:00.00 grep php

with $connection->set_close_on_destruct(false);

$ ps -A | grep php
  391 ??         0:22.32 /usr/local/opt/php70/sbin/php-fpm --fpm-config /usr/local/etc/php/7.0/php-fpm.conf
  419 ??         0:00.00 /usr/local/opt/php70/sbin/php-fpm --fpm-config /usr/local/etc/php/7.0/php-fpm.conf
  420 ??         0:00.00 /usr/local/opt/php70/sbin/php-fpm --fpm-config /usr/local/etc/php/7.0/php-fpm.conf
43670 ttys002    0:00.15 php work.php
43713 ttys004    0:00.00 grep php

All is well and this also solved my problem.

[andrefigueira]

@o10g I thank you so much, I was having this issue on only a specific environment and this has also solved my issue

[donarmando89]

@o10g please share a sample code of forking consumers as I am getting an error

Broken pipe or closed connection

When spawning workers using the same connection but different channels. Thanks in advance.

This is the basic code:

<?php
use PhpAmqpLib\Connection\AMQPLazyConnection;
use PhpAmqpLib\Message\AMQPMessage;

$start_script = getMicroDate();

$queue_name =Q_NAME;
$exchange_name = EXCHANGE;
$routing_key = ROUTING_KEY;
$maxConsumers = Q_MAX_CONSUMERS;
$log_file = FILE;
$host = RABBIT_MQ_HOST;
$port = RABBIT_MQ_PORT;
$user = RABBIT_MQ_USER;
$password = RABBIT_MQ_PWD;
$vhost = RABBIT_MQ_VHOST;

$connection = new AMQPLazyConnection($host, $port, $user, $password, $vhost);
$connection->set_close_on_destruct(false);

forkWorkers();

function forkWorkers() {
    global $connection;
    $channel = $connection->channel();

    list(, $messageCount, $consumerCount) = $channel->queue_declare($queue_name, false, true, false, false); // durable queue
    try {
        $worker_count = $messageCount > TPS ? TPS : $messageCount;

        for ($i = 0; $i < $worker_count; $i++) {
            $pid = pcntl_fork();
            pcntl_signal(SIGCHLD, SIG_IGN);
            if (!$pid) {
                $child_channel = $connection->channel();
                Consume($i, $child_channel);
                exit($i);
            }
        }
        $status = null;
        do {
            $pid = pcntl_wait($status);
        } while ($pid > 0);
    } catch (Exception $ex) {
        die("Consumer Error: " . $ex->getMessage() . "\n");
    }
    $connection->close();
}

function consume($i, $channel) {
    $queue_name = Q_NAME;
    $exchange_name = EXCHANGE;
    $routing_key = ROUTING_KEY;

    $channel->queue_declare($queue_name, false, true, false, false); // durable queue
    $channel->queue_bind($queue_name, $exchange_name, $routing_key);
    $channel->basic_qos(null, 1, null);


    $callback = function($msg) {
        global $i;
        echo "Worker $i [x] Received " . $msg->body . "\n";
        sleep(1);
        echo "Worker $i [x] Done" . "\n";
        $msg->delivery_info['channel']->basic_ack($msg->delivery_info['delivery_tag']);
    };

    $channel->basic_consume($queue_name, '', false, false, false, false, $callback);
    $ratelimit = ratelimiter(1, 1);
    while (count($channel->callbacks)) {
        $ratelimit();
        $channel->wait();
    }
    $channel->close();
}