Why we need an Async While?

If you need to run your coroutines and each must be executed independent, like receive stream data from a list of sources, in which every source have different frecuencies, you need this.

TaskLoop provides a set of functions and asyncio coroutines who can help you on your mission.

You need

fn1 ->->->->->->->-> |

fn2 —>—>—>—> |

. |

. |

. |

fnN –>–>–>–>–>–> |

You can’t do a /for/ and obtain good results. You need free tasks.

So, the model you need is for every source-coroutine is:

The implementation consider:

1. Define coroutine

2. The couroutine need some input?

3. The input is the same ever or change with the result?

4. Create a couroutine mask who await the couroutine and manage the input

5. Call the coro-mask in a task generator, scheduling it.

6. Set a callback when task is done

7. When task is done the callback must be the same task generator

In a schemma, the system is:

TaskLoop State Model

The previous graph explains the callback chain. It is not a state diagram.

TaskLoop also has an explicit control model, evaluated inside renew when the active task finishes:

The important detail is that PAUSE and CONTINUE are control signals for the next callback decision:

1. PAUSE prevents rescheduling the user coroutine and schedules a short pause coroutine instead.

2. CONTINUE resumes the user coroutine with the last payload that was produced before the pause loop.

3. STOP ends the callback chain after the current task completes.

4. CANCEL cancels the current task immediately.

The methods or functions available for this schema are:

Function	Inputs	Do
coromask	[coro, args, kwargs, fargs] | await coro
renew	[task, coro, fargs]	renew task
simple_fargs	[input, output]	return input
simple_fargs_out	[input, output]	return output

The following code is an example about how to use the taskloop methods, we have two coroutines, define a fargs for every coroutine, and use the methods on a asyncio event loop.

import asyncio
import functools

async def holacoro(v):
    print("Hola %d" % v)
    await asyncio.sleep(1)
    return v+1

async def sumacoro(*args):
    c=sum(args)
    print("La suma es %d" %c)
    await asyncio.sleep(3)
    return c

def fargs_holacoro(args, obtained):
    return [obtained]

def fargs_sumacoro(args, obtained):
    result= [args[-1], obtained]
    return result

#every coroutine
async def coromask(coro, args, kwargs, fargs):
     _in=args
     obtained=await coro(*args, **kwargs)
     result=fargs(_in, obtained)
     return result

def renew(task, coro, fargs):
    result_args, result_kwargs = task.result()
    task=loop.create_task(coromask(coro, result_args, result_kwargs, fargs))
    task.add_done_callback(functools.partial(renew, coro=coro, fargs=fargs))

loop=asyncio.get_event_loop()

args=[1]
task1=loop.create_task(coromask(holacoro,args,{},fargs_holacoro))
task1.add_done_callback(functools.partial(renew, coro=holacoro, fargs=fargs_holacoro))

args2=[1,2]
task2=loop.create_task(coromask(sumacoro,args2,{},fargs_sumacoro))
task2.add_done_callback(functools.partial(renew, coro=sumacoro, fargs=fargs_sumacoro))


try:
     loop.run_forever()
except KeyboardInterrupt:
     print('Loop stopped')

Now, those async functions are encapsulated on the Tasktools module inside the taskloop file. Hence, to call them you have to use the tradicional python system to use modules and functions.

from tasktools.taskloop import coromask, renew, simple_fargs