uasyncio: best way to write to a file from a coroutine

[Jim.S]

I am writing a data logger that uses one coroutine to write data continuously to a file and another coroutine to terminate the event loop.
I am finding that if the loop is terminated before the "with open('filename') as f" loop completes, the file is never closed.

The code below shows the essential problem. On the Unix port, running on linux, all three files (sync_test.txt, async_test.txt, term_file.txt) are successfully closed and written. On the pyboard port, the file async_test.txt is not written. I’m guessing that this has something do do with the buffer that the file is being written to , or garbage collection, but don’t really understand how they work. Would manually calling the garbage collector at the end of the terminator ensure the "async_test.txt" file gets written?

What is a better way to write to a file from a coroutine that ensures that the file is written if the loop is terminated by an independent coroutine?

Code: Select all

import uasyncio as asyncio
def sync_file_write(name):
    with open(name,'w') as f:
        for i in range(5):
            f.write('sync_'+str(i)+'_')

async def file_write(name):
    with open(name,'w') as f:
        for i in range(50):
            f.write('async_'+str(i)+'_')
            await asyncio.sleep(1)

async def terminator(name, n):
     with open(name,'w') as f:
        for i in range(n):
            f.write('term_'+str(i)+'_')
            await asyncio.sleep(1)

print('start')
sync_file_write('sync_test.txt')
loop=asyncio.get_event_loop()
loop.create_task(file_write('async_test.txt'))
loop.run_until_complete(terminator('term_file.txt',10))
print('stop')

[kevinkk525]

Try this:

Code: Select all

async def file_write(name):
    for i in range(50):
        with open(name,"a") as f:
            f.write('async_'+str(i)+'_')
        await asyncio.sleep(1)

It opens the file only for writing and closes it afterwards. That way no file is kept open.
open(name,"a") opens the file for appending and does not replacing the previously added strings.

[pythoncoder]

@Jim.S I think you're asking a lot from a 'micro' implementation of asyncio That poor coro file_write is diligently writing data to a file when suddenly the rug is pulled out from under it and the scheduler is terminated. The code disappears down the rabbit hole of asyncio.sleep(1) and never returns, so the finalisation of the with block never gets a chance to run. At this point the data for the file is still buffered; because the file never gets closed it remains empty.

There are numerous solutions and that from @kevinkk525 is probably the simplest.

The best solution depends on what you actually want to happen when termination occurs. If you look at the tutorial in this repo you'll find various methods of coroutine synchronisation which will provide more flexible alternatives.

[Jim.S]

Figuring out how to use the synchronisation primitives was fun.

Appending works well [open(filename, 'a')], but as suggested by the name, it will append to that file on subsequent runs of the code. Use of barriers to synchronise the termination of the event loop with the closing of the file is more satisfying - I does require the use of a common flag (loop_flag), which I would normally implement as a class variable rather than a global.

I tried to figure out if I could use events (which I have successfully used to trigger coros from switches) but couldn’t easily get a solution.

I suppose that 'cancellable coroutines" might be simpler than using a common flag,

Code: Select all

import uasyncio as asyncio
import asyn

barrier = asyn.Barrier(2)

async def file_write_append(name):
    i=0
    with open(name,'a') as f:
        while True:
            i=i+1
            f.write('append_'+str(i)+'_')
            await asyncio.sleep(1)

async def file_write_barrier(name):
    i=0
    global loop_flag
    loop_flag=True
    with open(name,'w') as f:
        while loop_flag:
            i=i+1
            f.write('barrier_'+str(i)+'_')
            await asyncio.sleep(1)
    #file ought to be closed
    print('file closed')
    await barrier
    print('last line in file_write_barrier')

 async def terminator(time):
    await asyncio.sleep(time)
    global loop_flag
    loop_flag=False
    print('terminator waiting')
    await barrier #wait for file_write_barrier to finish
    print('terminator exit')

print('start')
#
loop=asyncio.get_event_loop()
loop.create_task(file_write_append('async_test.txt'))
loop.create_task(file_write_barrier('barrier_test.txt'))
loop.run_until_complete(terminator(5))
print('stop')

[pythoncoder]

[EDIT]This would be my approach. Note that the Try-Except is there only to demonstrate the mechanism - the context manager should close the files even in its absence. This avoids the need for a global and for the Barrier.

Code: Select all

import uasyncio as asyncio
import asyn

@asyn.cancellable
async def file_write_append(name):
    i=0
    try:
        with open(name,'a') as f:
            while True:
                i=i+1
                f.write('append_'+str(i)+'_')
                await asyn.sleep(1)  # Interruptible
    except asyn.StopTask:
        print('Append file closed')
 
@asyn.cancellable
async def file_write(name):
    i=0
    try:
        with open(name,'w') as f:
            while True:
                i=i+1
                f.write('barrier_'+str(i)+'_')
                await asyn.sleep(1)
    except asyn.StopTask:
        print('Write file closed')

 async def terminator(time):
    await asyncio.sleep(time)
    print('terminator waiting')
    await asyn.Cancellable.cancel_all()
    print('terminator exit')

print('start')
#
loop=asyncio.get_event_loop()
loop.create_task(asyn.Cancellable(file_write_append, 'async_test.txt')())
loop.create_task(asyn.Cancellable(file_write, 'write_test.txt')())
loop.run_until_complete(terminator(5))
print('stop')

This version without the exception trap also works. It's hard to prove that the files are closed, but their contents are correct and I have confidence in the MicroPython context manager.

Code: Select all

import uasyncio as asyncio
import asyn

@asyn.cancellable
async def file_write_append(name):
    i=0
    with open(name,'a') as f:
        while True:
            i=i+1
            f.write('append_'+str(i)+'_')
            await asyn.sleep(1)  # Interruptible
 
@asyn.cancellable
async def file_write(name):
    i=0
    with open(name,'w') as f:
        while True:
            i=i+1
            f.write('barrier_'+str(i)+'_')
            await asyn.sleep(1)

async def terminator(time):
    await asyncio.sleep(time)
    print('terminator waiting')
    await asyn.Cancellable.cancel_all()
    print('terminator exit')

print('start')
#
loop=asyncio.get_event_loop()
loop.create_task(asyn.Cancellable(file_write_append, 'async_test.txt')())
loop.create_task(asyn.Cancellable(file_write, 'write_test.txt')())
loop.run_until_complete(terminator(5))
print('stop')

[Jim.S]

Wow, I'm amazed that the cancellable option gives such a neat solution. It is interesting that the only essential differences between my original naive code and the cancellable version, without the explicit try except block, is the addition of the @asyn.cancellable decorator and the use of await asyn.Cancellable.cancel_all() in the terminator.

[Jim.S]

Of more interest to my purpose is the ability to terminate specific coroutines. I discovered that Cancellable coroutines can be assigned to groups in their construction , see the example code below, but specifically

loop.create_task(asyn.Cancellable(file_write_append, 'async_test.txt', group='a')())

Cancellation of group 'a' is then done with

await asyn.Cancellable.cancel_all(group='a')

Code: Select all

import uasyncio as asyncio
import asyn

@asyn.cancellable
async def file_write_append(name):
    i=0
    try:
        with open(name,'a') as f:
            while True:
                i=i+1
                f.write('append_'+str(i)+'_')
                await asyn.sleep(1)  # Interruptible
    except asyn.StopTask:
        print('Append file closed')
 
@asyn.cancellable
async def file_write(name):
    i=0
    try:
        with open(name,'w') as f:
            while True:
                i=i+1
                f.write('write_'+str(i)+'_')
                await asyn.sleep(1)
    except asyn.StopTask:
        print('Write file closed')

async def terminator(time):
    await asyncio.sleep(time)
    print ('terminator cancels group a')
    await asyn.Cancellable.cancel_all(group='a')## NOTE group key word
    await asyncio.sleep(1)
    print ('terminator cancels group 1')
    await asyn.Cancellable.cancel_all(group=1)
    print('terminator exit')

print('start')
#
loop=asyncio.get_event_loop()
## NOTE use of group key word
loop.create_task(asyn.Cancellable(file_write_append, 'async_test.txt', group=1)())
loop.create_task(asyn.Cancellable(file_write, 'write_test.txt', group='a')())
loop.run_until_complete(terminator(5))
print('stop')

[kevinkk525]

This is actually really awesome.
On the other hand it's only usable on the esp32 as the esp8266 has such low RAM that I would waste too much by using somehting like that.

[pythoncoder]

@kevinkk525 Not really so, especially if you're prepared to use frozen bytecode. @pfalcon has done an amazing job making uasyncio lightweight and RAM-efficient. Scheduling uses zero allocation. I have used it in fairly sizeable ESP8266 applications such as this repository.

uasyncio is very much more efficient than threading.

@Jim.S The concept of task groups is taken further with the NamedTask class where individual coroutines may be cancelled. They may also be interrogated to see if they are running.

[kevinkk525]

@pythoncoder: I'm using frozen bytecode and removed every unneccessary class but my project is already that big, it hardly fits into the RAM, and I only use a compressed version of your mqtt-library, a general component registration function, a htu21d, a generic gpio mqtt listener and a buzzer on pwm.. So I could not really afford to introduce a new class but if your project is more specialized towards a single usage then you probably have no problem using it.

I just meant that I have to be careful not to get carried away by all the nice high-level possibilities because they use more RAM and I'll end up having not enough.

Asyncio is very efficient and lightweight but even as frozen bytecode it needs ~3kB on import.

But I'm sorry, I did not want to get this thread in the wrong direction by my comment.
Cancellable coroutines are a great possibility.