Continue child task execution after parent exception for wait and wait_any types

[tribal-tec]

As wait and wait_any don't take parents' results, execution should continue. This
allows for a parent.get() in the child task to also handle error/exception cases.

[bloomen]

That seems like a reasonable change. So before this change a child is never run if a parent emits an exception. But after this change the child would continue but only for the wait* types.

Doesn't it make sense that the children do not get run if there's an error in a parent? Also, the exception can't propagate through to all children anymore after this change.

Do you have a specific use case in mind?

Cheers!

[tribal-tec]

My current use case probably is a bit of a misuse of tasks :)

Instead of doing .get() on the 'leaf' of a task chain, I want to call a callback from within its operator(). Reason is a .get() would block the caller which I don't want for an asynchronous processing (handler for RPCs from websocket clients). I just want to callback after execution has finished.

So I do a .get() of the parent task in the task-which-does-callback to get it's result, or now an exception.

Another option would be to extend consume* to also pass exceptions, but I haven't seen yet how to accomplish this.

Let me know if this makes sense to you and if you have maybe another approach how to solve this problem.

Thanks!

Edit: Just found this, which is exactly what I was aiming for: https://github.com/Amanieu/asyncplusplus/wiki/Tasks#exceptions, task-based continuations. I'm basically faking this now by remember the parent task in the child functor, and .get() there to get the value or the exception and then callback accordingly.

[christianblume-serato]

So my rough understanding of your problem is that, in essence, you want to execute a wait task regardless of whether the parent threw an exception or not. Is this correct?

I think this would make a lot of sense and sounds like a valid use case.

I suppose it would be nice to then also give access to the exception that was thrown in the parent task (similar to how async++ does it)?

But what should happen if a consume task consumes output from a wait task and the parent of the wait task threw an exception? Does the wait task act as a guard for exceptions and from there on it's not propagated further?

Without fully understanding your use case, you might want to try executors to address your asynchronous problem (which seems somewhat unrelated to the exception problem?).

[christianblume-serato]

I think that once the wait task has access to the exception object it itself can decide whether to propagate further or not.

[tribal-tec]

The approach that async++ did wrt having value-consume AND task-consume seems the right approach. So wait() should stay untouched imo, but a new consume type or an extension that passes the task instead the value should suffice. In this case, the consumer will get the exception if any.

So in my use case, the consumer task is of void return type, and will call a success or an error function, depending on the result of the parent task. So the consumer task will never be .get() from anywhere, so after scheduling, the task is just remembered for an eventual cancel(). I'm not sure that this model is fully applicable with your library, but from what I've seen so far, it should be.

[bloomen]

Agreed. There should be a new task type that requires the task's functor to accept objects of std::shared_future<T>. How about these task types then:

enum class task_type {
    root,        // The task has no parents
    accept,      // The task's functor accepts all parent futures
    accept_any,  // The task's functor accepts the first parent future that becomes ready
    consume,     // The task's functor consumes all parent results
    consume_any, // The task's functor consumes the first parent result that becomes ready
    wait,        // The task's functor takes no arguments but waits for all parents to finish
    wait_any,    // The task's functor takes no arguments but waits for the first parent to finish
};

[bloomen]

Check this out when you have a moment: c44179c

[tribal-tec]

Looks great, much appreciated! The new 'value' type is what 'consume' was before? Or shall I use the commit with 'accept' for testing?

I would really like to use your library rather than async++. However, their event task type is also something I need: a external condition to be set when to start/schedule a task. I have to think if there is maybe a way around this though.

[bloomen]

If you could do your testing on the develop branch that'd be great! The new task types interesting for you are accept and accept_any.
The value task type was an experiment that I reverted in the meantime. Didn't work out as I had hoped and the benefits are also quite small.

Can you point me to an example of how async++ uses this event task type?

[tribal-tec]

Ok, I'll give the 'access' types a try, hopefully today.

Here is the event task type: https://github.com/Amanieu/asyncplusplus/wiki/Tasks#event-tasks

The closest today would be to have a shared variable and then a call to schedule(). It's actually a schedule() with a value parameter as the input for the task functor.

[tribal-tec]

The access type works like a charm!

The event task is the only missing thing now for me, for both, setting value and exception to start/continue task execution. I was wondering if this can be already achieved somehow with schedule() each task individually other than schedule_all(). I'll be playing with this right now.

Edit: I noticed something else. Is there a way to have a runtime-list of parents for consume/accept? I hope it's not a constraint by your implementation that the number of parents must be compile-time constant.

[bloomen]

Since parents don't know about their children you can remove children at any time and add new ones. However, you cannot remove a parent of a child without destroying the child first. The graph structure is NOT fixed at compile time.

I had a look at the event task example: You can achieve something very similar with transwarp by changing the input to a root task and then calling schedule_all on the final task. In fact, the basic three-task example of transwarp does exactly that:

    double value1 = 13.3;
    int value2 = 42;

    auto task1 = tw::make_task(tw::root, [&value1] { return value1; });
    auto task2 = tw::make_task(tw::root, [&value2] { return value2; });
    auto task3 = tw::make_task(tw::consume, add_em_up, task1, task2);

    tw::parallel executor{4};

    task3->schedule_all(executor);  // schedules all tasks for execution
    std::cout << "result = " << task3->get() << std::endl;  // result = 55.3

    // modifying data input
    value1 += 2.5;
    value2 += 1;

    task3->schedule_all(executor);  // re-schedules all tasks for execution
    std::cout << "result = " << task3->get() << std::endl;  // result = 58.8

Now, if you wanted you could put 'modifying data input' and scheduling into the same function.

You can also cancel tasks by calling cancel() or cancel_all() on the task object. If your task functor inherits from transwarp::functor you can even cancel tasks while they're running.

I will think about having an explicit way of setting a value or an exception to a task object. I don't see a reason why that shouldn't work. I'll have a stab at it.

[tribal-tec]

The modification of the functor from outside works like you described. It's just a bit cumbersome to make this work with values and exceptions, plus automatically schedule (or wait if scheduled already) once the external input is set.

On the runtime list of parents for a task I'm afraid this won't work. Let me elaborate by code:

tw::parallel executor{4};
std::vector<std::shared_ptr<tw::task<std::string>>> dataAvailableTasks;
std::vector<std::string> datas;
std::vector<std::shared_ptr<tw::task<void>>> tasks;

for (size_t i = 0; i < numDatas; ++i)
{
    datas.push_back({});
    dataAvailableTasks.push_back(tw::make_task(tw::root, [&val = datas[i]] {
        return val;
    }));
    tasks.push_back(tw::make_task(
        tw::consume, [](std::string data){ /* do something */ },
        dataAvailableTasks[i]));
}

_task = tw::make_task(tw::accept,
                      [](std::shared_future<void> result) {
                          result.get(); // potential exception is propagated to caller
                          return true;
                      }, tasks); // how to pass a list of parents here?
_task->schedule(executor);

// Later in time, when first data is available
datas[0] = "foo";
dataAvailableTasks[0]->schedule(executor);
tasks[0]->schedule(executor);

// TODO: set exception if for instance next data was broken, incomplete, etc.

I guess it could work by creating a graph where the next task also accepts the previous task, but I can't store those tasks uniformly as the task type differs if I have 1 or 2 parents.

[bloomen]

Yeah, I am afraid that this is not supported by transwarp. All parent types have to be known on construction of a child task. I think you'd want to go for a library that allows for more dynamic handling of dependencies.

[tribal-tec]

Yeah, I'm staying with async++ for the time being. A big thanks to you nonetheless for all the efforts! Great library, fitting just in one file!

I close this PR, as your new accept types solve exactly that problem much better.

[bloomen]

set_value/exception and remove_value/exception are now on the develop branch. If you have time please check it out and let me know if this interface works for you. Thanks!

[tribal-tec]

The API definitely works for me, look good! Unfortunately, I can't test it with my code right now as it has evolved too far away now to use transwarp easily again. I might come back to this though in the future.