Idle Timeouts do not work

[zmhh]

Currently AMQP Idle-Timeouts do not work for Listener connections. Sample code:

class Program
    {
        static void Main(string[] args)
        {
            var url = "amqp://localhost:8881";
            var host = new ContainerHost(new Uri(url));

            foreach (var listener in host.Listeners)
            {
                listener.AMQP.IdleTimeout = 5000;
            }

            host.RegisterLinkProcessor(new LinkProcessor());
            host.Open();
            Console.ReadLine();
        }
    }

    class LinkProcessor : ILinkProcessor
    {
        private int clients;

        public void Process(AttachContext attachContext)
        {
            Console.WriteLine(Interlocked.Increment(ref clients));

            attachContext.Link.Closed += (s, e) => Console.WriteLine(Interlocked.Decrement(ref clients));
            var link = new TargetLinkEndpoint(new MessageProcessor(), attachContext.Link);

            attachContext.Complete(link, 300);
        }
    }

    class MessageProcessor : IMessageProcessor
    {
        public int Credit { get; } = 300;
        public void Process(MessageContext messageContext)
        {
            Console.WriteLine(messageContext.Message.GetBody<string>());
            messageContext.Complete();
        }
    }

Despite setting the AMQP.IdleTimeout parameter to 5000, when a listener connection is opened in Connection.OnOpen, the "open.IdleTimeout" parameter which is used to set up the heartbeat timer is always the maximum uint value. Setting the idle timeout parameter thus does not cause listener connections to close if the connection is idle for the specified time.

[zmhh]

Upon further inspection idle timeouts do not appear to work for clients either. Sample code:

class Program
    {

        static void Main(string[] args)
        {
            var clients = new Task[1];

            for (int i = 0; i < clients.Length; i++)
            {
                clients[i] = Task.Run(async () => await WriteForeverAsync());
            }

            Task.WaitAny(clients);
        }

        private static TimeSpan Timeout = TimeSpan.FromMilliseconds(60000);
        private static TimeSpan CloseTimeout = TimeSpan.Zero;

        static async Task WriteForeverAsync()
        {
            var x = false;
            long y = Convert.ToInt64(x);
            var id = Guid.NewGuid().ToString();
            var msg = 0;

            Connection connection = null;
            Session session = null;
            SenderLink senderlink = null;

            try
            {
                while (true)
                {
                    await Task.Delay(100);

                    try
                    {

                        if (senderlink == null)
                        {
                            ConnectionFactory connFactory = new ConnectionFactory();
                            connFactory.AMQP.IdleTimeout = 5000;

                            connection = connFactory.CreateAsync(new Address("amqp://mhh-dev:8881")).Result;
                            session = new Session(connection);
                            senderlink = new SenderLink(session, id, "ingress_queue");
                        }


                        Console.WriteLine($"{id} : sending...");
                        await Task.Run(() => senderlink.Send(new Message(msg.ToString()), Timeout));
                        msg++;
                    }
                    catch (AmqpException e)
                    {
                        Console.WriteLine($"{id} : {e.Error.Condition}, {e.Error.Description}");

                        senderlink?.Close(CloseTimeout);
                        session?.Close(CloseTimeout);
                        connection?.Close(CloseTimeout);

                        senderlink = null;
                    }
                    catch (Exception e)
                    {
                        Console.WriteLine($"{id} : {e}");

                        senderlink?.Close(CloseTimeout);
                        session?.Close(CloseTimeout);
                        connection?.Close(CloseTimeout);

                        senderlink = null;
                    }


                    Console.WriteLine($"{id} : end");
                }
            }
            catch (Exception e)
            {
                Console.WriteLine($"{id} : UNEXPECTED, {e}");
            }
        }
    }

[zmhh]

From the spec, regarding Idle Timeouts:

"Connections are subject to an idle timeout threshold. The timeout is triggered by a local peer when no frames are received after a threshold value is exceeded. The idle timeout is measured in milliseconds, and starts from the time the last frame is received. If the threshold is exceeded, then a peer SHOULD try to gracefully close the connection using a close frame with an error explaining why. If the remote peer does not respond gracefully within a threshold to this, then the peer MAY close the TCP socket.

Each peer has its own (independent) idle timeout. At connection open each peer communicates the maximum period between activity (frames) on the connection that it desires from its partner.The open frame carries the idle-time-out field for this purpose. To avoid spurious timeouts, the value in idle-time-out SHOULD be half the peer's actual timeout threshold.

If a peer can not, for any reason support a proposed idle timeout, then it SHOULD close the connection using a close frame with an error explaining why. There is no requirement for peers to support arbitrarily short or long idle timeouts.

The use of idle timeouts is in addition to any network protocol level control. Implementations SHOULD make use of TCP keep-alive wherever possible in order to be good citizens.

If a peer needs to satisfy the need to send traffic to prevent idle timeout, and has nothing to send, it MAY send an empty frame, i.e., a frame consisting solely of a frame header, with no frame body. Implementations MUST be prepared to handle empty frames arriving on any valid channel, though implementations SHOULD use channel 0 when sending empty frames, and MUST use channel 0 if a maximum channel number has not yet been negotiated (i.e., before an open frame has been received). Apart from this use, empty frames have no meaning.

Empty frames can only be sent after the open frame is sent. As they are a frame, they MUST NOT be sent after the close frame has been sent.

As an alternative to using an empty frame to prevent an idle timeout, if a connection is in a permissible state, an implementation MAY choose to send a flow frame for a valid session.

If during operation a peer exceeds the remote peer's idle timeout's threshold, e.g., because it is heavily loaded, it SHOULD gracefully close the connection by using a close frame with an error explaining why."

So there's a few things that should be implemented for the idle-timeout. The main ones:

-Developer can configure AMQP idle-timeout for peer 1 (done)
-idle-timeout is communicated to peer 2 from peer 1
-peer 2 sends heartbeat messages to peer 1 at a rate of half of the idle-timeout period
-peer 1 closes the connection if it has not received messages in the specified idle-timeout period

[xinchen10]

idle-timeout is set locally and communicated to the remote peer. The remote peer is expected to honor this timeout by sending anything before it expires. You do not see a timer being created on the listener because the client did not set idle-timeout. When it is set by the listener, the client should start the timer and sends heartbeats when it is idle.

Right now only the last piece is missing. When the listener sets the timeout, it does not enforce it. Same for the client as amqp is more a peer-to-peer protocol. I will fix this missing part.

[zmhh]

hey @xinchen10, thanks for the quick response as always. if we use the above client and listener, with both parties setting idle timeouts, I don't see an idle timeout communicated or a heartbeat message triggered on either end. Are you seeing different behavior?

[zmhh]

Pull request #239 addresses communicating the idle-timeout in the open frame. This correctly triggers the heartbeat. Enforcing the timeout remains to be addressed

[zmhh]

Pull request #240 addresses enforcing the idle timeout

[zmhh]

Fixed in commit f6a6e70