Asyncio in Python is deceptively intuitive, and with its pseudo real time / event processing /single threaded nature allows most features of concurrent programming while avoiding or hiding most of the complexities associated with multi-threaded/multi-processing programming.  In this article the versatility of asyncio program is examined by looking at a very specific problem: Assume that there is a task list and each task is being worked on.  When one or more tasks complete, another task or tasks should start. This is a very typical problem where there is a limit on the number of concurrent tasks in execution, but the number of tasks are not. Here is a contemporary example: Imagine the Covid 19 vaccine administration in a Parking lot.  Four vaccination stations have been set up. Each station administers a vaccine shot which takes between 1-3 minutes. The task is to complete boxes of vaccines with each box having 100 doses. Figure 1: Schematic of the parking lot vaccine administration Let us see how Python asyncio will administer the shots! Before you get too eager, let me reveal and briefly discuss the real title of this article: ' '. Dynamic addition of new tasks to a list of awaited tasks : A list of tasks are being executed. Situations will arise where you want to add another task to the list dynamically. Using Python asyncio we write this problem in pseudo code: Problem description True:
    #a tasklist t1,t2,t3 is being executed 
    asyncio.wait(t1(),t2(),t3(),return_when=asyncio.FIRST_COMPLETED) 
    <some tasks completes and execution gets here..
    Now we want another task or run the same task again into the list> while of To solve this problem, we maintain a dict of the running tasks, and insert new tasks into the list as tasks get completed. This is coded below: #Create a dict running tasks. These are { :t1, :t2, :t3} #

    #start the tasks to run concurrently. Here the functions are called or initiated
    tasks={ :asyncio.create_task(v(),name=k ) k,v taskDict.items()}
    #wait one or more tasks to complete and insert tasks into the dict True:
        done,pending = asyncio.wait(tasks.values(),return_when=asyncio.FIRST_COMPLETED) t done:  #we are only concerned about the tasks done. Only these need to be restarted not t.exception():
                print(f )
                tasks[tNm]=asyncio.create_task(taskDict[tNm](),name=tNm) :  #some exception occurred during task execution and has to be resolved
                    print(f )
                    #recover/restart the task
                    tNm=t.get_name()
                    tasks[tNm]=asyncio.create_task(taskDict[tNm](),name=tNm) of . = function names only and hence they will not be called but a list is maintained runningTasks 't1' 't2' 't3' async k for in for new while await for in if 'Task {t.get_name()} is done! Starting the Next Run of the task..' else 'Task {t.get_name()} got exception {t.exception()}. Please take care of it!' I recommend reading the above code about 3 times and refer asyncio docs, especially if you are not familiar with the the asyncio functions used above. Now back to the problem of Vaccination. The above 'Dynamic addition of new tasks to a list of awaited tasks' is used to manage the allocation of vaccines to each station. This in in the async procedure  dynamicTaskList. The vaccination station is simulated by the async procedure .  Each time a vaccinateStation uses up a box of vaccines, it returns, and then is restarted with a new box of vaccines. vaccinateStation random, asyncio,time def vaccinateStation():
    #Task that administers vaccine one by one a box vaccines
    boxCapacity= #capacity truck
    vaccinesLeft=boxCapacity #initially all are left to administer True: asyncio.sleep(random.uniform( , )) # to seconds each shot
        vaccinesLeft -= vaccinesLeft == : #box is now empty boxCapacity # number vaccines administered def dynamicTaskList():
    taskDict={ :vaccinateStation, :vaccinateStation, :vaccinateStation} # stations
    #start the tasks. Here the functions are called or initiated
    tasks={ :asyncio.create_task(v(),name=k ) k,v taskDict.items()}
    # boxes Vaccines
    noOfVaccineBoxes= #time tracker
    timeNow=time.time() i range(noOfVaccineBoxes):
        # the tasks
        done,pending = asyncio.wait(tasks.values(),return_when=asyncio.FIRST_COMPLETED) t done:  #we are only concerned about the tasks done. Only these need to be restarted
            tNm=t.get_name() t.exception():
                    print(f )
                    #recover/restart the task
                    tasks[tNm]=asyncio.create_task(taskDict[tNm](),name=tNm) :
                print(f )
                tasks[tNm]=asyncio.create_task(taskDict[tNm](),name=tNm)

asyncio.run(dynamicTaskList()) import async from of 100 100 of while await 0.001 0.1 0.001 0.1 for 1 if 0 break return return of async 'b1' 'b2' 'b3' 3 async k for in 100 of 100 for in await await for in if 'Task {tNm} got exception {t.exception()}. Please take care of it!' else 'vaccine box {i} is done! Vaccinated  {t.result()} and and time is {(time.time() - timeNow):.2f} seconds. Starting on the Next box..' Running this program gives the following output: (venv) vv@vv-GL63 RD:~ vv3/test$ python dynamicVaccinator.py 
vaccine box is done! Vaccinated and and time is seconds. Starting on the Next box..
vaccine box is done! Vaccinated and and time is seconds. Starting on the Next box..
vaccine box is done! Vaccinated and and time is seconds. Starting on the Next box..
vaccine box is done! Vaccinated and and time is seconds. Starting on the Next box..
vaccine box is done! Vaccinated and and time is seconds. Starting on the Next box.. -8 /git/ 0 100 5.11 1 100 5.26 2 100 5.33 3 100 10.31 4 100 10.37 Here we notice that about 100 vaccinations are being administered in about 5 seconds. We are shooting out vaccines from a machine gun!  The vaccination rate is controlled in line 8: asyncio.sleep(random.uniform( , )) # to seconds each shot await 0.001 0.1 0.001 0.1 for Other parameters like number of vaccines per box are also defined. Try executing this program (written/tested on Python 3.8), change parameters, add slow/fast vaccinateStation's etc. to explore the asyncio features. Please note that if you put in realistic vaccination rates (1-3 minutes), this program will take hours to execute, since it operates in pseudo real-time. Enjoy!

2022 - HackerNoon Contributor of the Year - Html

Nominated for 2022 - HackerNoon Contributor of the Year - Html

Too Long; Didn't Read

In your car, at home, or at work — Bosch technology shapes many areas of life.

Concurrent Task Execution in Python With Asyncio Is A Breeze

Too Long; Didn't Read

[email protected]

About Author

TOPICS

THIS ARTICLE WAS FEATURED IN...

Concurrent Task Execution in Python With Asyncio Is A Breeze

Too Long; Didn't Read

[email protected]

About Author

TOPICS

THIS ARTICLE WAS FEATURED IN...

RELATED STORIES