Support repeatedly calling WhenAny on a progressively smaller list of already scheduled tasks

[davidmrdavid]

Fixes: #349

Today, we following orchestrator code throws an exception.

def orchestrator_function(context: df.DurableOrchestrationContext):
    #  Create some tasks
    task1_until = context.current_utc_datetime + datetime.timedelta(seconds=30)
    task1 = context.create_timer(task1_until)
    task2_until = context.current_utc_datetime + datetime.timedelta(seconds=15)
    task2 = context.create_timer(task2_until)
    pending_tasks = [task1, task2]

    #  Yield until first task is completed
    finished_task1 = yield context.task_any(pending_tasks)

    #  Remove completed task from pending tasks
    pending_tasks.remove(finished_task1)

    #  Yield remaining task
    yield context.task_any(pending_tasks)

This exception comes from the incorrect assumption that given list of sub-tasks will not be yield'ed multiple times on different parent tasks. In this case, we have two WhenAny tasks that both operate on a shared subset of the pending_tasks list.

This PR addresses this by ensuring that the second WhenAny only schedules new actions for the set of sub-tasks that have not been already scheduled. Put more simply: we keep track of whether or not an action for a given task was already generated and, if so, we ignore it future attempts at scheduling it.

[nytian]

Thanks for the explanations and LGTM!