|
1 |
| -# iotest4.py Test PR #3836. Demonstrate the anomaly with a read/write device. |
| 1 | +# iotest4.py Test PR #3836. |
2 | 2 | # User class write() performs unbuffered writing.
|
3 | 3 | # For simplicity this uses buffered read: unbuffered is tested by iotest2.py.
|
4 | 4 |
|
5 |
| -# Run iotest4.test() to see expected output |
6 |
| -# iotest4.test(False) to demonstrate the issue. |
| 5 | +# This test was to demonstrate the original issue. |
| 6 | +# With modified moduselect.c and uasyncio.__init__.py the test now passes. |
| 7 | + |
| 8 | +# iotest4.test() uses separate read and write objects. |
| 9 | +# iotest4.test(False) uses a common object (failed without the mod). |
7 | 10 |
|
8 |
| -# Pass/Fail is determined by whether the StreamReader and StreamWriter operate |
9 |
| -# on the same (fail) or different (pass) objects. |
10 |
| -# I suspect that the issue is with select/ipoll (uasyncio __init__.py) |
11 |
| -# The fault is either in select/poll or uasyncio __init__.py. |
12 |
| -# As soon as PollEventLoop.add_writer() is called, reading stops. |
13 |
| -# PollEventLoop.add_writer() is called when StreamWriter.awrite() issues |
14 |
| -# yield IOWrite(self.s), which for unbuffered devices is after the 1st char |
15 |
| -# of a multi-char buf is written. |
16 | 11 |
|
17 | 12 | import io, pyb
|
18 | 13 | import uasyncio as asyncio
|
|
25 | 20 | MP_STREAM_ERROR = const(-1)
|
26 | 21 |
|
27 | 22 | def printbuf(this_io):
|
28 |
| - print(this_io.wbuf[:this_io.wprint_len]) |
| 23 | + for ch in this_io.wbuf[:this_io.wprint_len]: |
| 24 | + print(chr(ch), end='') |
29 | 25 |
|
30 | 26 | class MyIO(io.IOBase):
|
31 | 27 | def __init__(self, read=False, write=False):
|
| 28 | + self.ready_rd = False # Read and write not ready |
| 29 | + self.wch = b'' |
32 | 30 | if read:
|
33 |
| - self.ready_rd = False |
34 | 31 | self.rbuf = b'ready\n' # Read buffer
|
35 | 32 | pyb.Timer(4, freq = 1, callback = self.do_input)
|
36 | 33 | if write:
|
37 | 34 | self.wbuf = bytearray(100) # Write buffer
|
38 | 35 | self.wprint_len = 0
|
39 | 36 | self.widx = 0
|
40 |
| - self.wch = b'' |
41 | 37 | pyb.Timer(5, freq = 10, callback = self.do_output)
|
42 | 38 |
|
43 | 39 | # Read callback: emulate asynchronous input from hardware.
|
|
0 commit comments