cocotb provides couple of routines to start concurrent coroutine. The implementation shows that start
calls start_soon
and yield for the forked process to start right now. That’s a big deal because start_soon
doesn’t star the coro until the parent coro yields control (ie await from something).
def start_soon(coro: Union[Task, Coroutine]) -> Task:
"""
Schedule a coroutine to be run concurrently.
Note that this is not an async function,
and the new task will not execute until the calling task yields control.
.. versionadded:: 1.6.0
"""
return scheduler.start_soon(coro)
async def start(coro: Union[Task, Coroutine]) -> Task:
"""
Schedule a coroutine to be run concurrently, then yield control to allow pending tasks to execute.
The calling task will resume execution before control is returned to the simulator.
.. versionadded:: 1.6.0
"""
task = scheduler.start_soon(coro)
await cocotb.triggers.NullTrigger()
return task
Let’s jump into start_soon
into the scheduler
def start_soon(self, coro: Union[Coroutine, Task]) -> Task:
task = self.create_task(coro)
if _debug:
self.log.debug("Queueing a new coroutine %s" % task._coro.__qualname__)
self._queue(task)
return task
create_task
always returns Task
. if passed coroutine, It will return Task created from that coroutine.
@staticmethod
def create_task(coroutine: Any) -> Task:
if isinstance(coroutine, Task):
return coroutine
if isinstance(coroutine, Coroutine):
return Task(coroutine)
_queue
puts the task in _pending_coros
.
def _queue(self, coroutine):
if coroutine not in self._pending_coros:
self._pending_coros.append(coroutine)
_pending_coros
is passed to _schedule
# Handle any newly queued coroutines that need to be scheduled
while self._pending_coros:
task = self._pending_coros.pop(0)
...
self._schedule(task)
in _schedule
, _advance
is called on that Task
with self._task_context(coroutine):
if trigger is None:
send_outcome = outcomes.Value(None)
else:
send_outcome = trigger._outcome
if _debug:
self.log.debug(f"Scheduling with {send_outcome}")
coroutine._trigger = None
result = coroutine._advance(send_outcome)
_advance
is defined with Task
which calls outcome.send
def _advance(self, outcome: outcomes.Outcome) -> typing.Any:
try:
self._started = True
return outcome.send(self._coro)
send
in Value
calls gen.send()
and gen
here is self_coro
class Value(Outcome):
def __init__(self, value):
self.value = value
def send(self, gen):
return gen.send(self.value)