Flow Graph scalability improvements

[kboyarinov]

Description

Add a comprehensive description of proposed changes

Fixes # - issue number(s) if exists

• - git commit message contains an appropriate signed-off-by string (see CONTRIBUTING.md for details)

Type of change

Choose one or multiple, leave empty if none of the other choices apply

Add a respective label(s) to PR if you have permissions

• bug fix - change that fixes an issue
• new feature - change that adds functionality
• tests - change in tests
• infrastructure - change in infrastructure and CI
• documentation - documentation update

Tests

• added - required for new features and some bug fixes
• not needed

Documentation

• updated in # - add PR number
• needs to be updated
• not needed

Breaks backward compatibility

• Yes
• No
• Unknown

Notify the following users

List users with @ to send notifications

Other information

[aleksei-fedotov]

I believe this can happen even without async_node. Consider the following case:

std::thread t1([]() {
            tbb::task_arena& gta = get_graph_task_arena();
            gta.execute([]() {
                             auto& some_graph_node = get_graph_node();
                             some_graph_node.try_put(/*node message*/1); // spawn of task A
            });
});

t1.join();


// task A starts its execution
graph.wait_for_all();

Don't we support such cases anymore?

[kboyarinov]

In this case it is guaranteed that t1 will either join the gta and execute the task if there is an available slot or send a task and it would be executed by another worker in gta. In each case, the thread executing the lambda and calling try_put is guaranteed to be in gta and hence has its reference node in the TLS (the code will choose the true branch).
If t1 would submit work in the graph without explicitly calling task_arena::execute, the false branch would be chosen and the task reference counting would work as before this PR by adding a reference to graph wait_context.

[vossmjp]

Looks good to me.