How to avoid recursive async() from hanging.

[bangerth]

I must be missing something fundamental about how to make tf::Executor::async() work in actual practice. In the following little program, I'm telling the executor that it can use 2 threads, and then I have three levels of tasks I create with async(). Each time I create a task, I wait for it:

#include <taskflow/algorithm/for_each.hpp>
#include <taskflow/taskflow.hpp>

#include <iostream>
#include <thread>


tf::Executor executor (2);   // executor having two threads


void bottom()
{
  std::cout << "      Starting task at the bottom\n";
  std::cout << "        ... ... ...\n";
  std::this_thread::sleep_for(std::chrono::seconds(1));
  std::cout << "      Ending task at the bottom\n";
}

void middle()
{
  std::cout << "    Starting task in the middle\n";
  auto t = executor.async([]() { bottom(); } );
  std::cout << "    Waiting for sub-task in the middle\n";
  t.wait();
  std::cout << "    Ending task in the middle\n";
}

void top()
{
  std::cout << "  Starting task at the top\n";
  auto t = executor.async([]() { middle(); } );
  std::cout << "  Waiting for sub-task at the top\n";
  t.wait();
  std::cout << "  Ending task at the top\n";
}


int
main()
{
  std::cout << "Starting task in main()\n";
  auto t = executor.async([]() { top(); } );
  std::cout << "Waiting for task in main\n";
  t.wait();
  std::cout << "Done in main\n";
}

This program, perhaps unsurprisingly hangs:

Starting task in main()
Waiting for task in main
  Starting task at the top
  Waiting for sub-task at the top
    Starting task in the middle
    Waiting for sub-task in the middle
^C

The reason, of course, is that we create a task in main(), which creates a task in top(), which gets us into middle() which queues up a third task, but because we already have two tasks running, that third task never runs, and so middle() never stops waiting.

This is not surprising. executor.async() returns a std::future object, and waiting for it does not communicate to the executor that the current task isn't doing anything and that now would be a good time to run other tasks.

The question is, then, how one is supposed to use async() in contexts in which one has no control over the number of recursive invocations of async()?

[bangerth]

For reference, what we're trying to do is convert the deal.II (https://www.dealii.org) library from using the TBB to Taskflow. The async() function is one we had previously requested in #172.

[bangerth]

For the record, I also tried to use the tf::Runtime approach shown in https://taskflow.github.io/taskflow/AsyncTasking.html#LaunchAsynchronousTasksFromARuntime . This leads to the following program, which alas also deadlocks:

#include <taskflow/algorithm/for_each.hpp>
#include <taskflow/taskflow.hpp>

#include <iostream>
#include <thread>


tf::Executor executor (2);   // executor having two threads


void bottom()
{
  std::cout << "      Starting task at the bottom\n";
  std::cout << "        ... ... ...\n";
  std::this_thread::sleep_for(std::chrono::seconds(1));
  std::cout << "      Ending task at the bottom\n";
}

void middle()
{
  std::cout << "    Starting task in the middle\n";
  tf::Taskflow taskflow;
  taskflow.emplace([] (tf::Runtime& rt){ rt.async([]() { bottom(); }); rt.corun_all(); } );
  auto t = executor.run(taskflow);
  std::cout << "    Waiting for sub-task in the middle\n";
  t.wait();
  std::cout << "    Ending task in the middle\n";
}

void top()
{
  std::cout << "  Starting task at the top\n";
  tf::Taskflow taskflow;
  taskflow.emplace([] (tf::Runtime& rt){ rt.async([]() { middle(); }); rt.corun_all(); } );
  auto t = executor.run(taskflow);
  std::cout << "  Waiting for sub-task at the top\n";
  t.wait();
  std::cout << "  Ending task at the top\n";
}


int
main()
{
  std::cout << "Starting task in main()\n";
  tf::Taskflow taskflow;
  taskflow.emplace([] (tf::Runtime& rt){ rt.async([]() { top(); }); rt.corun_all(); } );
  auto t = executor.run(taskflow);
  std::cout << "Waiting for task in main\n";
  t.wait();
  std::cout << "Done in main\n";
}

[makeuptransfer]

try this way

#include <taskflow/algorithm/for_each.hpp>
#include <taskflow/taskflow.hpp>

#include <iostream>
#include <thread>


tf::Executor executor (2);   // executor having two threads


void bottom()
{
  std::cout << "      Starting task at the bottom\n";
  std::cout << "        ... ... ...\n";
  std::this_thread::sleep_for(std::chrono::seconds(1));
  std::cout << "      Ending task at the bottom\n";
}

void middle(tf::Runtime& rt)
{
  std::cout << "    Starting task in the middle\n";
  rt.silent_async([]() { bottom(); } );
  std::cout << "    Waiting for sub-task in the middle\n";
  rt.corun_all();
  std::cout << "    Ending task in the middle\n";
}

void top(tf::Runtime& rt)
{
  std::cout << "  Starting task at the top\n";
  rt.silent_async([](tf::Runtime& rt) { middle(rt); } );
  std::cout << "  Waiting for sub-task at the top\n";
  rt.corun_all();
  std::cout << "  Ending task at the top\n";
}


int
main()
{
  std::cout << "Starting task in main()\n";
  executor.silent_async([](tf::Runtime& rt) { top(rt); } );
  std::cout << "Waiting for task in main\n";
  executor.wait_for_all();
  std::cout << "Done in main\n";
}

[bangerth]

@makeuptransfer I have not quite been able to find out what rt.corun_all() actually waits for. Is it the single task added by rt.silent_async(), or is the rt object shared between multiple contexts? In my code example, each of the places where I'm waiting should only wait for the one task emplaced two lines higher up, not any tasks that may have been emplaced anywhere else. (Think of running the top() function 100 times in parallel; each wait call should only wait for the task emplaced by the current invocation of the function, not the tasks emplaced by the other 99 concurrently running instances of top().)

In other words, the question comes down to where the rt lives. Is it specific to each task, or is there only a single runtime object in your example?

[makeuptransfer]

@bangerth, In my opinion, a runtime is tied to a node, when you silent_async a task(node) using executor, it will +1 to the whole executor pending_cnt, but when you silent_async a task using runtime, it will add the pending_cnt of rt's parent(which refers the task who silent_async this task, e.g. top 's rt silent_async mid will add top's pending). the runtime corun_all method will only wait the task it silent_async so it will not block in this case.

[tsung-wei-huang]

Hi @bangerth , yes, tf::runtime allows a task to acquire the handle to the running node, worker, and executor. The concept of corun means that the caller worker will not block but enter the work-stealing loop to co-run with other workers. This is primarily used in your scenarios when you would like to block.

tf::Executor::async is a drop-in replacement of std::async - which is convenient but also inherits its limitation such as the blocking wait. If you would like to avoid this blocking wait while allowing the worker to corun other tasks, you may consider the following example with corun_until:

tf::Executor executor(1);

auto fu1 = executor.async([](){
  // ... do some stuff
});

auto fu2 = executor.async([&](){
  executor.corun_until([&](){          // the worker will not block until fu1 completes but joins the work-stealing loop to run tasks until fu1 becomes ready
    return fu1.wait_for(std::chrono::seconds(0)) == future_status::ready;
  });
});

Does this make sense to you?

[bangerth]

@makeuptransfer and @tsung-wei-huang Thank you for your suggestions. I believe that the corun_until() is what I will need. I will try that out later today.

In the end, the kind of semantics I look for are basically like this:

  auto t = async( []() {...} );       // create a task, have it scheduled to be run at some point in the future
  ...do some other work...
  t.join();                                    // wait for the task to finish

In order for this to work without deadlocks, the t.join() call must inform the scheduler that the current task is suspended and that it should execute other tasks. This will eventually also take care of task t. Of course, in the meantime many other tasks may have been processed (other than t), and that's ok.

@tsung-wei-huang I believe that your solution works for this. It makes sure that we return as soon as t is finished, rather than only when the scheduler's queue has become empty (which I think is what happens when you just run corun_all()).

Out of curiosity, you wrote the second half of the code as a separate task:

auto fu2 = executor.async([&](){
  executor.corun_until([&](){          // the worker will not block until fu1 completes but joins the work-stealing loop to run tasks until fu1 becomes ready
    return fu1.wait_for(std::chrono::seconds(0)) == future_status::ready;
  });
});

Why not just the following?

executor.corun_until([&](){          // the worker will not block until fu1 completes but joins the work-stealing loop to run tasks until fu1 becomes ready
    return fu1.wait_for(std::chrono::seconds(0)) == future_status::ready;
  });

[tsung-wei-huang]

Hi @bangerth, corun_until can only be called from an internal worker of an executor, so that worker can corun tasks with other workers in that executor. Does that make sense :) ?

[bangerth]

Hm, but then we're back to a chicken and egg problem: If I can only corun from within that second task, how do I make the outer task wait for the first task? In your example, I'd have to wait for fu2, but that will deadlock again for the same reason as I originally had.

[tsung-wei-huang]

Yes, ultimately something needs to synchronize it. If you have complicated dependencies around async tasks like this, I would recommend using Dependent Async, which allows you to do more fine-grained dependency building around dynamic task graphs. Have you ever looked into this?

[bangerth]

But I don't actually have complicated dependencies. Waiting for a previously submitted sub-task is not an uncommon operation. I believe that I must be misunderstanding something conceptual because I am certain that you have run into this case before. I just don't understand how you have solved it :-)

The key piece is you can't use std::future::wait() (or std::future::get()) to wait for an async task because you're just blocking the current task without giving the scheduler room to execute something else in its stead. That's what leads to the deadlock. So in the example you show on the "Dependent Async" page you link to, the call to fuD.get() is going to lead to deadlocks if enough tasks are at that place at the same time. The call to executor.wait_for_all(); could work, though.

Do I read it right, then, that executor.corun_until(...) can only be called from a task that is run by said executor (which is how I understand why you had to put it into a sub-task), whereas executor.wait_for_all() can also be called from the context that has created/owns the executor?

[tsung-wei-huang]

Hi @bangerth , your understanding is correct - executor.corun_until(...) can only be called from a task that is run by an executor. executor.corun_all is behaving similar to executor.wait_for_all except corun_all doesn't block the caller (which must be an worker from the calling executor).

Yes, I was just giving a hint if you have a complicated task graph that needs to be created on the fly, dependent_async might be help. However, in your example, it is a chicken egg problem. I don't think there is any easy solution (?) - someone needs to synchronize on the task eventually. Any thought?

[makeuptransfer]

@bangerth ,i dont get it, there are two threads (one from main, one from executor), why wait fu2 will block?

[bangerth]

@makeuptransfer Try it out :-) The executor allows for at most 2 tasks to run concurrently, but I'm recursively creating 3 tasks, so the innermost task will simply never be scheduled because the outer two are still running, and they will never finish because they are waiting for the innermost one to complete.

[bangerth]

@tsung-wei-huang Let me start by saying thank you for your patience with my question. Your feedback is much appreciated!

I played with this some more. First, this approach with wait_for_all() works:

void bottom()
{
  std::cout << "      Starting task at the bottom\n";
  std::cout << "        ... ... ...\n";
  std::this_thread::sleep_for(std::chrono::seconds(1));
  std::cout << "      Ending task at the bottom\n";
}

void middle()
{
  std::cout << "    Starting task in the middle\n";
  tf::Executor executor(1);
  executor.async([] () { bottom(); });
  std::cout << "    Waiting for sub-task in the middle\n";
  executor.wait_for_all();
  std::cout << "    Ending task in the middle\n";
}

void top()
{
  std::cout << "  Starting task at the top\n";
  tf::Executor executor(1);
  executor.async([] () { middle(); });
  std::cout << "  Waiting for sub-task at the top\n";
  executor.wait_for_all();
  std::cout << "  Ending task at the top\n";
}


int
main()
{
  std::cout << "Starting task in main()\n";
  tf::Executor executor(1);
  executor.async([] () { top(); });
  std::cout << "Waiting for task in main\n";
  executor.wait_for_all();
  std::cout << "Done in main\n";
}

But this is not what one wants. It creates an executor(1) at every scope, so every new task creates its own thread. There is no global control on the number of tasks that are running at the same time.

So we need a global executor object. This suggests the following code, but (see below) this does not work:

tf::Executor executor (2);   // executor having two threads


void bottom()
{
  std::cout << "      Starting task at the bottom\n";
  std::cout << "        ... ... ...\n";
  std::this_thread::sleep_for(std::chrono::seconds(1));
  std::cout << "      Ending task at the bottom\n";
}

void middle()
{
  std::cout << "    Starting task in the middle\n";
  executor.async([] () { bottom(); });
  std::cout << "    Waiting for sub-task in the middle\n";
  executor.wait_for_all();
  std::cout << "    Ending task in the middle\n";
}

void top()
{
  std::cout << "  Starting task at the top\n";
  executor.async([] () { middle(); });
  std::cout << "  Waiting for sub-task at the top\n";
  executor.wait_for_all();
  std::cout << "  Ending task at the top\n";
}


int
main()
{
  std::cout << "Starting task in main()\n";
  executor.async([] () { top(); });
  std::cout << "Waiting for task in main\n";
  executor.wait_for_all();
  std::cout << "Done in main\n";
}

This deadlocks (of course) because the executor.wait_for_all() call in middle() also waits for the tasks that are running middle() and top(), and these can of course not finish before wait_for_all() is finished.

---

I then went through the list of members of tf::Executor again. What I need is, conceptually, something of the sort

  auto t = executor.async(...);    // or executor.run(), or similar -- something that spawns jobs
  executor.wait_for(t);                // waits for completion of 't'

The second line needs to be something that references executor because it needs to allow the executor to work on scheduled tasks until t is known to have completed. It cannot be std::future::join() for the reasons discussed in previous comments.

The functions of tf::Executor that one could base the wait() functionality on are:

1. executor.run_until ( /* empty TaskFlow */, ...check that std::future returned by executor.async() is ready ...). That's perhaps wasteful, given that it keeps running empty task graphs, but at least conceptually does what we need. I'll show the code below, but it segfaults. (I think I also misunderstood that run_until() blocks while it is running the flows; in hindsight, I think it just schedules them with a conditional back-arc task from the end of the flow back to the beginning. That doesn't help, because I needed something that waits.)
2. executor.corun_until(...check that std::future returned by executor.async() is ready ...). This is actually exactly what I want (and what you described above). It also actually works :-) The key issue, though, is that it is documented that you can only call this from a task that is currently running under executor. So you can call it in top() and middle(), but you shouldn't be able to call from main() -- though that seems to be working nonetheless.

Implementations of these approaches would look like this:

1. Using executor.run_until():

tf::Executor executor (2);   // executor having two threads


void bottom()
{
  std::cout << "      Starting task at the bottom\n";
  std::cout << "        ... ... ...\n";
  std::this_thread::sleep_for(std::chrono::seconds(1));
  std::cout << "      Ending task at the bottom\n";
}

void middle()
{
  std::cout << "    Starting task in the middle\n";
  auto t = executor.async([] () { bottom(); });
  std::cout << "    Waiting for sub-task in the middle\n";

  tf::Taskflow flow;
  flow.emplace([](){});
  executor.run_until (flow, [&t](){ return (t.wait_for(std::chrono::seconds(0)) == std::future_status::ready); });
  std::cout << "    Ending task in the middle\n";
}

void top()
{
  std::cout << "  Starting task at the top\n";
  auto t = executor.async([] () { middle(); });
  std::cout << "  Waiting for sub-task at the top\n";

  tf::Taskflow flow;
  flow.emplace([](){});
  executor.run_until (flow, [&t](){ return (t.wait_for(std::chrono::seconds(0)) == std::future_status::ready); });
  std::cout << "  Ending task at the top\n";
}


int
main()
{
  try
    {
      std::cout << "Starting task in main()\n";
      auto t = executor.async([] () { top(); });
      std::cout << "Waiting for task in main\n";

      tf::Taskflow flow;
      flow.emplace([](){});
      executor.run_until (flow, [&t](){ return (t.wait_for(std::chrono::seconds(0)) == std::future_status::ready); });
      std::cout << "Done in main\n";
    }

  catch (std::exception &e)
    {
      std::cout << "Exception: " << e.what() << std::endl;
    }
}

Unfortunately, this segfaults in the very first invokation of the wait operation in main(). I don't actually understand why this is so, but it may be because I'm calling run_until() in main() from a context that is not actually a child-task of the executor, as is documented in https://taskflow.github.io/taskflow/classtf_1_1Executor.html#a0f52e9dd64b65aba32ca0e13c1ed300a. But using the version that takes a moved taskflow object does not help either. But as mentioned above, I think that I also misunderstood what the function does.

2. Using corun_until():

tf::Executor executor (2);   // executor having two threads


void bottom()
{
  std::cout << "      Starting task at the bottom\n";
  std::cout << "        ... ... ...\n";
  std::this_thread::sleep_for(std::chrono::seconds(1));
  std::cout << "      Ending task at the bottom\n";
}

void middle()
{
  std::cout << "    Starting task in the middle\n";
  auto t = executor.async([] () { bottom(); });
  std::cout << "    Waiting for sub-task in the middle\n";

  executor.corun_until ([&t](){ return (t.wait_for(std::chrono::seconds(0)) == std::future_status::ready); });
  std::cout << "    Ending task in the middle\n";
}

void top()
{
  std::cout << "  Starting task at the top\n";
  auto t = executor.async([] () { middle(); });
  std::cout << "  Waiting for sub-task at the top\n";

  executor.corun_until ([&t](){ return (t.wait_for(std::chrono::seconds(0)) == std::future_status::ready); });
  std::cout << "  Ending task at the top\n";
}


int
main()
{
  try
    {
      std::cout << "Starting task in main()\n";
      auto t = executor.async([] () { top(); });
      std::cout << "Waiting for task in main\n";

      executor.corun_until ([&t](){ return (t.wait_for(std::chrono::seconds(0)) == std::future_status::ready); });
      std::cout << "Done in main\n";
    }
  catch (std::exception &e)
    {
      std::cout << "Exception: " << e.what() << std::endl;
    }
}

As mentioned, this actually works even though the call to corun_until() in main() should not be working according to https://taskflow.github.io/taskflow/classtf_1_1Executor.html#a0fc6eb19f168dc4a9cd0a7c6187c1d2d .

[bangerth]

So I'm afraid that in truth, I'm not really any further towards a solution than I was before. Two key observations:

• executor.corun_until(....) almost works. It is functionally what I need, namely executor.keep_scheduling_until_t_is_done(t). The reason why it does not quite work is the documented restriction that it can only be called from a dependent task of executor, though the second long code above shows that that is apparently not enforced.
• To give context, what I want to replace in the deal.II software library is calls to the TBB that in essence look like this:

  tbb::task_group tg;
  tg.run ( []() {...} );                   // implicitly uses a global 'executor' object
  ... do something else...
  tg.wait();                                // waits for all tasks in 'tg', but keeps scheduling tasks in the global executor

tg.wait() (see https://spec.oneapi.io/versions/1.1-rev-1/elements/oneTBB/source/task_scheduler/task_group/task_group_cls.html) is basically like tf::Executor::corun_until(...) except without the restriction of where it can be called.

[makeuptransfer]

tf::Executor executor (2);   // executor having two threads
void bottom()
{
  std::cout << "      Starting task at the bottom\n";
  std::cout << "        ... ... ...\n";
  std::this_thread::sleep_for(std::chrono::seconds(1));
  std::cout << "      Ending task at the bottom\n";
}

void middle()
{
  std::cout << "    Starting task in the middle\n";
  auto t = executor.async([] () { bottom(); });
  std::cout << "    Waiting for sub-task in the middle\n";

  executor.corun_until ([&t](){ return (t.wait_for(std::chrono::seconds(0)) == std::future_status::ready); });
  std::cout << "    Ending task in the middle\n";
}

void top()
{
  std::cout << "  Starting task at the top\n";
  auto t = executor.async([] () { middle(); });
  std::cout << "  Waiting for sub-task at the top\n";

  executor.corun_until ([&t](){ return (t.wait_for(std::chrono::seconds(0)) == std::future_status::ready); });
  std::cout << "  Ending task at the top\n";
}
int
main()
{
  try
    {
      std::cout << "Starting task in main()\n";
      auto t = executor.async([] () { top(); });
      std::cout << "Waiting for task in main\n";

      // executor.corun_until ([&t](){ return (t.wait_for(std::chrono::seconds(0)) == std::future_status::ready); });
      // just wait_for_all()
      executor.wait_for_all();
      std::cout << "Done in main\n";
    }
  catch (std::exception &e)
    {
      std::cout << "Exception: " << e.what() << std::endl;
    }
}

Hi, @bangerth, why not wait_for_all in the main ? Totally there are three threads in total, on main thread, two executor threads, the exexcutor threads will never block when you call corun_until in the task.

[bangerth]

@makeuptransfer I don't want to wait for all tasks. In the example, there are no others around, but in other contexts I may have started more asynchronous tasks. I want to wait for a specific one.

[makeuptransfer]

@bangerth thx, i get it. it seems like you need let the tasks after the "specific one" scheduled by the "specific one" , or use Dependent Async,will they help?

[bangerth]

I think I have finally figured it out. My solution 2 above was the right approach, but suffered from the fact that one can only call corun_until() from a worker thread. The solution I had above also called it from main(), which is not a worker thread (or at least isn't guaranteed to be), but I don't have to do it that way because I can figure out whether we are on a worker thread or not. The following, then, finally does what I want I believe:

tf::Executor executor (2);   // executor having two threads


template <typename T>
void wait_for_task (tf::Executor &executor,
                    std::future<T> &future)
{
  // We want to call executor.corun_until() to keep scheduling tasks
  // until the task we are waiting for has actually finished. The
  // problem is that TaskFlow documents that you can only call
  // corun_until() on a worker of the executor. In other words, we can
  // call it from *inside* other tasks, but not from the main thread
  // (or other threads that might have been created outside of
  // TaskFlow).
  //
  // Fortunately, we can check whether we are on a worker thread:
  if (executor.this_worker_id() >= 0)
    executor.corun_until ([&future](){ return (future.wait_for(std::chrono::seconds(0)) ==
                                               std::future_status::ready); });
  else
    // We are on a thread not managed by TaskFlow. In that case, we
    // can simply stop the current thread to wait for the task to
    // finish (i.e., for the std::future object to become ready). We
    // can do this because we need not fear that this leads to a
    // deadlock: The current threads is waiting for completion of a
    // task that is running on a completely different set of threads,
    // and so not making any progress here can not deprive these other
    // threads of the ability to schedule their tasks.
    //
    // Indeed, this is even true if the current thread is a worker of
    // one executor and we are waiting for a task running on a
    // different executor: The current task being stopped may block
    // the current executor from scheduling more tasks, but it is
    // unrelated to the tasks of the scheduler for which we are
    // waiting for something, and so that other executor will
    // eventually get arond to scheduling the task we are waiting for,
    // at which point the current task will also complete.
    future.wait();
}

  


void bottom()
{
  std::cout << "      Starting task at the bottom\n";
  std::cout << "        ... ... ...\n";
  std::this_thread::sleep_for(std::chrono::seconds(1));
  std::cout << "      Ending task at the bottom\n";
}

void middle()
{
  std::cout << "    Starting task in the middle\n";
  auto t = executor.async([] () { bottom(); });
  std::cout << "    Waiting for sub-task in the middle\n";

  wait_for_task(executor, t);
  std::cout << "    Ending task in the middle\n";
}

void top()
{
  std::cout << "  Starting task at the top\n";
  auto t = executor.async([] () { middle(); });
  std::cout << "  Waiting for sub-task at the top\n";

  wait_for_task(executor, t);
  std::cout << "  Ending task at the top\n";
}


int
main()
{
  try
    {
      std::cout << "Starting task in main()\n";
      auto t = executor.async([] () { top(); });
      std::cout << "Waiting for task in main\n";

      wait_for_task(executor, t);
      std::cout << "Done in main\n";
    }
  catch (std::exception &e)
    {
      std::cout << "Exception: " << e.what() << std::endl;
    }
}

[tsung-wei-huang]

@bangerth , yes, your understanding about the execution logic and wait_for_task are correct. Sorry I was a bit late in response due to the final week of the semester.

[bangerth]

I put that into deal.II in dealii/dealii#16976 and it runs successfully with all 13,000 of our tests. So this seems to work as intended :-)

Thanks for your help!