Add uio.IOBase class to allow user-defined streams #3836

dpgeorge · 2018-06-04T06:40:45Z

This PR adds the uio.IOBase class which can be used to define user streams, which can then be passed into code that expects native stream support. The mapping from C to Python is:

stream_p->read  --> readinto(buf)
stream_p->write --> write(buf)
stream_p->ioctl --> ioctl(request, arg)

Among other things it allows the user to:

create an object which can be passed as the file arg to print: print(..., file=myobj), and then print will pass all the data to the object via the objects write method (same as CPython)
pass a user object to uio.BufferedWriter to buffer the writes (same as CPython)
use select.select on a user object
register user objects with select.poll
create user files that can be returned from user filesystems, and import can import scripts from these user files

For example, polling to "user space" is done via:

import io, select

class MyIO(io.IOBase):
    def ioctl(self, req, arg):
        print("IOCTL", req, arg)
        if req == 3: # MP_STREAM_POLL
            return 1 # I'm readable
        return 0

myio = MyIO()

print(select.select([myio], [], [], 0))

poller = select.poll()
poller.register(myio)
print(poller.poll(0))

I haven't tried it, but this should now allow to hook user objects into uasyncio, to allow waiting on user objects (eg, which fires when an external event happens, like a switch is pressed).

See the added test for an example of how to import from a user defined file.

This PR takes some code from #3034.

dpgeorge · 2018-06-04T06:44:30Z

I chose to call it uio.IOBase instead of uio.RawIOBase because IOBase is more primitive (in the CPython class hierarchy), and RawIOBase implies there's also a TextIOBase, but there is not. The code here doesn't care about bytes vs text. In the future if user TextIO is important then we can add TextIOWrapper to (among other things) wrap a base user stream (which would be derived from IOBase) to give that user stream text handling abilities.

For background reading on the IO class hierarchy, see https://www.python.org/dev/peps/pep-3116/ and https://docs.python.org/3/library/io.html

peterhinch · 2018-06-04T10:44:22Z

Awesome. This works with uasyncio, using a timer to create asynchronous events:

import io, pyb
import uasyncio as asyncio
MP_STREAM_POLL = const(3)
MP_STREAM_POLL_RD = const(1)

class MyIO(io.IOBase):
    def __init__(self):
        self.ready = False
        tim = pyb.Timer(4)
        tim.init(freq=1)
        tim.callback(self.setready)
        
    def ioctl(self, req, arg):
        if req == MP_STREAM_POLL and (arg & MP_STREAM_POLL_RD):
            state = pyb.disable_irq()
            r = self.ready
            self.ready = False
            pyb.enable_irq(state)
            return r
        return 0

    def readline(self):
        return 'ready\n'

    def setready(self, t):
        self.ready = True

myio = MyIO()

async def receiver():
    sreader = asyncio.StreamReader(myio)
    while True:
        res = await sreader.readline()
        print('Recieved', res)

loop = asyncio.get_event_loop()
loop.create_task(receiver())
loop.run_forever()

[EDIT] Fix bug above: must disable IRQ's in ioctl()

dpgeorge · 2018-06-04T11:36:51Z

This works with uasyncio

Great, thanks for testing. That means this PR would close #2664.

An alternative to providing this uio.IOBase is to implement duck-typing as per #2824.

peterhinch · 2018-06-04T14:09:54Z

I've tested this further, running it in the presence of other coroutines which perform an infinite loop with an asyncio.sleep(0) statement. The script below has the coros to toggle a pin on each iteration. The ISR sets another pin, with MyIO.readline() clearing it.

The outcome was that each pending coroutine ran in the period between the ISR running and MyIO.readline(). This accords with my understanding of uasyncio.core, where EventLoop.wait() (which schedules I/O) only runs when runq is exhausted.

I'm not sure if this has any practical implications, but I thought it worth putting on record.

Test script:

import io, pyb
import uasyncio as asyncio
import micropython
micropython.alloc_emergency_exception_buf(100)

MP_STREAM_POLL = const(3)
MP_STREAM_POLL_RD = const(1)

y1 = pyb.Pin('Y1', pyb.Pin.OUT)

class MyIO(io.IOBase):
    def __init__(self):
        self.ready = False
        self.count = 0
        tim = pyb.Timer(4)
        tim.init(freq=1)
        tim.callback(self.setready)
        
    def ioctl(self, req, arg):
        if req == MP_STREAM_POLL and (arg & MP_STREAM_POLL_RD):
            state = pyb.disable_irq()
            r = self.ready
            self.ready = False
            pyb.enable_irq(state)
            return r
        return 0

    def readline(self):
        y1.value(0)
        return '{}\n'.format(self.count)

    def setready(self, t):
        self.count += 1
        y1.value(1)
        self.ready = True

myio = MyIO()

async def foo(p):
    print('start foo', p)
    pin = pyb.Pin(p, pyb.Pin.OUT)
    while True:
        pin.value(1)
        await asyncio.sleep(0)
        pin.value(0)
        await asyncio.sleep(0)

async def receiver():
    last = None
    nmissed = 0
    sreader = asyncio.StreamReader(myio)
    while True:
        res = await sreader.readline()
        print('Recieved {} Missed {}'.format(res, nmissed))
        ires = int(res)
        if last is not None:
            if last != ires -1:
                print('Missed {}'.format(ires - 1))
                nmissed += 1
        last = ires

loop = asyncio.get_event_loop()
loop.create_task(receiver())
loop.create_task(foo('Y2'))
loop.create_task(foo('Y3'))
loop.run_forever()

peterhinch · 2018-06-05T17:36:54Z

I have now tested this under uasyncio with reading and writing, with buffered and unbuffered write devices, and failed to break it :)

The PR is excellent and has multiple use cases with uasyncio. In addition to polling hardware it enables devices which support serial communications to have an identical interface to the UART: via uasyncio's StreamReader and StreamWriter objects.

As an example I have a SynCom class which enables asynchronous communications between a pair of MicroPython devices using a bit banged interface (for the UART-starved ESP8266). With this PR SynCom would have the same interface as a UART. So modules could accept StreamReader and StreamWriter instances enabling them to run unchanged on devices with or without UARTs.

Clearly this could apply to other communications classes.

Footnote.
This effort has uncovered an issue with uasyncio's IORead mechanism that occurs in a specific circumstance and has a simple workround. I'll write this up in due course but my view is that it may be best documented rather than fixed.

A user class derived from IOBase and implementing readinto/write/ioctl can now be used anywhere a native stream object is accepted. For example: class MyOut(io.IOBase): def write(self, buf): print('write', repr(buf)) return len(buf) print('hello', file=MyOut())

It's a core feature, in particular required for user-streams with uasyncio.

dpgeorge · 2018-06-06T14:25:47Z

Thanks for the additional tests, good to know it's a useful feature.

I've now rebased this on top of current master and all checks pass. I've also enabled this uio.IOBase class on the standard unix port, esp8266, esp32 and stm32. Given its utility I think it's worth enabling on these ports.

peterhinch · 2018-06-07T06:27:29Z

On the subject of uasyncio do you have any thoughts on this problem on the forum?

peterhinch · 2018-06-08T10:11:55Z

The issue I mentioned seems to be an interaction between uasyncio and select/poll. It is as follows.

Assume a user class based on io.IOBase. If it supports reading only or writing only it works. If it supports both, it works only if writing is buffered (cf. the UART class). If writing is unbuffered, as soon as you start writing data, reading ceases. This script demonstrates it.

The workround for device drivers where the hardware does unbuffered writing is to declare separate classes for reading and writing, but we should know the cause of this issue.

If writing is buffered, StreamWriter.awrite() simply calls the device's buffered write() method once, which writes the entire buffer. This means that PollEventLoop only has to handle reading, which works.

If writing is unbuffered, StreamWriter.awrite() calls write() which writes a single byte. So

 yield IOWrite(self.s)

runs, which calls PollEventLoop.add_writer(). I believe the problem is in that method with

self.poller.register(sock, select.POLLOUT)

which seems to prevent read events from being detected. I think I've eliminated other possible causes. In the working case of my test script this is called with different sock instances, whereas in the failing case there is just one.

[EDIT] This is what I think is going on:
The "one shot" behaviour here causes the polled object's flags to be zeroed. So if you register an object for POLLIN and POLLOUT the registration of both will be cancelled whenever either event occurs. This will cause it to be unresponsive to the other event.

If separate polled objects are used for reading and for writing, zeroing the flags will affect only that object, so the problem does not arise.

I'm not sure

If this is actually correct.
How to prove or disprove it.
If it is correct, whether to regard it as a bug or a behaviour/workround to be documented.

dpgeorge · 2018-06-08T14:41:01Z

The "one shot" behaviour here causes the polled object's flags to be zeroed. So if you register an object for POLLIN and POLLOUT the registration of both will be cancelled whenever either event occurs. This will cause it to be unresponsive to the other event.

A way to test if this is causing the issue is to change the moduselect.c code to, in one shot mode, clear only those flags that were triggered. Instead of poll_obj->flags = 0; try:

poll_obj->flags &= ~(poll_obj->flags_ret & (MP_STREAM_POLL_RD | MP_STREAM_POLL_WR));

peterhinch · 2018-06-09T08:01:44Z

Thanks for that - I now have that in place.

Dealing with this case involves significant added complexity to uasyncio __init__.py and I'm concerned whether it is justified given the simple workround. I'd appreciate feedback as to if this is worth pursuing, or if there is some fundamental simplification I'm missing.

Here are the issues. In the following discussion the polled object is referred to as sock.

Given the one-shot ipoll behaviour it is necessary to repeatedly register sock. This implies keeping track of the flags already assigned to sock (in the existing code as soon as you add a writer the reader is cancelled). So the PollEventLoop needs a dict indexed by id(sock) and containing the flags as a bitmap.
The methods to remove readers and writers is adapted: you might remove a reader when a writer is still in place and vice-versa. Again the current flags dict is checked. If one flag is removed it registers the sock with it removed. If both flags are removed the sock is unregistered.
The ev value in wait() may have two bits set if ioctl has indicated that reader and writer are ready. This implies two objmap dicts, one for reading and one for writing; add_reader() and add_writer() assign to the relevant dict. wait() now schedules whichever tasks are pending depending on the ev bits.
With these changes the test almost works, but there is still a bug which causes it to fail after a short while. There is still some subtlety I'm missing and I have a nasty feeling it may be difficult...

dpgeorge · 2018-06-09T13:31:52Z

Here are the issues. In the following discussion the polled object is referred to as sock....

The steps you describe here to fix the issue seem to be a very close description of how CPython's asyncio handles the situation: to handle objects that may be registered as a reader and writer at the same time, CPython will keep track of the flags (POLLIN, POLLOUT) for registered objects and then use poller.modify() to update the flags of a given object if it is already registered. It does this when both adding and removing readers/writers.

@peterhinch you found this issue with a custom stream object that has slow reading and writing, but I think it would be possible to trigger the same issue with a native UART (with hardware flow control enabled) or even a socket object. You just have to register it for reading then try to awrite more data than it can handle at once.

This is a limitation of uasyncio as it currently stands: it cannot handle objects registered as both a reader and writer at the same time.

I'd appreciate feedback as to if this is worth pursuing, or if there is some fundamental simplification I'm missing.

It might be worth trying to fix your point (4), because this looks like the right direction to go in to support both reading and writing. At least then we'll know what's needed to make it work.

peterhinch · 2018-06-09T16:51:03Z

I now have code which runs my test scripts reliably. Point (4) was easily fixed despite its weird symptoms.

The bug was a consequence of the fact that the current code does not issue yield IOWriteDone(self.s) on completion, so ioctl continues to be queried for writer ready (which it is). I think IOWriteDone should be issued on completion of relevant write methods to stop ioctl being queried for writing until the next add_writer call.

In the light of your comments above I agree it should be fixed.

I have posted modified uasyncio code here and would appreciate comments. I have tested with simulated devices and also with pysical UARTs (tests directory).

dpgeorge · 2018-06-12T02:35:35Z

Ok, the addition of uio.IOBase was merged in af0932a.

peterhinch · 2018-06-12T08:26:14Z

@dpgeorge When you get a chance please could you look at my proposed fix for the read/write issue. If you think this is the way to go I'll raise a PR.

dpgeorge · 2018-06-13T07:42:53Z

When you get a chance please could you look at my proposed fix for the read/write issue.

It looks correct. It might be possible to rejig it so it doesn't need both rdobjmap and wrobjmap by storing both read and write in the same map, eg: objmap[id(sock)] = [rd_cb, rd_cb_args, wr_cb, wr_cb_args], and if an entry is None in this 4-list then that means there's nothing registered for that read/write yet. But it's not clear if this would make the code simpler/more efficient or not.

Separate to the implementation, we need to decide if this fix is needed or not. The only case where the problem arises is when the same stream object is registered with two (or more) coroutines, and then one coroutine waits to write while the other waits to read. Note that if the same stream is registered with two coroutines which both read (or both write) then it's never going to work properly (basically due to contention/race conditions), so having the same stream with two coroutines is already a dangerous thing to do, and potentially out of scope of uasyncio (but apparently not out of scope of CPython's asyncio...).

The only use for registering th 8000 e same stream with two (or more) coroutines, apart from tests, is to do pass-through (eg UART to UART, or UART to USB, TCP stream to UART, etc). For pass-through the read and write channels are really independent so can be handled by separate coroutines. But I can't think of any other reasonable protocol that would need the feature of being able to be registered with multiple coroutines.

peterhinch · 2018-06-14T05:42:50Z

Agreed. But with an inherently bidirectional device like a UART its normal use is to register precisely two coroutines, one to read and one to write. You pointed out that the existing UART code will break if running at a low baudrate or with hardware flow control. Low baudrate applications do exist e.g. RTTY in amateur radio. This reasoning convinced me (contrary to my original opinion) that this was a bug worth fixing.

For a practical example I wrote an asynchronous driver for GPS modules. These run a full duplex protocol: they send unsolicited messages and also respond to unsolicited command messages. The obvious (only?) way to handle this is with two coroutines. In this instance this works as there is no flow control, baudrates are high and the frequency of sending commands can be throttled. My point here is that full duplex protocols do exist which necessarily require two coroutines (or a ban on bidirectional objects).

With a half-duplex protocol there is no requirement concurrently to wait on reading and writing.

dpgeorge · 2018-06-14T06:18:54Z

But with an inherently bidirectional device like a UART its normal use is to register precisely two coroutines, one to read and one to write

I guess it depends on the protocol, whether the TX channel is independent to the RX channel or not. If they are dependent then uasyncio should work OK as-is, because a TX event will usually need to wait for a certain RX event, and vice versa. Although this is not entirely true: some protocols (think the TCP implementation of flow control) do allow some level of independence between the TX/RX path, where multiple packets are sent while responses are waited for.

For a practical example I wrote an asynchronous driver for GPS modules.

Ok, nice example. So that adds to the list of uses for handling a stream in two coroutines at the same time.

The obvious (only?) way to handle this is with two coroutines

If there was only one coroutine it would need to alternate reads and writes. That's obviously bad because what if it blocks on a read forever? (Is there a way to do reads with a timeout in uasyncio?)

Alternatively, with two coroutines, the sending coroutine could always do blocking writes. Since writes usually go through without too much delay it could work but is obviously defeating the purpose of uasyncio because nothing else can run while the write is waiting to go through.

peterhinch · 2018-06-14T08:21:56Z

I guess it depends on the protocol

You are describing a half-duplex protocol. Given access to both ends of the interface I prefer half-duplex; alas choice is absent when interfacing to a foreign device.

In my view neither of your solutions to full-duplex protocols are satisfactory, as you hinted.

A full duplex device requires the driver to always be listening. Given that the UART firmware uses interrupts your solutions will work if the tx buffer is big enough, tx baudrate is high enough, tx flow control is absent, and rx buffer size is big enough to accommodate incoming data. Those are a lot of constraints to meet in a guaranteed way, let alone to explain to inexperienced users :). The flow control issue is the killer here as we probably don't know how long a foreign device might block our tx. In this instance it may be impossible to produce a guaranteed worst-case solution.

You have convinced me that our UART driver is broken for asynchronous use. I do think it needs to be fixed, either by this proposal or otherwise. The ability to use the UART asynchronously is a great feature, hugely simplifying the task of dealing with full duplex devices. If it actually works :)

You also mentioned sockets as a possible area where we may fall foul of this.

The problem of user-written drivers is a lesser issue as we could decree that for devices which are physically bidirectional, two logically unidirectional drivers should be written (my original workround).

dpgeorge · 2018-06-14T14:32:40Z

You are describing a half-duplex protocol.

Yes you are right!

You also mentioned sockets as a possible area where we may fall foul of this.

I tried to think of some socket examples that were full-duplex. Eg a server that used UDP, receives packets and sends out packets could be full-duplex (like a DHCP server). But I'm not yet convinced that these would break uasyncio. Maybe you can think of a UDP example which is truly full-duplex and doesn't currently work with uasyncio?

The problem of user-written drivers is a lesser issue as we could decree that for devices which are physically bidirectional, two logically unidirectional drivers should be written

Right. For things like UART and sockets it's not really sensible to split them into, eg, UART_TX and UART_RX classes. For user classes it is possible but not very satisfactory because it breaks the abstraction of a single class (and instance object) representing the full device/channel. (Although there is at least one example of this: stdin and stdout.)

dhylands · 2018-06-14T17:28:19Z

An example which I've run into recently where full duplex is needed is having a pyboard which is monitoring a/some sensors. The host sends packets to the device to query information, and control output devices. If say, a button was pushed on the pyboard you may want to send that notification to the host immediately and not wait for the host to poll for a status change. However, the host might be in the process of sending packets to the device for something else (perhaps controlling an output) at the time that the device decides to send the notification to the host. Similar scenarios can happen when using wireless protocols like Zigbee over serial.

peterhinch · 2018-06-15T04:42:55Z

My knowledge of network protocols is insufficient for me to comment.

I do think we have a serious problem with the UART: full duplex protocols are common and should be simple to support. An example of a wi A93C dely used full duplex protocol is software flow control (ctrl-S/ctrl-Q AKA XOFF/XON). By definition ctrl-S can be expected to arrive while the UART is transmitting. With the current uasyncio this would break if the StreamWriter happened to be overloaded. A version allowing concurrent I/O would enable this to be supported in user code.

I entirely agree that splitting is not ideal for user classes, interesting observation about stdin/stdout notwithstanding.

dpgeorge · 2018-06-18T04:08:27Z

I think this discussion about duplex streams with uasyncio needs to be split out into a new issue.

peterhinch · 2018-06-18T07:29:00Z

I see one logical issue: the current uasyncio does not properly allow for a bidirectional stream device.

A solution will enable users to construct bidirectional classes. It will also enable the UART driver reliably to handle full duplex protocols. I think the case for handling full duplex protocols is unassailable: such hardware exists and people will want to interface to it.

The fix previously presented addresses this issue. I think further discussion should be around these points:

Do we intend to produce a solution?
Is the proposed solution acceptable, or can the team see a better way?

dpgeorge · 2018-06-29T07:40:10Z

I think the case for handling full duplex protocols is unassailable: such hardware exists and people will want to interface to it.

I agree.

Do we intend to produce a solution?

A solution would be good, but perhaps the right way to go is to create an alternative uasyncio event loop (built upon existing uasyncio components), instead of modifying the existing one. That way users can select the implementation that they need.

peterhinch · 2018-06-30T06:22:55Z

I think this depends on how you view the problem. I see it as a bug which needs to be fixed, because the official UART driver will fail under some circumstances. The handling of the poll flags seems incorrect. For all that it saves some code, is it wise for official uasyncio to contain a known bug?

An alternative event loop is a maintenance headache. When I believed Python I/O would be unsupported I used this approach to implement a priority-based version. When uasyncio 2 came along, considerable work was required to restore functionality. Part of my enthusiasm for uio.IOBase is that I can abandon uasyncio_priority with its associated support issue; changes to the event loop are hard to get right and time-consuming to develop.

dpgeorge force-pushed the py-add-iobase branch from 593a0be to 9c98d89 Compare June 6, 2018 04:58

dpgeorge added 4 commits June 6, 2018 23:05

ports: Enable IOBase on unix, stm32, esp8266 and esp32.

f0f5018

It's a core feature, in particular required for user-streams with uasyncio.

tests/io: Add simple IOBase test.

4f82e6c

tests/extmod: Add test for VFS and user-defined filesystem and files.

819d7ce

dpgeorge force-pushed the py-add-iobase branch from 9c98d89 to 819d7ce Compare June 6, 2018 13:05

tests: Exclude vfs_userfs test from native execution.

adef0a3

dpgeorge closed this Jun 12, 2018

dpgeorge deleted the py-add-iobase branch June 12, 2018 02:35

This was referenced Jun 15, 2018

RFC asyncio: efficiently polling devices #2664

Open

asyncio_priority peterhinch/micropython-async#16

Closed

peterhinch mentioned this pull request Jun 20, 2018

uasyncio fast_io EventLoop option reduces I/O scheduling latency micropython/micropython-lib#287

Open

peterhinch mentioned this pull request Oct 29, 2019

Cancelling coroutines: can't pend throw to just-started generator. #5242

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Uh oh!

Add uio.IOBase class to allow user-defined streams #3836

Add uio.IOBase class to allow user-defined streams #3836

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Add uio.IOBase class to allow user-defined streams #3836

Add uio.IOBase class to allow user-defined streams #3836

Uh oh!

Conversation

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!