Question on streams

[zcattacz]

I tried to port mqtt_as to cpython for easier testing. sockets are replaced with streams. I have the following question on micro-python:

• How to set timeout to stream read ? the first read() in mqtt.connect() occasionally hang when tested in practically unstable network.
• I tried sr.s.settimeout(0) this doesn't seem to stop the hang ?
• If I use wait_for() on sr.read(), what would happen to socket?
• Why the error raise by wait_for() has no exception info ex.args=() , is this expected ? asyncio.TimeoutError
• Does closing the read stream close the socket, or write stream close the socket, or either will close ?

[peterhinch]

To close the socket associated with a stream issue

await stream_instance.wait_closed()

At least that is how it's done in MicroPython. I would expect this to be correct in CPython.

As a general comment on mqtt_as there is an assumption that, on initial connection, the network is reliable. In general if initial connection fails, the cause is something requiring manual intervention (bad broker address, bad credentials etc). Once a connection has been established outages are handled as per the docs.

[zcattacz]

On cpython, I use threads+queue often. My asyncio journey started on mpy. So I have no preceding experience with astreams. I may have silly questions :-D

I read the cpy doc, which states socket is transferred to StreamWritter, when open_connection() returns sw and sr. but e.g. on read error in my case, I wonder what happens to the sr,
do I just need to sw.close() and forget about sr ? or close sr+sw both ?
OK, on CPython StreamReader has no close().

I can't seem to find a way to set effective timeout on sr.read(). Actually I don't know what timeout I am looking for, a socket timeout or a stream timeout to avoid sr.read long blocking ? Either way I can't find any doc on how to set a effective timeout. So I suppose maybe it is designed to use with wait_for() like other async interface.
Then I start to wonder how this async timeout affect underlying i/o, if I do wait_for(connect(....inside..connect()...awit sr.read()...)) will it abort the read on socket immediately after wait_for() timeout? Or should I do something before any immediate read/write ?

PS: I am using a stripped down mqtt_as , without connectivity maintenance code, so that it runs on both mpy and cpy. will upload it when it's further tested. cpy lacks socket.read/write, mpy connect() still waiting timeout patch, so I just dropped socket with stream.

[peterhinch]

Much as I like and use streams I don't think they are the best way to deal with complex protocols like MQTT. There are reasons why I used nonblocking sockets in the design of mqtt_as and I think you're discovering them. I'm not clear why you've abandoned their use: CPython supports nonblocking sockets. Using streams changes the fundamental source of concurrency to select.poll() with blocking sockets.

However, to answer your question.

If you have a task that reads a stream you can subject it to a timeout with asyncio.wait_for(). You need to consider whether to close the socket in the mandatory error trap (see tutorial). If you've abandoned the connectivity maintenance you can (perhaps?) assume the socket is still connected.

If you cancel a stream read with a timeout you need to consider whether there may be circumstances where characters are left on the stream.

As a very general point if I were writing an asynchronous MQTT client using streams I'm not sure I'd use mqtt_as as a starting point because its basic concurrency model is so different. I'd start from scratch using the official synchronous libraries as a guide for managing the protocol. Having got that working I'd look to mqtt_as to provide inspiration for the parts of the protocol that the official clients don't properly support.

[zcattacz]

There are reasons why I used nonblocking sockets in the design of mqtt_as and I think you're discovering them.
Sure. Another advantage with socket, I can which IP address to connect.

I'm not clear why you've abandoned their use: CPython supports nonblocking sockets

I need my code to run on both cpython and micropython, since many of my devices run embedded linux. On this project, I think I can maintain the same code to support both.
Mainly because current socket.connect sometimes blocks > 10sec (server off lan segment) on mpy. I did a few loop test with this, if keep testing sometimes it hangs like forever until ^C. cpython doesn't have this issue, and once it's fixed in new mpy release, existing poll workaround will be redundant.

I kept both stream/socket version. The stream version has two differences. so far the stream version seems to work on par. I haven't noticed any issue except for the proper handling of timeout.

If you cancel a stream read with a timeout you need to consider whether there may be circumstances where characters are left on the stream.

I assume that means the socket buffer and socket state both in undetermined state, where a stream is aborted? Since I am not sure how that works in each individual case, I choose to leave the streams alone, until reconnect. They are closed and recreated right before connect.

[peterhinch]

I assume that means the socket buffer and socket state both in undetermined state, where a stream is aborted? Since I am not sure how that works in each individual case, I choose to leave the streams alone, until reconnect. They are closed and recreated right before connect.

That sounds wise. You need to be sure open sockets are closed: these are a source of memory leaks.