Switch await core.sleep to await core.io_queue.queue_[read|write]

asyncio/core.py

@@ -148,11 +148,15 @@ def _dequeue(self, s):
         del self.map[id(s)]
         self.poller.unregister(s)
 
-    def queue_read(self, s):
+    async def queue_read(self, s):
         self._enqueue(s, 0)
+        _never.state = False
+        await _never
 
-    def queue_write(self, s):
+    async def queue_write(self, s):
         self._enqueue(s, 1)
+        _never.state = False
+        await _never
 
     def remove(self, task):
         while True:

asyncio/stream.py

@@ -60,8 +60,7 @@ async def read(self, n):
         This is a coroutine.
         """
 
-        core._io_queue.queue_read(self.s)
-        await core.sleep(0)
+        await core._io_queue.queue_read(self.s)
         return self.s.read(n)
 
     async def readinto(self, buf):
@@ -72,8 +71,7 @@ async def readinto(self, buf):
         This is a coroutine, and a MicroPython extension.
         """
 
-        core._io_queue.queue_read(self.s)
-        await core.sleep(0)
+        await core._io_queue.queue_read(self.s)
         return self.s.readinto(buf)
 
     async def readexactly(self, n):
@@ -87,8 +85,7 @@ async def readexactly(self, n):
 
         r = b""
         while n:
-            core._io_queue.queue_read(self.s)
-            await core.sleep(0)
+            await core._io_queue.queue_read(self.s)
             r2 = self.s.read(n)
             if r2 is not None:
                 if not len(r2):
@@ -105,8 +102,7 @@ async def readline(self):
 
         l = b""
         while True:
-            core._io_queue.queue_read(self.s)
-            await core.sleep(0)
+            await core._io_queue.queue_read(self.s)
             l2 = self.s.readline()  # may do multiple reads but won't block
             l += l2
             if not l2 or l[-1] == 10:  # \n (check l in case l2 is str)
@@ -129,7 +125,7 @@ async def drain(self):
         mv = memoryview(self.out_buf)
         off = 0
         while off < len(mv):
-            yield core._io_queue.queue_write(self.s)
+            await core._io_queue.queue_write(self.s)
             ret = self.s.write(mv[off:])
             if ret is not None:
                 off += ret
@@ -166,8 +162,7 @@ async def open_connection(host, port):
     except OSError as er:
         if er.errno != EINPROGRESS:
             raise er
-    core._io_queue.queue_write(s)
-    await core.sleep(0)
+    await core._io_queue.queue_write(s)
     return ss, ss
 
 
@@ -201,7 +196,7 @@ async def _serve(self, s, cb):
         # Accept incoming connections
         while True:
             try:
-                yield core._io_queue.queue_read(s)
+                await core._io_queue.queue_read(s)
             except core.CancelledError:
                 # Shutdown server
                 s.close()

examples/serial_examples.py

@@ -0,0 +1,81 @@
+# SPDX-FileCopyrightText: 2022 Phil Underwood
+#
+# SPDX-License-Identifier: Unlicense
+"""
+example that reads from the cdc data serial port in groups of four and prints
+to the console. The USB CDC data serial port will need enabling. This can be done
+by copying examples/usb_cdc_boot.py to boot.py in the CIRCUITPY directory
+
+Meanwhile a simple counter counts up every second and also prints
+to the console.
+"""
+
+
+import asyncio
+
+USE_USB = True
+USE_UART = True
+USE_BLE = True
+
+
+if USE_USB:
+    import usb_cdc
+
+    async def usb_client():
+
+        usb_cdc.data.timeout = 0
+        s = asyncio.StreamReader(usb_cdc.data)
+        while True:
+            text = await s.readline()
+            print("USB: ", text)
+
+
+if USE_UART:
+    import board
+
+    async def uart_client():
+
+        uart = board.UART()
+        uart.timeout = 0
+        s = asyncio.StreamReader(board.UART())
+        while True:
+            text = await s.readexactly(4)
+            print("UART: ", text)
+
+
+if USE_BLE:
+    from adafruit_ble import BLERadio
+    from adafruit_ble.advertising.standard import ProvideServicesAdvertisement
+    from adafruit_ble.services.nordic import UARTService
+
+    async def ble_client():
+        ble = BLERadio()
+        uart = UARTService()
+        advertisement = ProvideServicesAdvertisement(uart)
+        ble.start_advertising(advertisement)
+        s = asyncio.StreamReader(uart._rx)  # pylint: disable=protected-access
+        while True:
+            text = await s.read(6)
+            print("BLE: ", text)
+
+
+async def counter():
+    i = 0
+    while True:
+        print(i)
+        i += 1
+        await asyncio.sleep(1)
+
+
+async def main():
+    clients = [asyncio.create_task(counter())]
+    if USE_USB:
+        clients.append(asyncio.create_task(usb_client()))
+    if USE_UART:
+        clients.append(asyncio.create_task(uart_client()))
+    if USE_BLE:
+        clients.append(asyncio.create_task(ble_client()))
+    await asyncio.gather(*clients)
+
+
+asyncio.run(main())

examples/usb_cdc_boot.py

@@ -0,0 +1,9 @@
+# SPDX-FileCopyrightText: 2022 Phil Underwood
+#
+# SPDX-License-Identifier: Unlicense
+"""
+Save this file as boot.py on CIRCUITPY to enable the usb_cdc.data serial device
+"""
+import usb_cdc
+
+usb_cdc.enable(data=True, console=True)