Python asyncio wait for threads

Question

I have a situation where I have a "server" thread which should listen calls/events from other server threads while at the same time executing some other code. Lately I have worked a lot with Node.js so I thought that it would be nice to use async/await to create an event loop where I can wait other threads to join in event loop and handle their response when they finally join.

To test the idea I wrote the following test script in Python 3.5:

# http://stackabuse.com/python-async-await-tutorial/
# Testing out Python's asynchronous features
import asyncio
from time import sleep
import threading
from threading import Thread
import random

class MyThread(Thread):

    def __init__(self, message):
        Thread.__init__(self)
        self._message = message

    def run(self):
        self._return = self._message + " oli viesti"
        a = random.randint(1, 5)
        print("Sleep for ", a)
        sleep(a)
        print("Thread exiting...")


    def join(self):
        Thread.join(self)
        return self._return



async def send(message):
    t = MyThread(message)  # daemon = True
    t.start()
    print("asd")
    return t.join()

async def sendmsg(msg):
    response = await send(msg)
    print("response is ", response)


if __name__ == "__main__":
    # Initiate a new thread and pass in keyword argument dictionary as parameters
    loop = asyncio.get_event_loop()
    tasks = [
        asyncio.ensure_future(sendmsg("hippa1"), loop=loop),
        asyncio.ensure_future(sendmsg("hippa2"), loop=loop),
        asyncio.ensure_future(sendmsg("hippa3"), loop=loop)
    ]

    loop.run_until_complete(asyncio.wait(tasks))
    loop.close()

In the example I want to kick off three worker threads with different strings and wait for them to finish. Workers sleep random amount of time so I expect them to finish in random order when script is run multiple times. Turns out that they seem to execute sequentially instead, second thread starting after first.

What is my error here? Shouldn't sleep block only the thread it is in? Is my event loop set up correctly? Can I async/await joins?

Ultimately I want to send messages to other threads and wait for their response to then run a callback function with returned value.

EDIT: To clarify, ultimately I want to wait for conditional variables with async/await in my main thread and run other code until some of the conditional variables let execution through. In this example code I was trying to do the same with worker thread's join.

Answer 1

Ultimately, it is running sequentially because of this code:

async def send(message):
    t = MyThread(message)  # daemon = True
    t.start()
    print("asd")
    return t.join()

You start a thread, then immediately wait on that thread to finish before you continue. This is why they're executed sequentially.

Node.js and asyncio do not necessarily create new threads to execute their operations. Node.js for example only ever uses a single thread, but it uses kernel level functions ( for example 'epoll' ) to call the callbacks that you indicate when some new network activity occurred. This allows a single thread to manage hundreds of network connections.

That's why probably when you executed this without the Thread instance, you'd call sleep on the currently running thread, which is the same as the main thread. When you use asyncio with networking functions, then you can use "yield from" structures, which allows other code blocks to execute while the other tasks are doing things with other remote services.

The main structure is correct. You want this block of code:

loop.run_until_complete(asyncio.wait(tasks))

But don't rely on 'sleep' to test the function, you need to make a networking call, or use:

yield from asyncio.sleep(1)

And there's no need to start separate threads in that case.