Issue with KeyboardInterrupt and generally STDIN

[dzemchmielewski]

During the development, it is nice to have an option to have control over starting and stopping the code. For this, I'm using the webrepl that at least allows me to issue exec(open('my_main.py').read()) and press CRTL-C to stop it. At least, because using select.poll() I can also send something into the running code from the STDIN.
The connect() mqtt_as call makes an issue with STDIN. Let's consider the following example code that is started using exec(open(...).read()):

from configuration import Configuration
from mqtt_as import MQTTClient
from mqtt_as import config as mqttconfig
import asyncio

TOPIC = "test/foo"

async def main(client):
    try:
        await client.connect()
    except OSError:
        print("Connection failed.")
        return
    n = 0
    while True:
        await asyncio.sleep(5)
        print("publish", n)
        await client.publish(TOPIC, f"counter: {n}")
        n += 1

mqttconfig["server'] = Configuration.MQTT_SERVER
mqttconfig["user"] = Configuration.MQTT_USERNAME
mqttconfig["password"] = Configuration.MQTT_PASSWORD
mqttconfig["ssid"] = Configuration.WIFI_SSID
mqttconfig["wifi_pw"] = Configuration.WIFI_PASSWORD
mqttconfig["port"] = 1883
mqttconfig["queue_len"] = 1

MQTTClient.DEBUG = True
client = MQTTClient(mqttconfig)

try:
    asyncio.run(main(client))
except KeyboardInterrupt:
    print("interrupt")
finally: 
    client.disconnect()
    asyncio.new_event_loop()

I expect, that after CTRL-C key press, an "interrupt" would appear on the output and program finish. Unfortunately, it doesn't happen - there is no reaction and only a hard reset is available.

After some digging I've found that the problem is with the async def _handle_msg(self) method:

    async def _handle_msg(self):
        try:
            while self.isconnected():
                async with self.lock:
                    await self.wait_msg()  # Immediate return if no message
                await asyncio.sleep_ms(0)  # Let other tasks get lock

        except OSError:
            pass
        self._reconnect()  # Broker or WiFi fail.

(lines 844-853 of mqtt_as/__init__.py). Making the non-zero sleep_ms makes things work as expected. Tested on ESP32-S3 and ESP32-C3.

I'm not sure why zero sleep prevents handling STDIN. I have some theorems in my mind, but I don't feel strong enough in async world to share with them. If anyone has a good understanding of the cause, please let me know.

It looks like the small delay (20ms) was replaced with zero last year. The potential solution is to make the delay configurable with some default value (like zero, what is at this moment).

Please, express your thoughts.

[peterhinch]

Interesting. The 20ms delays existed to support a project (mqtt-bridge) that has been retired. With a USB connection ctrl-c works fine so it must be an interaction with webrepl. I have limited knowledge of webrepl but it must use some clever techniques to enable concurrent support for the web interface.

Hopefully someone with more knowledge will advise.

[peterhinch]

Do you have a view on the necessary size of this delay for WebRepl to work? It might be worth testing 10ms and 5ms - I have no objection to reverting to 20ms so long as the value provides is a reasonable safety margin.

Removing the 20ms delay did not affect behaviour or performance.

[dzemchmielewski]

It looks like it is enough to make a 1ms delay.

I've also verified the ability to read other, then CTRL-C inputs from stdin, using the following task:

async def task_echo():
    reader = StdInReaderAs()
    while True:
        print("TYPE: ")
        line = await reader.get_input()
        line = line.strip()
        if len(line) > 0:
            print(f"ECHO: {line}")

where the stdin reader is defined that way:

class StdInReaderAs:
    def __init__(self):
        self.poller = select.poll()
        self.poller.register(sys.stdin, select.POLLIN)

    def _get_chr(self):
        for s, ev in self.poller.poll(0):
            return s.read(1)

    async def get_input(self):
        result = ""
        while True:
            c = self._get_chr()
            if c:
                if ord(c) == 10: # enter
                    print()
                    return result
                elif ord(c) == 27: # esc
                    return ""
                elif ord(c) == 127: # bs
                    result = result[:-1]
                else:
                    result += c
            else:
                await asyncio.sleep_ms(100)

Putting any non-zero time sleep allows you to type anything (although characters are not written back on the console) followed by enter, and "ECHO: anything" is displayed.
With zero time sleep, nothing like this is happening.

[peterhinch]

Thanks for testing. I will push an update with a 5ms sleep which will fix your problem without any adverse effects (mqtt_as worked fine with 20ms).

It does seem as if you've uncovered an issue with select.poll and asyncio. A sleep of zero should ensure that pending tasks are scheduled. It is odd because asyncio uses select.poll internally. You might want to raise an issue.

[EDIT]
We are currently working on a branch protocol_tests which is awaiting review. I have pushed the change to that branch - hopefully it will be merged in the next couple of weeks.