How to exchange data between threads and uasyncio tasks and vice versa

[beyonlo]

Hi all!

I have a question about if is possible to write in a variable (msg_dict) using a thread and just read that same variable (msg_dict) at the same time, using a uasyncio task, to dump that variable to JSON to send via socket. I know that read and write the same variable using just uasyncio is not a problem, because each task happen at a different moment, but if is a thread writing and a uasyncio task reading at the same time?

I did a very simple example below as proof of concept, and I have no error - just luck? So I would like to know if this is safe to do and why not happen problem if a thread can write at the same time that a uasyncio will read.

In my understand, there is no problem to write on thread at same time that uasyncio are reading because the Python variable accept read and write at the same time. But I'm not sure if that is true, and if is safe!

I'm using ESP32-S3 with MicroPython 1.19.1

import time, _thread, random
import uasyncio as asyncio
import json

global msg_dict
msg_dict = None

def thread1():
    global msg_dict
    while True:
        r1, r2, time_now = random.randint(1, 999), random.randint(3000, 5000), time.ticks_ms()
        msg_dict = {'number_1': r1, 'number_2': r2, 'time': time_now}
        time.sleep_ms(1)

_thread.start_new_thread(thread1, ())

async def async_task1():
    global msg_dict
    while True:
        await asyncio.sleep_ms(1)
        res = json.dumps(msg_dict)
        #send JSON 'res' via socket()
        print(res)

async def main():
    asyncio.create_task(async_task1())
    await asyncio.sleep(10)

asyncio.run(main())

Output:
$ mpremote run thread3.py 
{"number_2": 4370, "time": 3525, "number_1": 6}
{"number_2": 4550, "time": 3535, "number_1": 778}
{"number_2": 3994, "time": 3545, "number_1": 666}
{"number_2": 4703, "time": 3555, "number_1": 978}
{"number_2": 4270, "time": 3565, "number_1": 719}

[dlech]

If the GIL is enabled, then it isn't possible for two threads to run at the same time. There is also ThreadSafeFlag if you need additional synchronization between a thread and an async task.

[beyonlo]

@dlech I see about the ThreadSafeFlag, I also read this tutorial, but how I use ThreadSafeFlag to be useful to transfer data from a thread to a async task in safe mode? Could you to provide an example?

Because, if I do inside the thread a ThreadSafeFlag.set() when I have new data wrote in the variable, and in the async task a ThreadSafeFlag.wait() waiting to read the variable, is possible that in the moment that uasyncio loop schedule the task to read the variable, the thread are writing a new data. Am I right?

Edit: I'm using MicroPython for the ESP32-S3 that I think the GIL is enabled. Please, correct me if I'm wrong, but the only MicroPython port that has the GIL disabled is the rp2040 port, right?

[dlech]

How about using a Queue so you never have to worry about writing over existing data?

[beyonlo]

@dlech excellent - it works fine, thank you for the tip. I pasted a simple example below (other reply) using the Queue and a note that as I use thread in one side, I need to use just Queue in the Synchronous mode.

[jimmo]

As @dlech says, on ESP32 the GIL means that threads do not run concurrently. We enable the GIL on ESP32 because it's single-core anyway. @beyonlo see the recent conversation we had about this on Discord (that I think you were part of?) https://discordapp.com/channels/574275045187125269/574275045611012097/1035373153809088512

The GIL essentially says that two threads cannot be executing bytecode at the same time, and that one bytecode instruction is atomic. So assigning a global variable for example, or two threads updating a shared dictionary. That said, it's still not safe to do higher level things like concurrently modifying a data structure where the update might span multiple bytecode operations (e.g. if you wrote your own data structure in Python).

If you were on rp2 then there is no GIL, so the rules are different. It's not safe for two threads to modify the same dictionary because the internal C code that mutates the dictionary structure could be running concurrently.

That said, in this particular case, assigning to an existing global variable is safe even without the GIL.

However, remember that global variables are themselves just a dictionary. So two threads that modify globals (i.e. doing a del variable) is not safe concurrently.

[peterhinch]

In terms of concurrency the multi-core, no GIL, case is worse than a hard ISR.

A MP hard ISR is not like a machine code interrupt. The MP runtime handles interrupts in such a way that Python instructions are considered atomic. If the main code is appending a tuple to a list, and a hard IRQ occurs, the bytecodes representing that line of Python source will run to completion before the ISR is executed. If this were not the case, a hard ISR would not be able reliably to read global data.

In a GIL-free multi-core environment where code running on each core contends for a common resource, all bets are off. You need a mutex.

In this context (referencing discussions earlier in this thread) I should be clear about mutexes. There is a hierarchy of Lock objects. In ascending order of rigour:

1. uasyncio.Lock Works only for uasyncio tasks running on a single uasyncio instance.
2. _thread.Lock Works for threads running under a single GIL. But see my final comment.
3. mutex This is the full-fat read-modify-write mutex which operates at machine code level.

@jonathanhogg In my opinion (though I'm willing to be proved wrong) your idea of a common event queue would require a true mutex to be multi core safe.

There is a question, though, which I haven't been able to answer either by trawling the code or by experiment. On RP2, does the _thread.Lock use a mutex?

[jimmo]

There is a question, though, which I haven't been able to answer either by trawling the code or by experiment. On RP2, does the _thread.Lock use a mutex?

Yes. It uses the non-recursive mutex primitive provided by the pico-sdk, which is implemented with a spin lock (irqs disabled, mem fence, etc).

[beyonlo]

My impression is that we may have such a queue with the means MP offers.

I agree with @rkompass - while MP has a non perfect implementation, we need a secure way to exchange data between thread and uasync task and vice versa.

@rkompass for now, can we to consider just the RQueue1 safe right? (many threads produce and only one uasyncio task consume). The RQueue2 and RQueue3 I see has no consense if it's safe or not. Anyway, I would like as well to do simultaneously a inverse way: to put data (produce) from uasyncio task and get data (consume) from the thread. Could you please add on the RQueue1 to be safe as well in this way (uasyncio task put and thread get)? Just to be clear: I would like that a thread can to put data to a uasyncio task get data and that same thread can as well get data from a usyncio task. Of course I will need in this case to have two queue right? queue1 = RQueue1; queue2 = RQueue1;

Thank you.

[rkompass]

For the case you mentioned:

many threads produce and only one uasyncio task consumes

and considering the restrictions (no upper memory limit) RQueue1 works and looks elegant, due to its conciseness. With several uasyncio tasks consuming, a new more elegant solution appears: The reformulated worker idea cast into a new ThreadSafeEvent class in the end lead to the abolishment of the need of a worker. I think @peterhinch will code a new Queue with the ability of consuming from different threads on this basis in the near future.?!?

The other way around: A single (or more?) asyncio tasks producing and different threads consuming should be doable with the help the _thread.Lock of which we now learnt that it is a true mutex. I will think about it in the next days and report here.
Of course it will be another Queue, but not a new Queue for each task.

[beyonlo]

many threads produce and only one uasyncio task consumes

and considering the restrictions (no upper memory limit) RQueue1 works and looks elegant, due to its conciseness.

All right, so for now I will use RQueue1, and if I need several uasyncio tasks consuming, I will to create a new queue for each task.

With several uasyncio tasks consuming, a new more elegant solution appears: The reformulated worker idea cast into a new ThreadSafeEvent class in the end lead to the abolishment of the need of a worker. I think @peterhinch will code a new Queue with the ability of consuming from different threads on this basis in the near future.?!?

That's a great news, I hope that @peterhinch has success :)

The other way around: A single (or more?) asyncio tasks producing and different threads consuming should be doable with the help the _thread.Lock of which we now learnt that it is a true mutex. I will think about it in the next days and report here. Of course it will be another Queue, but not a new Queue for each task.

If we have just a single uasyncio task producing to threads consuming will be excellent for now. If the solution for more tasks producing is more complex, a work around for now, maybe, is to create a new queue, like as on the RQueue1 (for several tasks conuming) - I don't know if follow the same rationale. Of course if we have a better solution for many uasyncio tasks producing, will be better, but for now a new queue can be a work around.

Note: We talk about in this topic much about threads producing and tasks consuming, but is need the inverse way as well: tasks producing and threads consuming - that is my case. I changed the title of this discussion clearer about this: How to exchange data between threads and uasyncio tasks and vice versa

Thank you!

[GM-Script-Writer-62850]

I have had no issues doing this, but i used a class instead of a global variable, note that i am using a pico and core 1 is only reading data set by core 0

[beyonlo]

I have had no issues doing this, but i used a class instead of a global variable, note that i am using a pico and core 1 is only reading data set by core 0

@GM-Script-Writer-62850 Could you please to paste a simple example here?

[GM-Script-Writer-62850]

https://github.com/GM-Script-Writer-62850/PICO_W_Thermostat/blob/main/PICO/main.py#L86-L133
function called on line 915, classes made at top of script
also see this file for some of my classes: https://github.com/GM-Script-Writer-62850/PICO_W_Thermostat/blob/main/PICO/GPIO.py

demo script i have from testing: https://pastebin.com/VxF4WsVT

[peterhinch]

I hope that @peterhinch has success :)

I am currently trying to define requirements. These are my current thoughts:

1. It should be thread-safe at the highest level: i.e. for GIL-free environments like RP2.
2. The queue should be non-allocating. It will hence have a maximum size defined on instantiation. I would expect to use a ringbuf.
3. It is uni-directional but an instance can be set to work in either direction - i.e. with uasyncio sourcing or sinking data.
4. One end of the queue will be one or more uasyncio tasks. The other end will be a single thread. See below for the reason.

Issues

I suspect it is impossible to code a Python function which will be the subject of contention from more than one GIL-free threads. The bytecode of the function call statement will be under contention before the code can set the lock. Unless some guru can correct me?

It may be possible to relax the single thread rule for environments with a GIL. I'm unsure on this. A ringbuf is not thread-safe with multiple sources or sinks, and waiting on a lock in an ISR is naughty. It may be that this limitation has to stay.

One nagging question. If I manage to write such a beast, how do we test it properly?

[beyonlo]

@rkompass For now I did just a test with only one thread producing and only one uasyncio task consuming. I was thinking firstly to paste here the results of this test before to do other tests, like as uasyncio task producing and thread consuming, and so on.

Works, but I see some strange behaviour:
Is not raising a Exception when queue is empty. When it is empty, it is just waiting until have data - like as an await. Look that has no await on the res = queue.get() and no await sleep in the async_task_consume function. And, after thread_produce finish and after await asyncio.sleep(7) end time, app do not end, I need to do a CTRL+C.

import _thread
from random import randint
from time import ticks_ms, ticks_diff, sleep_ms
import uasyncio as asyncio
import ThrgetQueue # The ThrgetQueue.py file with your code pasted above

queue = ThrgetQueue.ThrgetQueue()
ticks_start = ticks_ms()

def thread_produce():
    for i in range(4):
        sleep_ms(1000)
        r1, r2, now = randint(1, 999), randint(3000, 5000), ticks_diff(ticks_ms(), ticks_start)
        res = {'number_1': r1, 'number_2': r2, 'time': now}
        print('Putting result onto queue: {}'.format(res))
        queue.put_nowait(res)
    _thread.exit()

async def async_task_consume(queue):
    while True:
        print("Consume from queue...")
        print('queue lenght: {}'.format(len(queue.queue)))
        res = queue.get()
        print('Result queue is: {}'.format(res))

async def main():
    asyncio.create_task(async_task_consume(queue))
    await asyncio.sleep(7)
    print('Done')

thread = _thread.start_new_thread(thread_produce, ())
asyncio.run(main())

Output:

$ mpremote run queue1_ThrgetQueue.py 
Consume from queue...
queue lenght: 0
Putting result onto queue: {'number_2': 3140, 'time': 1002, 'number_1': 109}
Result queue is: {'number_2': 3140, 'time': 1002, 'number_1': 109}
Consume from queue...
queue lenght: 0
Putting result onto queue: {'number_2': 3769, 'time': 2002, 'number_1': 18}
Result queue is: {'number_2': 3769, 'time': 2002, 'number_1': 18}
Consume from queue...
queue lenght: 0
Putting result onto queue: {'number_2': 4261, 'time': 3003, 'number_1': 330}
Result queue is: {'number_2': 4261, 'time': 3003, 'number_1': 330}
Consume from queue...
queue lenght: 0
Putting result onto queue: {'number_2': 4747, 'time': 4004, 'number_1': 471}
Result queue is: {'number_2': 4747, 'time': 4004, 'number_1': 471}
Consume from queue...
queue lenght: 0
^C^C

[rkompass]

@beyonlo : What you describe is the intended behavior. IIRC this now is about the other way round: Threads consume and asyncio task produces. Because the consuming thread has no await (it is not in an asyncio context) it simply halts until there are data in the queue (thread gets started then by releasing its lock).
Did you use the same/similar example code like before? Then RQueue1 or RQueue3 are appropriate - for this direction: Threads produce and one (RQueue1) or more (RQueue3) asyncio tasks consume.

[beyonlo]

@rkompass Ohh, Now I understood: this new class above (named ThrgetQueue) is just for the uasyncio tasks to produce and threads to consume. If I want the inverse: to threads to produce and uasyncio tasks to consume I need to use RQueue1 or RQueue3. Can I to use all them simultaneously right? I will need the same thread to consume and produce for example.

Because the consuming thread has no await (it is not in an asyncio context) it simply halts until there are data in the queue (thread gets started then by releasing its lock).

Understood, so it is working, in the line res = queue.get() it simply halt, like as an await used on the uasyncio tasks - very good. So, for this class (ThrgetQueue) the consumer (just threads) will be not raise when queue is empty, right? But your comment on the class ThrgetQueue say that will raise if the queue is empty: def get(self, block=True): # get item from queue, wait for something if block is true; otherwise may raise Empty exception I think is just a wrong/old comment, am I right?

Follow the new test, for now, using just only one uasyncio task to produce, and only one thread do consume - working like a charm.

import _thread
from random import randint
from time import ticks_ms, ticks_diff, sleep_ms
import uasyncio as asyncio
import ThrgetQueue

queue = ThrgetQueue.ThrgetQueue()
ticks_start = ticks_ms()

def thread_consume():
    for i in range(4):
        sleep_ms(100)
        print("Consumer: consume from queue...")
        res = queue.get()
        print('Consumer: result queue is: {}'.format(res))
    _thread.exit()

async def async_task_produce(queue):
    while True:
        await asyncio.sleep_ms(1500)
        r1, r2, now = randint(1, 999), randint(3000, 5000), ticks_diff(ticks_ms(), ticks_start)
        res = {'number_1': r1, 'number_2': r2, 'time': now}
        print('Producer: putting result onto queue: {}'.format(res))
        queue.put_nowait(res)
        print('Producer: queue lenght: {}'.format(len(queue.queue)))

async def main():
    asyncio.create_task(async_task_produce(queue))
    await asyncio.sleep(5)
    print('Done')

thread = _thread.start_new_thread(thread_consume, ())
asyncio.run(main())

Output:

$ mpremote run queue2_ThrgetQueue.py 
Consumer: consume from queue...
Producer: putting result onto queue: {'number_2': 4013, 'time': 1506, 'number_1': 745}
Producer: queue lenght: 1
Consumer: result queue is: {'number_2': 4013, 'time': 1506, 'number_1': 745}
Consumer: consume from queue...
Producer: putting result onto queue: {'number_2': 3539, 'time': 3016, 'number_1': 541}
Producer: queue lenght: 1
Consumer: result queue is: {'number_2': 3539, 'time': 3016, 'number_1': 541}
Consumer: consume from queue...
Producer: putting result onto queue: {'number_2': 4953, 'time': 4526, 'number_1': 897}
Producer: queue lenght: 1
Consumer: result queue is: {'number_2': 4953, 'time': 4526, 'number_1': 897}
Consumer: consume from queue...
Done

[rkompass]

In def get(self, block=True): the block=True is the default, so if you use just queue.get() it blocks i.e. halts till there is something to get. I you use queue.get(False) or queue.get(block=False) it does not block but raises the Exception if there is nothing to get.

[beyonlo]

In def get(self, block=True): the block=True is the default, so if you use just queue.get() it blocks i.e. halts till there is something to get. I you use queue.get(False) or queue.get(block=False) it does not block but raises the Exception if there is nothing to get.

@rkompass I got it. Thank you for the clarification.

Can I to use all them simultaneously right?

And about that, is safe?

[peterhinch]

A question for the gurus (@jimmo ?). Assume two threads on different cores accessing a common object. One runs

     self._wi = (self._wi + 1) % self._size  # Values are small ints

and the other runs

   if self._ri == self._wi:

Obviously there is uncertainty as to whether, at the time of reading ._wi, the first expression has run to completion. But is the code safe? Can we guarantee that a valid result is returned rather than complete garbage? My gut feeling is that it is valid because a 32 bit write is atomic at machine code level but I might be missing something...

[jimmo]

@peterhinch Yes to the best of my knowledge this is safe.

Rationale:

1. self is a dictionary (i.e. py/map.c)
2. Loads and stores of 8/16/32 bit values are atomic. Ref. The follow-up reply saying "Yes" is from Graham at RPi. (Furthermore the dictionary value in the mp_map_elem_t must be 32-bit aligned so this is safe even on platforms that only provide this guarantee for aligned access).
3. A dictionary read can cause at most one write -- to the map cache, and this is to a uint_8.
4. A dictionary write to an existing key is implemented using exactly a read (another map cache update), followed by updating the value member of the returned item, which is mp_obj_t (i.e. single write of a 32-bit value).
5. There is no race for the cache, even between concurrent threads accessing different keys, i.e. with A reading and B writing, it's safe to do A-read-cache,B-read-cache,B-write-cache,B-update-value,A-write-cache because the cache is just a single read and is a hint to the search and the search only cares about the key. (And in most cases their IP will be different anyway).

Where this would stop being a safe is when you have a concurrent thread that is causing the self dict to change the underlying map, e.g. adding/removing a key.

Additionally, of course there's no guarantee here on write ordering. So two threads, one writing self.a and reading self.b, the other writing self.b and reading self.a could see surprising results. I don't know if we want to consider adding fence/barrier/cas/acquire-release-semantics to MicroPython. 😬

[peterhinch]

Here is my first pass at this with rudimentary docs here.

To test, on a Pico or Pico W, create a directory threadsafe and populate with __init__.py and threadsafe_queue.py (as a minimum).

In my opinion there is no need for a mutex under one condition. The end of the queue in the non-uasyncio thread must be driven by a single producer or consumer only. On the uasyncio side there can also only be a single producer or consumer task because of the restriction on ThreadSafeFlag. This restriction could be relaxed using ThreadSafeTask, but at some cost in complexity.

So far this simple test script has processed >7M items without error. Note that bi-directional communication is achieved with two queues.

import uasyncio as asyncio
from threadsafe import ThreadSafeQueue
import _thread
from time import sleep_ms

def core_2(getq, putq):  # echo server on core 2
    buf = []
    while True:
        while getq.qsize():
            buf.append(getq.get_sync())
        for x in buf:
            putq.put_sync(x, block=True)
        buf.clear()
        sleep_ms(30)
        
async def sender(to_core2):
    x = 0
    while True:
        await to_core2.put(x := x + 1)

async def main():
    to_core2 = ThreadSafeQueue([0 for _ in range(10)])
    from_core2 = ThreadSafeQueue([0 for _ in range(10)])
    _thread.start_new_thread(core_2, (to_core2, from_core2))
    asyncio.create_task(sender(to_core2))
    n = 0
    async for x in from_core2:
        if not x % 1000:
            print(f"Received {x} queue items.")
        n += 1
        assert x == n

asyncio.run(main())

This is beta quality and a lot more testing is needed. Comments and bug reports are welcome!

[beyonlo]

@peterhinch Amazing work, congratulations! I have your code above processing by hours and no errors. So I implemented, just for now, a very simple example: only one uasyncio task producing and only one thread consuming.

Works, but I note a error: if I create a thread just with only one queue, I have error. Details below. Will be necessary of course to pass two queue to thread when will be need that the same thread is a producer and consumer at the same time. But for cases (like this test) where the thread will need just to be a consumer, only one queue is enough, right?

Example working - was need to create two queue (queue1 and queue2 (not used))

import uasyncio as asyncio
from primitives import ThreadSafeQueue
import _thread
from time import sleep_ms, ticks_ms, ticks_diff
from random import randint

ticks_start = ticks_ms()

def thread_consume(get_queue, put_queue):
    while True:
        sleep_ms(1000)
        print("Consumer: consume from queue...")
        if get_queue.qsize():
            res = get_queue.get_sync(block=False)
            print('Consumer: result queue is: {}'.format(res))
        else:
            print('Consumer: nothing on the queue')

async def async_task_produce(put_queue):
    while True:
        await asyncio.sleep_ms(1500)
        r1, r2, now = randint(1, 999), randint(3000, 5000), ticks_diff(ticks_ms(), ticks_start)
        res = {'number_1': r1, 'number_2': r2, 'time': now}
        print('Producer: putting result onto queue: {}'.format(res))
        await put_queue.put(res)
        res = put_queue.qsize()
        print('Producer: queue lenght: {}'.format(res))

async def main():
    queue1 = ThreadSafeQueue([0 for _ in range(10)])
    queue2 = ThreadSafeQueue([0 for _ in range(10)])
    _thread.start_new_thread(thread_consume, (queue1, queue2))
    asyncio.create_task(async_task_produce(queue1))
    await asyncio.sleep(5)
    print('Done')

asyncio.run(main())

Output:

$ mpremote run queue2_threadsafe.py 
Consumer: consume from queue...
Consumer: nothing on the queue
Producer: putting result onto queue: {'number_2': 3141, 'time': 1517, 'number_1': 572}
Producer: queue lenght: 1
Consumer: consume from queue...
Consumer: result queue is: {'number_2': 3141, 'time': 1517, 'number_1': 572}
Consumer: consume from queue...
Consumer: nothing on the queue
Producer: putting result onto queue: {'number_2': 4138, 'time': 3027, 'number_1': 145}
Producer: queue lenght: 1
Consumer: consume from queue...
Consumer: result queue is: {'number_2': 4138, 'time': 3027, 'number_1': 145}
Producer: putting result onto queue: {'number_2': 4918, 'time': 4537, 'number_1': 646}
Producer: queue lenght: 1
Consumer: consume from queue...
Consumer: result queue is: {'number_2': 4918, 'time': 4537, 'number_1': 646}
Done

From code above, but using just only one queue:

import uasyncio as asyncio
from primitives import ThreadSafeQueue
import _thread
from time import sleep_ms, ticks_ms, ticks_diff
from random import randint

ticks_start = ticks_ms()

def thread_consume(get_queue):
    while True:
        sleep_ms(1000)
        print("Consumer: consume from queue...")
        if get_queue.qsize():
            res = get_queue.get_sync(block=False)
            print('Consumer: result queue is: {}'.format(res))
        else:
            print('Consumer: nothing on the queue')

async def async_task_produce(put_queue):
    while True:
        await asyncio.sleep_ms(1500)
        r1, r2, now = randint(1, 999), randint(3000, 5000), ticks_diff(ticks_ms(), ticks_start)
        res = {'number_1': r1, 'number_2': r2, 'time': now}
        print('Producer: putting result onto queue: {}'.format(res))
        await put_queue.put(res)
        res = put_queue.qsize()
        print('Producer: queue lenght: {}'.format(res))

async def main():
    queue1 = ThreadSafeQueue([0 for _ in range(10)])
    _thread.start_new_thread(thread_consume, (queue1))
    asyncio.create_task(async_task_produce(queue1))
    await asyncio.sleep(5)
    print('Done')

asyncio.run(main())

Output:

$ mpremote run queue2_threadsafe.py 
Traceback (most recent call last):
  File "<stdin>", line 36, in <module>
  File "uasyncio/core.py", line 1, in run
  File "uasyncio/core.py", line 1, in run_until_complete
  File "uasyncio/core.py", line 1, in run_until_complete
  File "<stdin>", line 31, in main
TypeError: object 'ThreadSafeQueue' isn't a tuple or list

Important: another problem that I found was in the docs, where say: data = q.get(block=True), but there is no get() method, just the get_sync().

Now I will to do more two different tests:

a) 1 queue - where only one thread producing and only one uasyncio task consuming - inverse of this test above.
b) 2 queue - where only one thread producing and consuming, and only one uasyncio task consuming and another only one uasyncio task producing - like as your working example.

Thank you very much!

[beyonlo]

Hey @peterhinch Here the results for that two different tests that I told you before.

Short answer: all ok.
I didn't put the tests in a stressed system yet, but examples works very fine - Thank you!

a) one queue - where only one thread producing and only one uasyncio task consuming.
Ps: two queue still needed as reported in the other test above.

import uasyncio as asyncio
from primitives import ThreadSafeQueue
import _thread
from time import sleep_ms, ticks_ms, ticks_diff
from random import randint

ticks_start = ticks_ms()
def thread_producer(get_queue, put_queue):
    while True:
        sleep_ms(1000)
        r1, r2, now = randint(1, 999), randint(3000, 5000), ticks_diff(ticks_ms(), ticks_start)
        res = {'number_1': r1, 'number_2': r2, 'time': now}
        print('Producer: putting result onto queue: {}'.format(res))
        try:
            put_queue.put_sync(res, block=False)
        except IndexError:
            print('Producer: Queue is full')
        res = put_queue.qsize()
        print('Producer: queue lenght: {}'.format(res))

async def async_task_consumer(get_queue):
    async for res in get_queue:
        print("Consumer from queue...")
        print('Consumer: result queue is: {}'.format(res))
        print('--------')

async def main():
    queue1 = ThreadSafeQueue([0 for _ in range(10)])
    queue2 = ThreadSafeQueue([0 for _ in range(10)])
    _thread.start_new_thread(thread_producer, (queue2, queue1))
    asyncio.create_task(async_task_consumer(queue1))
    await asyncio.sleep(4)
    print('Done')

asyncio.run(main())

Output:

$ mpremote run one_queue_thread_producer_and_task_consumer-threadsafe.py
Producer: putting result onto queue: {'number_2': 4821, 'time': 1013, 'number_1': 861}
Producer: queue lenght: 1
Consumer from queue...
Consumer: result queue is: {'number_2': 4821, 'time': 1013, 'number_1': 861}
--------
Producer: putting result onto queue: {'number_2': 4779, 'time': 2015, 'number_1': 436}
Producer: queue lenght: 1
Consumer from queue...
Consumer: result queue is: {'number_2': 4779, 'time': 2015, 'number_1': 436}
--------
Producer: putting result onto queue: {'number_2': 4099, 'time': 3016, 'number_1': 682}
Producer: queue lenght: 1
Consumer from queue...
Consumer: result queue is: {'number_2': 4099, 'time': 3016, 'number_1': 682}
--------
Done

b) two queue - where only one thread producing and consuming, and only one uasyncio task consuming and another only one uasyncio task producing.

import uasyncio as asyncio
from primitives import ThreadSafeQueue
import _thread
from time import sleep_ms, ticks_ms, ticks_diff
from random import randint

ticks_start = ticks_ms()
def thread_producer_and_consumer(get_queue, put_queue):
    while True:
        sleep_ms(1000)
        print('Start Thread Producer')
        r1, r2, now = randint(1, 999), randint(3000, 5000), ticks_diff(ticks_ms(), ticks_start)
        res = {'number_1': r1, 'number_2': r2, 'time': now}
        print('Producer Thread: putting result onto queue: {}'.format(res))
        try:
            put_queue.put_sync(res, block=False)
        except IndexError:
            print('Producer Thread: Queue is full')
        res = put_queue.qsize()
        print('Producer Thread: queue lenght: {}'.format(res))
        print('End Thread Producer')


        # Start Consumer
        print('Start Thread Consumer')
        print("Consumer Thread: consume from queue...")
        if get_queue.qsize():
            res = get_queue.get_sync(block=False)
            print('Consumer Thread: result queue is: {}'.format(res))
        else:
            print('Consumer Thread: nothing on the queue')
        print('End Thread Consumer')

async def async_task_producer(put_queue):
    c = 0
    while True:
        await asyncio.sleep_ms(1500)
        now = ticks_diff(ticks_ms(), ticks_start)
        res = {'counter': c, 'time': now}
        print('Producer Task: putting result onto queue: {}'.format(res))
        await put_queue.put(res)
        res = put_queue.qsize()
        print('Producer Task: queue lenght: {}'.format(res))
        c += 1

async def async_task_consumer(get_queue):
    async for res in get_queue:
        print("Consumer Task from queue...")
        print('Consumer Task: result queue is: {}'.format(res))

async def main():
    queue1 = ThreadSafeQueue([0 for _ in range(10)])
    queue2 = ThreadSafeQueue([0 for _ in range(10)])
    _thread.start_new_thread(thread_producer_and_consumer, (queue2, queue1))
    asyncio.create_task(async_task_consumer(queue1))
    asyncio.create_task(async_task_producer(queue2))
    await asyncio.sleep(4)
    print('Done')

asyncio.run(main())

Output:

$ mpremote run two_queue_thread_producer-and-consumer_and_task_consuming_and_another_task_producing-threadsafe.py 
Start Thread Producer
Producer Thread: putting result onto queue: {'number_2': 3149, 'time': 1013, 'number_1': 539}
Producer Thread: queue lenght: 1Consumer Task from queue...
End Thread Producer
Start Thread Consumer
Consumer Thread: consume from queue...
Consumer Thread: nothing on the queue
Consumer Task: result queue is: {'number_2': 3149, 'time': 1013, 'number_1': 539}

End Thread Consumer
Producer Task: putting result onto queue: {'time': 1516, 'counter': 0}
Producer Task: queue lenght: 1
Start Thread Producer
Producer Thread: putting result onto queue: {'number_2': 3829, 'time': 2038, 'number_1': 485}
Producer Thread: queue lenght: 1Consumer Task from queue...
Consumer Task: result queue is: {'number_2': 3829, 'time': 2038, 'number_1': 485}
End Thread Producer
Start Thread Consumer

Consumer Thread: consume from queue...
Consumer Thread: result queue is: {'time': 1516, 'counter': 0}
End Thread Consumer
Producer Task: putting result onto queue: {'time': 3026, 'counter': 1}
Producer Task: queue lenght: 1
Start Thread Producer
Producer Thread: putting result onto queue: {'number_2': 3813, 'time': 3067, 'number_1': 323}
Producer Thread: queue lenght: 1Consumer Task from queue...
Consumer Task: result queue is: {'number_2': 3813, 'time': 3067, 'number_1': 323}

End Thread Producer
Start Thread Consumer
Consumer Thread: consume from queue...
Consumer Thread: result queue is: {'time': 3026, 'counter': 1}
End Thread Consumer
Done

[beyonlo]

@peterhinch I would like to clarify a situation, I see that is documented: This queue is designed to interface between one uasyncio task and a single thread running in a different context. Is possible (is safe?) to have many threads, where each thread is consumer and producer, and many uasyncio tasks where each uasyncio task are producing and consuming from/to those threads?

Example: I have two threads, where each thread is consumer and producer at same time, and I have four uasyncio tasks for that, where two uasyncio tasks are consumers and two uasyncio tasks are producers, so total of four queue. Here the code for this example, it works very fine. I would like just to know if as I have more than one thread, (like as two, or five, or more), can still be safe? Of course always using only one queue for each producer/consumer.

import uasyncio as asyncio
from primitives import ThreadSafeQueue
import _thread
from time import sleep_ms, ticks_ms, ticks_diff
from random import randint

ticks_start = ticks_ms()
def thread_producer_and_consumer1(get_queue, put_queue):
    while True:
        sleep_ms(1000)
        print('Start Thread Producer_1')
        r1, r2, now = randint(1, 999), randint(3000, 5000), ticks_diff(ticks_ms(), ticks_start)
        res = {'number_1': r1, 'number_2': r2, 'time': now}
        print('Producer_1 Thread: putting result onto queue: {}'.format(res))
        try:
            put_queue.put_sync(res, block=False)
        except IndexError:
            print('Producer_1 Thread: Queue is full')
        res = put_queue.qsize()
        print('Producer_1 Thread: queue lenght: {}'.format(res))
        print('End Thread Producer_1')

        print('Start Thread Consumer_1')
        print("Consumer_1 Thread: consume from queue...")
        if get_queue.qsize():
            res = get_queue.get_sync(block=False)
            print('Consumer_1 Thread: result queue is: {}'.format(res))
        else:
            print('Consumer_1 Thread: nothing on the queue')
        print('End Thread Consumer_1')

def thread_producer_and_consumer2(get_queue, put_queue):
    while True:
        sleep_ms(1000)
        print('Start Thread Producer_2')
        r1, r2, now = randint(9999, 99999), randint(70000, 95000), ticks_diff(ticks_ms(), ticks_start)
        res = {'number_1': r1, 'number_2': r2, 'time': now}
        print('Producer_2 Thread: putting result onto queue: {}'.format(res))
        try:
            put_queue.put_sync(res, block=False)
        except IndexError:
            print('Producer_2 Thread: Queue is full')
        res = put_queue.qsize()
        print('Producer_2 Thread: queue lenght: {}'.format(res))
        print('End Thread Producer_2')

        print('Start Thread Consumer_2')
        print("Consumer_2 Thread: consume from queue...")
        if get_queue.qsize():
            res = get_queue.get_sync(block=False)
            print('Consumer_2 Thread: result queue is: {}'.format(res))
        else:
            print('Consumer_2 Thread: nothing on the queue')
        print('End Thread Consumer_2')

async def async_task_producer1(put_queue):
    c = 0
    while True:
        await asyncio.sleep_ms(1500)
        now = ticks_diff(ticks_ms(), ticks_start)
        res = {'counter': c, 'time': now}
        print('Producer_1 Task: putting result onto queue: {}'.format(res))
        await put_queue.put(res)
        res = put_queue.qsize()
        print('Producer_1 Task: queue lenght: {}'.format(res))
        c += 1

async def async_task_producer2(put_queue):
    c = 5000
    while True:
        await asyncio.sleep_ms(1500)
        now = ticks_diff(ticks_ms(), ticks_start)
        res = {'counter': c, 'time': now}
        print('Producer_2 Task: putting result onto queue: {}'.format(res))
        await put_queue.put(res)
        res = put_queue.qsize()
        print('Producer_2 Task: queue lenght: {}'.format(res))
        c += 1

async def async_task_consumer1(get_queue):
    async for res in get_queue:
        print("Consumer_1 Task from queue...")
        print('Consumer_1 Task: result queue is: {}'.format(res))

async def async_task_consumer2(get_queue):
    async for res in get_queue:
        print("Consumer_2 Task from queue...")
        print('Consumer_2 Task: result queue is: {}'.format(res))

async def main():
    queue1 = ThreadSafeQueue([0 for _ in range(10)])
    queue2 = ThreadSafeQueue([0 for _ in range(10)])
    queue3 = ThreadSafeQueue([0 for _ in range(10)])
    queue4 = ThreadSafeQueue([0 for _ in range(10)])
    _thread.start_new_thread(thread_producer_and_consumer1, (queue2, queue1))
    _thread.start_new_thread(thread_producer_and_consumer2, (queue4, queue3))
    asyncio.create_task(async_task_consumer1(queue1))
    asyncio.create_task(async_task_producer1(queue2))
    asyncio.create_task(async_task_consumer2(queue3))
    asyncio.create_task(async_task_producer2(queue4))

    await asyncio.sleep(4)
    print('Done')

asyncio.run(main())

Output:

$ mpremote run four_queue_two-thread_producer-and-consumer_and_two-task_consuming_and_another_two-task_producing-threadsafe.py 
Start Thread Producer_1Start Thread Producer_2
Producer_1 Thread: putting result onto queue: {'number_2': 3097, 'time': 1015, 'number_1': 45}
Producer_2 Thread: putting result onto queue: {'number_2': 71077, 'time': 1023, 'number_1': 20031}
Producer_1 Thread: queue lenght: 1
Producer_2 Thread: queue lenght: 1
End Thread Producer_2
Consumer_1 Task from queue...End Thread Producer_1
Start Thread Consumer_2
Consumer_2 Thread: consume from queue...
Consumer_1 Task: result queue is: {'number_2': 3097, 'time': 1015, 'number_1': 45}
Consumer_2 Thread: nothing on the queue
Start Thread Consumer_1

Consumer_1 Thread: consume from queue...Consumer_2 Task from queue...
Consumer_2 Task: result queue is: {'number_2': 71077, 'time': 1023, 'number_1': 20031}
Consumer_1 Thread: nothing on the queue
End Thread Consumer_2


End Thread Consumer_1
Producer_1 Task: putting result onto queue: {'time': 1523, 'counter': 0}
Producer_1 Task: queue lenght: 1
Producer_2 Task: putting result onto queue: {'time': 1525, 'counter': 5000}
Producer_2 Task: queue lenght: 1
Start Thread Producer_2
Producer_2 Thread: putting result onto queue: {'number_2': 82398, 'time': 2083, 'number_1': 30317}
Producer_2 Thread: queue lenght: 1Start Thread Producer_1Consumer_2 Task from queue...
Consumer_2 Task: result queue is: {'number_2': 82398, 'time': 2083, 'number_1': 30317}
End Thread Producer_2
Start Thread Consumer_2
Consumer_2 Thread: consume from queue...

Producer_1 Thread: putting result onto queue: {'number_2': 3083, 'time': 2104, 'number_1': 102}
Consumer_2 Thread: result queue is: {'time': 1525, 'counter': 5000}
Producer_1 Thread: queue lenght: 1

End Thread Producer_1
Start Thread Consumer_1
End Thread Consumer_2Consumer_1 Thread: consume from queue...
Consumer_1 Thread: result queue is: {'time': 1523, 'counter': 0}
End Thread Consumer_1Consumer_1 Task from queue...


Consumer_1 Task: result queue is: {'number_2': 3083, 'time': 2104, 'number_1': 102}
Producer_1 Task: putting result onto queue: {'time': 3033, 'counter': 1}
Producer_1 Task: queue lenght: 1
Producer_2 Task: putting result onto queue: {'time': 3035, 'counter': 5001}
Producer_2 Task: queue lenght: 1
Start Thread Producer_2
Producer_2 Thread: putting result onto queue: {'number_2': 73993, 'time': 3145, 'number_1': 28109}
Producer_2 Thread: queue lenght: 1Start Thread Producer_1Consumer_2 Task from queue...
Consumer_2 Task: result queue is: {'number_2': 73993, 'time': 3145, 'number_1': 28109}
End Thread Producer_2

Start Thread Consumer_2Producer_1 Thread: putting result onto queue: {'number_2': 3849, 'time': 3173, 'number_1': 220}
Producer_1 Thread: queue lenght: 1
Consumer_2 Thread: consume from queue...
Consumer_2 Thread: result queue is: {'time': 3035, 'counter': 5001}
End Thread Producer_1

Start Thread Consumer_1
Consumer_1 Task from queue...Consumer_1 Thread: consume from queue...
Consumer_1 Task: result queue is: {'number_2': 3849, 'time': 3173, 'number_1': 220}

Consumer_1 Thread: result queue is: {'time': 3033, 'counter': 1}
End Thread Consumer_2

End Thread Consumer_1
Done

[peterhinch]

Thanks for testing. There's a lot to go on there! Firstly your error message with one queue:

    _thread.start_new_thread(thread_consume, (queue1))

should read

    _thread.start_new_thread(thread_consume, (queue1,))  # One element tuple

The .start_new_thread method expects a tuple of args. Creating a one-element tuple in Python requires a trailing comma.

Regarding multiple threads, the rules are simple. Each ThreadSafeQueue should have exactly one producer and one consumer. Using multiple queues as you are doing looks fine.

One general point. In my opinion, on MicroPython, using threaded code on anything other than RP2 achieves nothing that isn't better achieved with uasyncio. The RP2 exception (which motivated my interest in threading) is because the threaded code runs on another core: there is a real gain in computational power. On other platforms there is only one Python virtual machine and execution is passed between the threads using some kind of time slicing algorithm. In uasyncio execution is passed between tasks. In uasyncio the context switching is done more efficiently, and crucially you have control over when it occurs. I cover this in more detail here - also read the article referenced there.

there is no get() method, just the get_sync().

This is a deliberate omission. I intended the class to be used as an asynchronous iterator with async for. I suspect that the API of CPython's asyncio.Queue exists for backwards compatibility. I designed this queue from scratch and async for seems a more modern interface.

That said, if you or anyone else suggests a use case for an asynchronous get() I will provide it

[rkompass]

@peterhinch I have some headaches regarding ThreadSafeQueue. Seeing

     while self.empty():  # Block until an item appears
         pass

in get_sync(), I'm wondering whether the thread trying to get from an empty queue will not cycle idly for a time.
Did you decide to accept this as the thread will be switched off anyway after a short time-slice?
On the other side, if this while loop was passed, because the queue was not empty, then during stepping over to the next instruction

r = self._q[self._ri]

cannot another instance have taken this element? Say even an asyncio task? (You restricted the number of threads to one, therefore only an asyncio task is allowed - in parallel?).

[beyonlo]

Hi @peterhinch thanks for the clarify that points.

One general point. In my opinion, on MicroPython, using threaded code on anything other than RP2 achieves nothing that isn't better achieved with uasyncio. The RP2 exception (which motivated my interest in threading) is because the threaded code runs on another core: there is a real gain in computational power. On other platforms there is only one Python virtual machine and execution is passed between the threads using some kind of time slicing algorithm. In uasyncio execution is passed between tasks. In uasyncio the context switching is done more efficiently, and crucially you have control over when it occurs.

I agree completely with you - I love the uasyncio. Problem that somethings is not supported by uasyncio. Just an example: the wlan.scan() block the uasyncio by ~3 seconds, so alternative to not have the application blocked by that long time, I use that wlan.scan() inside a thread - it works fine. So, my goal is to use threads just in cases where something is not supported by uasyncio, and/or for things that works better using threads. So, threads, in my opinion, still is very important for specific cases, like this.

Ps: for this application I'm using ESP32-S3.

[peterhinch]

@rkompass

    while self.empty():  # Block until an item appears
         pass

in get_sync(), I'm wondering whether the thread trying to get from an empty queue will not cycle idly for a time.
Did you decide to accept this as the thread will be switched off anyway after a short time-slice?

Code running in a non-uasyncio thread must be synchronous. It is valid to spin on a flag since other threads will still be time-sliced. The get_sync routine has only two possible ways to deal with an empty queue: raise an exception or spin. However it is up to the application how it deals with an empty queue. It can check for empty and do something else, only calling get_sync when there is something to get.

On the other side, if this while loop was passed, because the queue was not empty, then during stepping over to the next instruction

r = self._q[self._ri]

cannot another instance have taken this element? Say even an asyncio task? (You restricted the number of threads to one, therefore only an asyncio task is allowed - in parallel?).

There are precisely two threads in contention, the uasyncio thread and that controlling the other end of the queue. One is the producer and the other is the consumer. A circular buffer with a single producer and a single consumer has the property that only the producer controls the write pointer, and only the consumer controls the read pointer. You never get contention for the same pointer. This Wikipedia article explains the basics but it doesn't explain the significance of the non-contention.

@beyonlo Dealing with blocking code is definitely one use case for threading. The other one is multi-core coding on RP2 which is what has aroused my interest.

[beyonlo]

@peterhinch Just to be sure, in case of rp2040, if I decide to use on the core0 just a main loop - without the uasyncio: I need just to use put_sync() and/or get_sync() on core0 to exchange data with the core1 that is running the thread (also put_sync() and/or get_sync()), right?

[peterhinch]

If I understand you correctly, you want to use a ThreadSafeQueue to link synchronous code running on core 0 with synchronous code on core 1. One core calls put_sync() and the other calls get_sync(). This will work.

If you also want communication in the opposite direction you would need another queue.