I2C: Add flow control. Reduce allocation.

Author: peterhinch

i2c/README.md

@@ -22,6 +22,11 @@ The `Initiator` implements a timeout enabling it to detect failure of the other
 end of the interface (the `Responder`). There is optional provision to reset
 the `Responder` in this event.
 
+## Changes
+
+V0.15 RAM allocation reduced and flow control implemented.
+V0.1 Initial release.
+
 ###### [Main README](../README.md)
 
 # Contents
@@ -36,7 +41,7 @@ the `Responder` in this event.
   4.3 [Responder class](./README.md#43-responder-class)  
  5. [Limitations](./README.md#5-limitations)  
   5.1 [Blocking](./README.md#51-blocking)  
-  5.2 [Buffering](./README.md#52-buffering)  
+  5.2 [Buffering and RAM usage](./README.md#52-buffering-and-ram-usage)  
   5.3 [Responder crash detection](./README.md#53-responder-crash-detection)  
 
 # 1. Files
@@ -184,6 +189,8 @@ Class variables:
  1. `timeout=1000` Timeout (in ms) before `Initiator` assumes `Responder` has
  failed.
  2. `t_poll=100` Interval (ms) for `Initiator` polling `Responder`.
+ 3. `rxbufsize=200` Size of receive buffer. This should exceed the maximum
+ message length.
 
 Class variables should be set before instantiating `Initiator` or `Responder`.
 See [Section 4.4](./README.md#44-configuration).
@@ -230,10 +237,12 @@ Constructor args:
 
 `Pin` instances passed to the constructor must be instantiated by `machine`.
 
-Class variable:
+Class variables:
  1. `addr=0x12` Address of I2C slave. This should be set before instantiating
  `Initiator` or `Responder`. If the default address (0x12) is to be overriden,
  `Initiator` application code must instantiate the I2C accordingly.
+ 2. `rxbufsize` Size of receive buffer. This should exceed the maximum message
+ length.
 
 ###### [Contents](./README.md#contents)
 
@@ -270,16 +279,19 @@ but involves explicit delays. I took the view that a 2-wire solution is easier
 should anyone want to port the `Responder` to a platform such as the Raspberry
 Pi. The design has no timing constraints and uses normal I/O pins.
 
-## 5.2 Buffering
+## 5.2 Buffering and RAM usage
 
-The protocol does not implement flow control, incoming data being buffered
-until read. To avoid the risk of memory errors a coroutine should read incoming
-data as it arrives. Since this is the normal mode of using such an interface I
-see little merit in increasing the complexity of the code with flow control.
+The protocol implements flow control: the `StreamWriter` at one end of the link
+will pause until the last string transmitted has been read by the corresponding
+`StreamReader`.
 
 Outgoing data is unbuffered. `StreamWriter.awrite` will pause until pending
 data has been transmitted.
 
+Efforts are under way to remove RAM allocation by the `Responder`. This would
+enable hard interrupts to be used, further reducing blocking. With this aim
+incoming data is buffered in a pre-allocated bytearray.
+
 ## 5.3 Responder crash detection
 
 The `Responder` protocol executes in a soft interrupt context. This means that

i2c/asi2c.py

@@ -25,7 +25,7 @@
 import uasyncio as asyncio
 import machine
 import utime
-from micropython import const
+from micropython import const, schedule
 import io
 
 _MP_STREAM_POLL_RD = const(1)
@@ -39,7 +39,7 @@
 
 # Base class provides user interface and send/receive object buffers
 class Channel(io.IOBase):
-    def __init__(self, i2c, own, rem, verbose):
+    def __init__(self, i2c, own, rem, verbose, rxbufsize):
         self.verbose = verbose
         self.synchronised = False
         # Hardware
@@ -49,9 +49,12 @@ def __init__(self, i2c, own, rem, verbose):
         own.init(mode=machine.Pin.OUT, value=1)
         rem.init(mode=machine.Pin.IN, pull=machine.Pin.PULL_UP)
         # I/O
-        self.txbyt = b''
+        self.txbyt = b''  # Data to send
+        self.txsiz = bytearray(2)  # Size of .txbyt encoded as 2 bytes
         self.rxbyt = b''
-        self.last_tx = 0  # Size of last buffer sent
+        self.rxbuf = bytearray(rxbufsize)
+        self.rx_mv = memoryview(self.rxbuf)
+        self.cantx = True  # Remote can accept data
 
     async def _sync(self):
         self.verbose and print('Synchronising')
@@ -67,18 +70,14 @@ def waitfor(self, val):  # Initiator overrides
         while not self.rem() == val:
             pass
 
-    # Concatenate incoming bytes instance.
+    # Get incoming bytes instance from memoryview.
     def _handle_rxd(self, msg):
-        self.rxbyt = b''.join((self.rxbyt, msg))
-
-    # Get bytes if present. Return bytes, len as 2 bytes, len
-    def _get_tx_data(self):
-        d = self.txbyt
-        n = len(d)
-        return d, n.to_bytes(2, 'little'), n
+        self.rxbyt = bytes(msg)
 
     def _txdone(self):
         self.txbyt = b''
+        self.txsiz[0] = 0
+        self.txsiz[1] = 0
 
 # Stream interface
 
@@ -91,7 +90,7 @@ def ioctl(self, req, arg):
                     if self.rxbyt:
                         ret |= _MP_STREAM_POLL_RD
                 if arg & _MP_STREAM_POLL_WR:
-                    if not self.txbyt:
+                    if (not self.txbyt) and self.cantx:
                         ret |= _MP_STREAM_POLL_WR
         return ret
 
@@ -110,13 +109,23 @@ def read(self, n):
         self.rxbyt = self.rxbyt[n:]
         return t.decode()
 
+    # Set .txbyt to the required data. Return its size. So awrite returns
+    # with transmission occurring in tha background.
     def write(self, buf, off, sz):
         if self.synchronised:
-            if self.txbyt:  # Waiting for existing data to go out
-                return 0
-            d = buf[off : off + sz]
+            if self.txbyt:  # Initial call from awrite
+                return 0  # Waiting for existing data to go out
+            # If awrite is called without off or sz args, avoid allocation
+            if off == 0 and sz == len(buf):
+                d = buf
+            else:
+                d = buf[off : off + sz]
+            d = d.encode()
+            l = len(d)
             self.txbyt = d
-            return len(d)
+            self.txsiz[0] = l & 0xff
+            self.txsiz[1] = l >> 8
+            return l
         return 0
 
 # User interface
@@ -136,8 +145,9 @@ def close(self):
 
 class Responder(Channel):
     addr = 0x12
+    rxbufsize = 200
     def __init__(self, i2c, pin, pinack, verbose=True):
-        super().__init__(i2c, pinack, pin, verbose)
+        super().__init__(i2c, pinack, pin, verbose, self.rxbufsize)
         loop = asyncio.get_event_loop()
         loop.create_task(self._run())
 
@@ -147,7 +157,7 @@ async def _run(self):
 
     # Request was received: immediately read payload size, then payload
     # On Pyboard blocks for 380μs to 1.2ms for small amounts of data
-    def _handler(self, _, sn=bytearray(2)):
+    def _handler(self, _, sn=bytearray(2), txnull=bytearray(2)):
 #        tstart = utime.ticks_us()  # TEST
         addr = Responder.addr
         self.rem.irq(handler = None, trigger = machine.Pin.IRQ_RISING)
@@ -156,28 +166,35 @@ def _handler(self, _, sn=bytearray(2)):
         self.own(1)
         self.waitfor(0)
         self.own(0)
-        n = int.from_bytes(sn, 'little')  # no of bytes to receive
+        n = sn[0] + ((sn[1] & 0x7f) << 8)  # no of bytes to receive
+        self.cantx = not bool(sn[1] & 0x80)  # Can Initiator accept a payload?
         if n:
             self.waitfor(1)
             utime.sleep_us(_DELAY)
-            data = self.i2c.readfrom(addr, n)
+            mv = memoryview(self.rx_mv[0 : n])  # allocates
+            self.i2c.readfrom_into(addr, mv)
             self.own(1)
             self.waitfor(0)
             self.own(0)
-            self._handle_rxd(data)
+            schedule(self._handle_rxd, mv)  # Postpone allocation
 
-        s, nb, n = self._get_tx_data()
         self.own(1)  # Request to send
         self.waitfor(1)
         utime.sleep_us(_DELAY)
-        self.i2c.writeto(addr, nb)
+        dtx = self.txbyt != b'' and self.cantx  # Data to send
+        siz = self.txsiz if dtx else txnull
+        if n or self.rxbyt:  # test n because .rxbyt may not be populated yet
+            siz[1] |= 0x80  # Hold off Initiator TX
+        else:
+            siz[1] &= 0x7f
+        self.i2c.writeto(addr, siz)
         self.own(0)
         self.waitfor(0)
-        if n:
+        if dtx:
             self.own(1)
             self.waitfor(1)
             utime.sleep_us(_DELAY)
-            self.i2c.writeto(addr, s)
+            self.i2c.writeto(addr, self.txbyt)
             self.own(0)
             self.waitfor(0)
             self._txdone()  # Invalidate source

i2c/asi2c_i.py

@@ -27,6 +27,7 @@
 import machine
 import utime
 import gc
+from micropython import schedule
 from asi2c import Channel
 
 # The initiator is an I2C slave. It runs on a Pyboard. I2C uses pyb for slave
@@ -36,9 +37,10 @@
 class Initiator(Channel):
     timeout = 1000  # ms Timeout to detect slave down
     t_poll = 100  # ms between Initiator polling Responder
+    rxbufsize = 200
 
     def __init__(self, i2c, pin, pinack, reset=None, verbose=True):
-        super().__init__(i2c, pin, pinack, verbose)
+        super().__init__(i2c, pin, pinack, verbose, self.rxbufsize)
         self.reset = reset
         if reset is not None:
             reset[0].init(mode=machine.Pin.OUT, value = not(reset[1]))
@@ -73,11 +75,12 @@ async def _run(self):
             self.rxbyt = b''
             await self._sync()
             await asyncio.sleep(1)  # Ensure Responder is ready
+            rxbusy = False
             while True:
                 gc.collect()
                 try:
                     tstart = utime.ticks_us()
-                    self._sendrx()
+                    rxbusy = self._sendrx(rxbusy)
                     t = utime.ticks_diff(utime.ticks_us(), tstart)
                 except OSError:
                     break
@@ -90,18 +93,26 @@ async def _run(self):
                 raise OSError('Responder fail.')
 
     # Send payload length (may be 0) then payload (if any)
-    def _sendrx(self, sn=bytearray(2)):
+    def _sendrx(self, rxbusy, sn=bytearray(2), txnull=bytearray(2)):
         to = Initiator.timeout
-        s, nb, n = self._get_tx_data()
-        self.own(1)  # Triggers interrupt on responder
-        self.i2c.send(nb, timeout=to)  # Must start before RX begins, but blocks until then
+        siz = self.txsiz if self.cantx else txnull
+        # rxbusy handles the (unlikely) case where last call received data but
+        # schedule() has not yet processed it
+        if self.rxbyt or rxbusy:
+            siz[1] |= 0x80  # Hold off further received data
+        else:
+            siz[1] &= 0x7f
+        # CRITICAL TIMING. Trigger interrupt on responder immediately before
+        # send. Send must start before RX begins. Fast responders may need to
+        # do a short blocking wait to guarantee this.
+        self.own(1)  # Trigger interrupt.
+        self.i2c.send(siz, timeout=to)  # Blocks until RX complete.
         self.waitfor(1)
         self.own(0)
         self.waitfor(0)
-        if n:
-            # start timer: timer CB sets self.own which causes RX
+        if self.txbyt and self.cantx:
             self.own(1)
-            self.i2c.send(s, timeout=to)
+            self.i2c.send(self.txbyt, timeout=to)
             self.waitfor(1)
             self.own(0)
             self.waitfor(0)
@@ -112,11 +123,15 @@ def _sendrx(self, sn=bytearray(2)):
         self.i2c.recv(sn, timeout=to)
         self.waitfor(0)
         self.own(0)
-        n = int.from_bytes(sn, 'little')  # no of bytes to receive
+        n = sn[0] + ((sn[1] & 0x7f) << 8)  # no of bytes to receive
+        self.cantx = not bool(sn[1] & 0x80)
         if n:
             self.waitfor(1)  # Wait for responder to request send
             self.own(1)  # Acknowledge
-            data = self.i2c.recv(n, timeout=to)
+            mv = memoryview(self.rx_mv[0 : n])
+            self.i2c.recv(mv, timeout=to)
             self.waitfor(0)
             self.own(0)
-            self._handle_rxd(data)
+            schedule(self._handle_rxd, mv) # Postpone allocation
+            return True
+        return False

i2c/i2c_esp.py

@@ -42,6 +42,12 @@
 
 async def receiver():
     sreader = asyncio.StreamReader(chan)
+    await chan.ready()
+    print('started')
+    for _ in range(5):  # Test flow control
+        res = await sreader.readline()
+        print('Received', ujson.loads(res))
+        await asyncio.sleep(4)
     while True:
         res = await sreader.readline()
         print('Received', ujson.loads(res))

i2c/i2c_init.py

@@ -44,6 +44,10 @@
 
 async def receiver():
     sreader = asyncio.StreamReader(chan)
+    for _ in range(5):  # Test flow control
+        res = await sreader.readline()
+        print('Received', ujson.loads(res))
+        await asyncio.sleep(4)
     while True:
         res = await sreader.readline()
         print('Received', ujson.loads(res))
@@ -63,6 +67,7 @@ async def test(loop):
     loop.create_task(receiver())
     loop.create_task(sender())
     while True:
+        await chan.ready()
         await asyncio.sleep(10)
         print('Blocking time {:d}μs max. {:d}μs mean.'.format(
             chan.block_max, int(chan.block_sum/chan.block_cnt)))

i2c/i2c_resp.py

@@ -41,6 +41,12 @@
 
 async def receiver():
     sreader = asyncio.StreamReader(chan)
+    await chan.ready()
+    print('started')
+    for _ in range(5):  # Test flow control
+        res = await sreader.readline()
+        print('Received', ujson.loads(res))
+        await asyncio.sleep(4)
     while True:
         res = await sreader.readline()
         print('Received', ujson.loads(res))