Broken pipe error if the task takes too much time.

[ixjosemi]

Hi guys!

I have been working with this library recently and I found an error when a task takes some much time to process.

Here is the example code that I have been using for my tests:

import asyncio
import time

from nats.aio.client import Client as NATS
from nats.errors import ConnectionClosedError, TimeoutError


async def main():
    nc = NATS()

    try:
        # It is very likely that the demo server will see traffic from clients other than yours.
        # To avoid this, start your own locally and modify the example to use it.
        await nc.connect(servers=["nats://127.0.0.1:4222"])
        # await nc.connect(servers=["nats://demo.nats.io:4222"])
    except Exception as e:
        print(f"Cannot connect to NATS: {e}")
        pass

    async def message_handler(msg):
        print(f"[Received on '{msg.subject}']: {msg.data.decode()}")
        print("Sleeping...")
        time.sleep(2400)

    try:
        # Interested in receiving 2 messages from the 'discover' subject.
        sub = await nc.subscribe("discover", "", message_handler)
        await sub.unsubscribe(2)

        await nc.publish("discover", b"hello")
        await nc.publish("discover", b"world")

        # Following 2 messages won't be received.
        await nc.publish("discover", b"again")
        await nc.publish("discover", b"!!!!!")
    except ConnectionClosedError:
        print("Connection closed prematurely")

    async def request_handler(msg):
        print("[Request on '{} {}']: {}".format(msg.subject, msg.reply, msg.data.decode()))
        await nc.publish(msg.reply, b"OK")

    if nc.is_connected:

        # Subscription using a 'workers' queue so that only a single subscriber
        # gets a request at a time.
        await nc.subscribe("help", "workers", cb=request_handler)

        try:
            # Make a request expecting a single response within 500 ms,
            # otherwise raising a timeout error.
            msg = await nc.request("help", b"help please", timeout=500)
            print(f"[Response]: {msg.data}")

            # Make a roundtrip to the server to ensure messages
            # that sent messages have been processed already.
            await nc.flush(timeout=1000)
        except TimeoutError:
            print("[Error] Timeout!")

        # Wait a bit for message to be dispatched...
        await asyncio.sleep(1)

        # Detach from the server.
        await nc.close()

    if nc.last_error is not None:
        print(f"Last Error: {nc.last_error}")

    if nc.is_closed:
        print("Disconnected.")


if __name__ == "__main__":
    asyncio.run(main())

As you see, there is a time.sleep that blocks the code for 40 minutes in order to test what happens if the process waits that much time.

After running the code I get the following error:

$ python nats_example.py 
[Received on 'discover']: hello
Sleeping...
 [Received on 'discover']: world
Sleeping...
[Request on 'help _INBOX.8u6cZWf5n6ErPpphJtYnzR.8u6cZWf5n6ErPpphJtYo1E']: help please
[Error] Timeout!
nats: encountered error
Traceback (most recent call last):
  File "/home/josemiguel.hernandez/.local/lib/python3.8/site-packages/nats/aio/client.py", line 1283, in _attempt_reconnect
    await self._io_writer.wait_closed()
  File "/usr/lib/python3.8/asyncio/streams.py", line 359, in wait_closed
    await self._protocol._get_close_waiter(self)
  File "/usr/lib/python3.8/asyncio/selector_events.py", line 910, in write
    n = self._sock.send(data)
BrokenPipeError: [Errno 32] Broken pipe
Disconnected.

Is there any way to avoid getting this Broken pipe error even if the process takes 5 minutes or 3 hours?

Thanks in advance!

[wallyqs]

The time sleep is blocking the single thread event loop from asyncio, so to workaround it you need to call the blocking task from another thread using something like the ThreadPoolExecutor: https://pymotw.com/3/asyncio/executors.html

After 6 minutes, the nats server will disconnect the client because of missing ping/pong replies, and the client would try to reconnect.

[ixjosemi]

Thanks for your reply @wallyqs! I'm curious about this:

After 6 minutes, the nats server will disconnect the client because of missing ping/pong replies, and the client would try to reconnect.

Why does the nats server disconnects after 6 minutes? Is there any way to increase/change this value?

[wallyqs]

That is the ping_interval setting which is by default 2 minutes, with max pings outstanding set to 3 so the third missing pong will disconnect it.

[ixjosemi]

Thanks again @wallyqs !

We are going to try to implement the threads approach instead of increasing the ping_interval value as we think that is a better way to solve this!

[alejandrogr]

Hello!

To give some more context, and what we end up doing to solve this particual problem:

What?

When a message is processed in a subscription callback function, if the processing takes a long time, trying to post a response the publish gives a Broken Pipe error and the consequent Timeout.

Why?

Subscription callback functions are run within an asyncio loop. But if the process synchronously blocks the thread for too long, all other processes living on that thread cannot continue. In this case, the subscription to NATs is blocked for a long time, losing the connection.

Users must be aware of this situation and act accordingly. They have two options:

• If the processing must be synchronous, they have to launch that processing in a separate thread, leaving the main thread free so that the rest of the processes that are executed in it can continue its correct operation. Using asyncio in combination with concurrent.futures. https://pymotw.com/3/asyncio/executors.html
• If the processing can be asynchronous, they must actively free the thread frequently enough to give the rest of the processes time to do their work using something like await asyncio.sleep(0) to free the thread

In this particular usecase, the callback function that was blocking the thread had this form:

    async def message_cb(msg):
        for i in range(x):  # x was a big number
            ### some code
            ### some code that blocks for 3~ min
            ### some more code

    await nc.publish("reply", b"Reply")
    await nc.flush(timeout=5)

With this scheme, when the process was executed, the thread was completely blocked until the loop finished. The solution to unblock the thread and not interrupt the processes has been the following:

    async def message_cb(msg):
        for i in range(x):  # x was a big number
            ### some code
            ### some code that blocks for 3~ min
            await asyncio.sleep(0) # this instruction frees the thread for the rest of the processes
            ### some more code

    await nc.publish("reply", b"Reply")
    await nc.flush(timeout=5)

[tamis-laan]

@wallyqs @alejandrogr
I'm having the same problem but I can't just inject a sleep(0) in the code to unblock the event loop.

Is it possible to run the nat's connection logic in a separate parallel thread?

This way my main code does not need to be heavily modified.

[alejandrogr]

If you cannot free the thread in the event loop, then your other option is to run your code in a separate thread.

My suggestion is to run a new thread inside the async def message_cb(msg) function and run your process inside that new thread.

[tamis-laan]

@alejandrogr
Yes that's what I'm doing now, I'm using

# Unblock event loop by separating process in separate thread
def unblock(fn):
    @wraps(fn)
    async def wrapped(*args, **kwargs):
        return await asyncio.to_thread(fn, *args, **kwargs)
    return wrapped 

But that doesn't make the code nicer to read. I'd figure the code would look a lot better if the NATS client connection loop would run in a parallel thread.