Send asyncio tasks to loop running in other thread

Question

How can I asynchronously insert tasks to run in an asyncio event loop running in another thread?

My motivation is to support interactive asynchronous workloads in the interpreter. I can't block the main REPL thread.

Example

My current flawed understanding says that the following should work. Why doesn't it? What is a better way to accomplish goal above?

import asyncio from threading import Thread  loop = asyncio.new_event_loop()  def f(loop):     asyncio.set_event_loop(loop)     loop.run_forever()  t = Thread(target=f, args=(loop,)) t.start()      @asyncio.coroutine def g():     yield from asyncio.sleep(1)     print('Hello, world!')  asyncio.async(g(), loop=loop) 

Answer 1

You must use call_soon_threadsafe to schedule callbacks from different threads:

import asyncio from threading import Thread  loop = asyncio.new_event_loop()  def f(loop):     asyncio.set_event_loop(loop)     loop.run_forever()  t = Thread(target=f, args=(loop,)) t.start()      @asyncio.coroutine def g():     yield from asyncio.sleep(1)     print('Hello, world!')  loop.call_soon_threadsafe(asyncio.async, g()) 

See https://docs.python.org/3/library/asyncio-dev.html#asyncio-multithreading for more information.

EDIT: Example of an interpreter supporting asynchronous workloads

# vim: filetype=python3 tabstop=2 expandtab  import asyncio as aio import random  @aio.coroutine def async_eval(input_, sec):   yield from aio.sleep(sec)   print("")   try:     result = eval(input_)   except Exception as e:     print("< {!r} does not compute >".format(input_))   else:       print("< {!r} = {} >".format(input_, result))  @aio.coroutine def main(loop):   while True:     input_ = yield from loop.run_in_executor(None, input, "> ")      if input_ == "quit":       break     elif input_ == "":       continue     else:       sec = random.uniform(5, 10)       print("< {!r} scheduled for execution in {:.02} sec>".format(input_, sec))       aio.async(async_eval(input_, sec))  loop = aio.get_event_loop()  loop.run_until_complete(main(loop)) loop.close() 

Answer 2

The first example in Jashandeep Sohi's answer does not work for me in 3.7+ and prints warnings about the deprecated annotation. I reworked this into a something that runs under 3.8. I tweaked it a little to meet my needs as well. I am new to multi-threading in Python (but no multithreading in general) so any advice, guidance, etc. is appreciated:

import asyncio from threading import Thread   loop = asyncio.new_event_loop() running = True   def evaluate(future):     global running     stop = future.result()     if stop:         print("press enter to exit...")         running = False   def side_thread(loop):     asyncio.set_event_loop(loop)     loop.run_forever()   thread = Thread(target=side_thread, args=(loop,), daemon=True) thread.start()   async def display(text):     await asyncio.sleep(5)     print("echo:", text)     return text == "exit"   while running:   text = input("enter text: ")   future = asyncio.run_coroutine_threadsafe(display(text), loop)   future.add_done_callback(evaluate)   print("exiting") 

The echo and other output will conflict with the prompts but it should be good enough to demonstrate it it working.

One thing I am unsure about is setting the global running from one thread and reading it from another. I think maybe the GIL synchronizes the thread cache but I'd love to get confirmation (or not) about that.