17. Flow Graphs: Beyond the Basics

In Chapter 16, we discussed Ranges and Partitioners and how these can be used to ensure that the tasks created by the TBB generic algorithms are large enough to amortize scheduling overheads while still being small enough to provide enough independent work items for scalability. The TBB flow graph does not have support for Ranges and Partitioners, but we still need to be concerned about task granularity.

To see if our rule of thumb for 1 microsecond tasks that we introduced in Chapter 16 applies as well to flow graph nodes as it does to parallel algorithm bodies, we will explore a few simple microbenchmarks that capture the extremes that can exist in flow graphs. We will compare the execution times of four functions and use different amounts of work per node execution. We will refer to these functions as Serial, FG loop, Master loop, and FG loop per worker.

It is our belief that studying these examples (Figures 17-1 to 17-4) is critical to having an intuitive grasp of some key issues that differentiate highly scalable flow graph usage and disappointing uses of flow graph. The APIs themselves, fully documented in Appendix B, do not provide this education – we hope you will study these examples enough to grasp the concepts as we believe this will make you much better at getting the most out of using TBB flow graphs (peek at Figure 17-5 to see a quantification of the benefits on performance of understanding these!).

The Serial loop is our baseline and contains a for-loop that calls an active spin-wait function N times, as shown in Figure 17-1.

The FG loop function is shown in Figure 17-2. This function builds a flow graph that has a single multifunction_node with an edge from its output to its input. A single message starts the cycle and the node then spin-waits and sends a message back to its input. The cycle repeats N-1 times. Because the node spins before sending the message back to its input, this graph is still a mostly serial loop – the bulk of the work in the body tasks will not overlap. However, because the message is sent before the body returns, there is still a small-time gap during which another thread can steal the task that the try_put generates. We can use this graph to see the basic overhead of the flow graph infrastructure.

Our next microbenchmarking function, Master loop shown in Figure 17-3, does not create a cycle. It instead sends all N messages to the multifunction_node directly from the master thread in a serial loop. Since the multifunction_node has unlimited parallelism and the serial for-loop will send messages very quickly, there are a lot of parallel tasks created. However, because the master thread is the only thread that calls the try_put method on node n, all body tasks are spawned into the master thread’s local deque. Worker threads that participate in executing this graph will be forced to steal each task they execute – and only after they have randomly selected the master as their victim. We can use this graph to see the behavior of a flow graph with sufficient parallelism but that requires an extreme amount of work-stealing.

Finally, Figure 17-4 shows the FG loop per worker function. This function spreads the tasks across the master and worker threads’ local deques, since once a thread has stolen its initial task, it will then spawn tasks into its own local deque. We can use this graph to see the behavior of a flow graph with a very small amount of stealing.

We ran these microbenchmarks using N=65,536 and spin-wait times of 100 ns, 1 us, 10 us, and 100 us. We collected their average execution times over 10 trials and present the results in Figure 17-5. From these results, we can see that when the task sizes are very small, 100 nanoseconds for example, the overhead of the flow graph infrastructure leads to degraded performance in all cases. With task sizes of at least 1 microsecond, we begin to profit from parallel execution. And by the time we reach a task size of 100 microseconds, we are able to reach close to perfect linear speedups.

We can further understand the performance of our microbenchmarks by collecting a trace and viewing the results in Flow Graph Analyzer (FGA) – FGA is described in more detail at the end of this chapter. Figure 17-6 shows per-thread timelines for the different functions when using a spin-wait time of 1 microsecond. These timelines, which are all of the same length, show the work done by each thread over time. The gaps (in gray) in the timelines indicate when a thread is not actively executing a node’s body. In Figure 17-6(a), we see the behavior of FG loop, which acts like a serial loop. But we can see that the small gap between the try_put in the body and the exit from the task allows the tasks to ping-pong between the threads since they are able to steal each task as it is spawned. This partially explains the fairly large overheads for this microbenchmark shown in Figure 17-5. As we explain later in this chapter, most functional nodes use scheduler bypass to follow their data to the next node when possible (see the discussion on Pipelines and data locality and thread affinity in Chapter 16 for a more detailed discussion of why scheduler bypass improves cache performance). Since a multifunction_node puts output messages to its output ports directly inside of the body implementation, it cannot immediately follow the data to the next node using scheduler bypass – it has to finish its own body first! A multifunction_node therefore does not use scheduler bypass to optimize for locality. In any case, this makes the performance in Figure 17-6(a) a worst-case overhead, since scheduler bypass is not used.

In Figure 17-6(b), we see the case where the master thread is generating all of the tasks and the workers must steal each task, but tasks can be executed in parallel once they are stolen. Because the worker threads must steal each task, they are much slower at finding tasks than the master thread. The master thread is continually busy in Figure 17-6(b) – it can quickly pop a next task from its local deque – while the worker threads’ timelines show gaps during which they are fighting with each other to steal their next task from the master’s local deque.

Figure 17-6(c) shows the good behavior of FG loop per worker, where each thread is able to quickly pop its next task from its local deque. Now we see very few gaps in the timelines.

Looking at these extremes of behavior and noting the performance in Figure 17-5, we feel comfortable recommending a similar rule of thumb for flow graph nodes. While a pathological case, like Master loop, shows a limited speedup of 2.8 with a 1 microsecond body, it still shows a speedup. If the work is more balanced, such as with FG loop per worker, a 1 microsecond body provides a good speedup. With these caveats in mind, we again recommend a 1 microsecond execution time as a crude guideline:

Both implicit and explicit task arenas impact the behavior of TBB tasks and the TBB generic parallel algorithms. The arena in which tasks are spawned controls which threads can participate in executing the tasks. In Chapter 11, we saw how we can use implicit and explicit arenas to control the number of threads that participate in executing parallel work. In Chapters 12–14, we saw that explicit task arenas can be used with task_sheduler_observer objects to set the properties of threads as they join arenas. Because of the impact of task arenas on available parallelism and data locality, in this section, we take a closer look at how task arenas mix with flow graphs.

                      Without reset:
                      default : 8
                      a2 : 8
                      a4 : 8
                    

                      With reset:
                      default : 8
                      a2 : 2
                      a4 : 4
                    

Just like with a pipeline, a flow graph can have great scalability if it uses parallel (flow::unlimited) nodes but can have limited scalability if it has serial nodes. One way to increase scaling is to use nested parallel algorithms inside of TBB flow graph nodes. TBB is all about composability, so we should use nested parallelism when possible.

As we have seen throughout this book, the TBB parallel algorithms such as parallel_for and parallel_reduce are highly optimized and include features like Ranges and Partitioners that let us optimize performance even more. We have also seen that the flow graph interface is very expressive – we can express graphs that include loops and use nodes like multifunction_node to output many messages from each invocation. We should therefore be on the lookout for cases where we create patterns in our graphs that are better expressed using nested parallelism. One simple example is shown in Figure 17-20.

In Figure 17-20, for each message that the multifunction_node receives, it generates many output messages that flow into a function_node with unlimited concurrency. This graph will act a lot like a parallel loop, with the multifunction_node acting as the control loop and the function_node as the body. But it will require a lot of stealing to distribute the work like the Master loop from Figures 17-3 and 17–5. While there may be valid uses of this pattern, it is likely more efficient to use a highly optimized parallel loop algorithm instead. This entire graph might be collapsed into a single node that contains a nested parallel_for, for example. Of course, whether or not this replacement is possible or desirable depends on the application.

Because a flow graph is less structured than a simple pipeline, we may sometimes need to establish an ordering of messages at points in the graph. There are three common approaches for establishing order in a data flow graph: use a key-matching join_node, use a sequencer_node, or use a multifunction_node.

For example, in Chapter 3, the parallelism in our stereoscopic 3D flow graph allowed the left and right images to arrive out of order at the mergeImageBuffersNode. In that example, we ensured that the correct two images were paired together as inputs to the mergeImageBuffersNode by using a tag-matching join_node. A tag-matching join_node is a type of key-matching join_node. By using this join_node type, inputs can arrive in different orders at the two input ports but will still be properly matched based on their tag or key. You can find more information on the different join policies in Appendix B.

Another way to establish order is to use a sequencer_node. A sequencer_node is a buffer that outputs messages in sequence order, using a user-provided body object to obtain the sequence number from the incoming message.

In Figure 17-21, we can see a three-node graph, with nodes first_node, sequencer, and last_node. We use a sequencer_node to reestablish the input order of the messages before the final serial output node last_node. Because function_node first_node is unlimited, its tasks can finish out of order and send their output as they complete. The sequencer_node reestablishes the input order by using the sequence number assigned when each message was originally constructed.

If we execute a similar example without a sequencer node and N=10, the output is scrambled as the messages pass each other on their way to last_node:

                    9 no sequencer
                    8 no sequencer
                    7 no sequencer
                    0 no sequencer
                    1 no sequencer
                    2 no sequencer
                    6 no sequencer
                    5 no sequencer
                    4 no sequencer
                    3 no sequencer
                  

When we execute the code in Figure 17-21, we see the output:

                    0 with sequencer
                    1 with sequencer
                    2 with sequencer
                    3 with sequencer
                    4 with sequencer
                    5 with sequencer
                    6 with sequencer
                    7 with sequencer
                    8 with sequencer
                    9 with sequencer
                  

As we can see, a sequencer_node can reestablish the order of the messages, but it does require us to assign the sequence number and also to provide a body to the sequencer_node that can obtain that number from an incoming message.

A final approach to establishing order is to use a serial multifunction_node. A multifunction_node can output zero or more messages on any of its output ports for a given input message. Since it is not forced to output a message for each incoming message, it can buffer incoming messages and hold them until some user-defined ordering constraint is met.

For example, Figure 17-22 shows how we can implement a sequencer_node using a multifunction_node by buffering incoming messages until the next message in sequencer order has arrived. This example assumes that at most N messages are sent to a node sequencer and that the sequence numbers start at 0 and are contiguous up to N-1. Vector v is created with N elements initialized as empty shared_ptr objects. When a message arrives at sequencer, it is assigned to the corresponding element of v. Then starting at the last sent sequence number, each element of v that has a valid message is sent and the sequence number is incremented. For some incoming messages, no output message will be sent; for others, one or more messages may be sent.

While Figure 17-22 shows how a multifunction_node can be used to reorder messages by sequence order, in general, any user-defined ordering or bundling of messages can be used.

In Chapter 12, we talked about how we may sometimes need to create isolation for performance or correctness reasons when using TBB algorithms. The same is true for flow graphs, and as with the generic algorithms, this can be especially true with nested parallelism. The implementation of the graph in Figure 17-23 shows a simple graph with nodes source and unlimited_node, and nested parallelism inside node unlimited_node. A thread may moonlight (see Chapter 12) while waiting for the nested parallel_for loop in node unlimited_node to complete, and pick up another instance of node unlimited_node. The node unlimited_node prints “X started by Y”, where X is the node instance number and Y is the thread id.

On our test system with eight logical cores, one output showed that our thread 0 was so bored it pick up not just one, but three different instances of unlimited_node, while waiting for its first parallel_for algorithm to finish as shown in Figure 17-24.

As we discussed in Chapter 12, moonlighting is typically benign, which is the case here since we’re not computing anything real. But as we highlighted in our previous discussions about isolation, this behavior is not always benign and can lead to correctness issues, or decreased performance, in some cases.

We can address moonlighting in a flow graph just as we did with general tasks in Chapter 12, with the this_task_arena::isolate function or with explicit task arenas. For example, instead of calling the parallel_for directly in the node body, we can invoke it inside of an isolate call:

                    tbb::this_task_arena::isolate([P,spin_time]() {
                      tbb::parallel_for(0, P-1, [spin_time](int i) {
                        spinWaitForAtLeast((i+1)∗spin_time);
                      });
                    });
                  

After changing our code to use this function, we see that the threads no longer moonlight and each thread stays focused on a single node until that node is complete as shown in Figure 17-25.

In Chapter 15, we discussed task cancellation and exception handling when using TBB tasks in general. Since we are already familiar with this topic, we will only highlight the flow graph related aspects in this section.

                      tbb::task_group_context tgc;
                      tbb::flow::graph g{tgc};
                    

                      tgc.cancel_group_excution();
                    

                      if (g.is_cancelled()) {
                        std::cout << "My graph was cancelled!" << std::endl;
                      }
                    

                      tbb::task::self().cancel_group_execution();Data flow graphsDoscancelling
                    

                      terminate called after throwing an instance of 'int'
                    

                      Caught 2
                      Caught 1
                    

                      Caught 2
                    

We can set priorities for all of the tasks spawned by a graph by using the graph’s task_group_context, for example:

                    if (auto t = g.root_task()) {
                      t->group()->set_priority(tbb::priority_high);
                    }
                  

Shortly before the publication of this book, support for relative priorities for functional nodes was added to TBB as a preview feature. Using this feature, we can pass a parameter to a node’s constructor to give it a priority relative to other functional nodes. This interface was first provided in TBB 2019 Update 3. Interested readers can learn more details about this new functionality in the online TBB release notes and documentation.

All graph nodes require a reference to a graph object as one of the arguments to their constructor. In general, it is only safe to construct edges between nodes that are part of the same graph. Connecting two nodes in different graphs can make it difficult to reason about graph behaviors, such as what task arenas will be used, if our calls to wait_for_all will properly detect graph termination, and so on. To optimize performance, the TBB library takes advantage of its knowledge about edges. If we connect two graphs by an edge, the TBB library will freely reach across this edge for optimization purposes. We may believe that we have created two distinct graphs, but if there are shared edges, TBB can start mixing their executions together in unexpected ways.

To demonstrate how we can get unexpected behavior, we implemented the class WhereAmIRunningBody shown in Figure 17-29. It prints max_concurrency and priority settings, which we will use to infer what task arena and task_group_context this body’s task is using when it executes.

Figure 17-30 provides an example that uses the WhereAmIRunningBody to demonstrate an unexpected behavior. In this example, we create two nodes: g2_node and g4_node. The node g2_node is constructed with a reference to g2. The graph g2 is passed a reference to a task_group_context that has priority_normal and g2 is reset() in a task_arena with a concurrency of 2. We should therefore expect g2_node to execute with normal priority in an arena with 2 threads, right? The node g4_node is constructed such that we should expect it to execute with high priority in an arena with four threads.

The first group of calls that include g2_node.try_put(0) and g4_node.try_put(1) match these expectations:

But, when we make an edge from g2_node to g4_node, we make a connection between nodes that exist in two different graphs. Our second set of calls that include g2_node.try_put(2) again cause the body of g2_node to execute with normal priority in arena a2. But TBB, trying to reduce scheduling overheads, uses scheduler bypass (see Scheduler Bypass in Chapter 10) when it invokes g4_node due to the edge from g2_node to g4_node. The result is that g4_node executes in the same thread as g2_node, but this thread belongs to arena a2 not a4. It still uses the correct task_group_context when the task is constructed, but it winds up being scheduled in an unexpected arena.

                    2:g2_node executing in arena 2 with priority normal
                    2:g4_node executing in arena 2 with priority high
                  

From this simple example, we can see that this edge breaks the separation between the graphs. If we were using arenas a2 and a4 to control the number of threads, for work isolation or for thread affinity purposes, this edge will undo our efforts. We should not make edges between graphs.

In the previous “Don’t,” we decided that we should not make edges between graphs. But what if we really need to communicate across graphs? The least dangerous option is to explicitly call try_put to send a message from a node in one graph to a node in another graph. We don’t introduce an edge, so the TBB library won’t do anything sneaky to optimize the communication between the two nodes. Even in this case though, we still need to be careful as our example in Figure 17-31 demonstrates.

Here, we create a graph g2 that sends a message to graph g1 and then waits for both graph g1 and g2. But, the waiting is done in the wrong order!

Since node g2_node2 sends a message to g1_node1, the call to g1.wait_for_all() will likely return immediately since nothing is going on in g1 at the time of the call. We then call g2.wait_for_all(), which returns after g2_node2 is done. After this call returns, g2 is finished but g1 has just received a message from g2_node2 and its node g1_node1 has just started to execute!

Luckily, if we call the waits in the reverse order, things will work as expected:

                    g2.wait_for_all();
                    g1.wait_for_all();
                  

In the previous two sections, we warned that communication between graphs can lead to errors. Often developers use more than one graph because they want to logically separate some nodes from others. Encapsulating a group of nodes is convenient if there is a common pattern that needs to be created many times or if there is too much detail in one large flat graph.

In both of these cases, we can use a tbb::flow::composite_node. A composite_node is used to encapsulate a collection of other nodes so they can be used like a first-class graph node. Its interface follows:

Unlike the other node types that we have discussed in this chapter and in Chapter 3, we need to create a new class that inherits from tbb::flow::composite_node to make use of its functionality. For example, let’s consider the flow graph in Figure 17-32(a). This graph combines two inputs from source1 and source2, and uses a token passing scheme to limit memory consumption.

If this token passing pattern is commonly used in our application, or by members of our development team, it might make sense to encapsulate it into its own node type, as shown in Figure 17-32(b). It also cleans up the high-level view of our application by hiding the details.

Figure 17-33 shows what a flow graph implementation looks like if we have a node that implements the dotted parts of Figure 17-32(a), replacing it with a single merge node. In Figure 17-33, we use the merge node object like any other flow graph node, making edges to its input and output ports. Figure 17-34 shows how we use tbb::flow::composite_node to implement our MergeNode class.

In Figure 17-34, MergeNode inherits from CompositeType, which is an alias for

The two template arguments indicate that a MergeNode will have two input ports, both that receive BigObjectPtr messages, and a single output port that sends BigObjectPtr messages. The class MergeNode has a member variable for each node it encapsulates: a tokenBuffer, a join, and a combine node. And these member variables are initialized in the member initializer list of the MergeNode constructor. In the constructor body, calls to tbb::flow::make_edge set up all of the internal edges. A call to set_external_ports is used to assign the ports from the member nodes to the external ports of the MergeNode. In this case, the first two input ports of join become the inputs of the MergeNode and the output of combine becomes the output the MergeNode. Finally, because the node is implementing a token passing scheme, the tokenBuffer is filled with tokens.

While creating a new type that inherits from tbb::flow::composite_node may appear daunting at first, using this interface can lead to more readable and reusable code, especially as your flow graphs become larger and more complicated.

The design workflow in FGA lets us graphically design TBB flow graphs, validate that they are correct, estimate their scalability, and, after we are satisfied with our design, generate a C++ implementation that uses the TBB flow graph classes and functions. FGA is not a full Integrated Development Environment (IDE) like Microsoft Visual Studio, Eclipse or Xcode. Instead, it gets us started with our flow graph design, but then we need to step outside of the tool to complete the development. However, if we use the design workflow in a constrained way, as we will describe later, iterative development in the designer is possible.

Figure 17-35 shows the FGA GUI used during the design workflow. We will only briefly describe the components of the tool here as we describe the typical workflow; the Flow Graph Analyzer documentation provides a more complete description.

The typical design workflow starts with a blank canvas and project. As highlighted by the black circle numbered 1 in Figure 17-35, we select nodes in the node palette and place them on the canvas, connecting them together by drawing edges between their ports. The node palette contains all of the node types available in the TBB flow graph interface and provides tooltips that remind us about the functionality of each type. For each node on the canvas, we can modify its type-specific properties; for a function_node for example, we can provide the C++ code for the body, set a concurrency limit, and so on. We can also provide an estimated “weight” that represents the computational complexity of the node so that later we can run a Scalability Analysis to see if our graph will perform well.

Once we have drawn our graph on the canvas, we run a Rule Check that analyzes the graph looking for common mistakes and anti-patterns. The Rule Check results, highlighted by the black circle numbered 2 in Figure 17-35, show issues such as unnecessary buffering, type mismatches, suspicious cycles in the graph, and so on. In Figure 17-35, the Rule Check has discovered that there is a type mismatch between the input of our limiter_node and the output of our multifunction_node. In response, we can then, for example, modify the port output type of our multifunction_node to fix this issue.

When we have fixed all correctness issues uncovered by the Rule Check, we can then run a Scalability Analysis. The Scalability Analysis constructs a TBB flow graph in memory, replacing the computational node bodies with dummy bodies that actively spin for a time proportional to their “weight” property. FGA runs this model of our graph on various numbers of threads and provides a table of the speedups, for example:

Using these features, we can iteratively refine our graph design. Along the way, we can save our graph design in GraphML format (a common standard for representing graphs). When we are satisfied with our design we can generate C++ code that uses the TBB flow graph interface to express our design. This code generator is more accurately viewed as a code wizard than an IDE since it does not directly support an iterative code development model. If we change the generated code, there is no way to reimport our changes into the tool.

The analysis workflow in FGA is independent of the design workflow. While we can surely analyze a flow graph that was designed in FGA, we can just as easily analyze a TBB flow graph that is designed and implemented outside of the tool. This is possible because the TBB library is instrumented to provide runtime events to the FGA trace collector. A trace collected from a TBB application lets FGA reconstruct the graph structure and the timeline of the node body executions – it does not depend on the GraphML files developed during the design workflow.

If we want to use FGA to analyze a TBB application that uses a flow graph, the first step is to collect an FGA trace. By default, TBB does not generate traces, so we need to activate trace collection. The FGA instrumentation in TBB was a preview feature prior to TBB 2019. We need to take extra steps if we are using an older version of TBB. We refer readers to the FGA documentation for instructions on how to collect traces for the version of TBB and FGA that they are using.

Once we have a trace of our application, the analysis workflow in FGA uses the activities highlighted by the numbered circles in Figure 17-36: (1) inspect the tree-map view for an overview of the graph performance and use this as an index into the graph topology display, (2) run the critical path algorithm to determine the critical paths through the computation, and (3) examine the timeline and concurrency data for insight into performance over time. Analysis is most commonly an interactive process that moves between these different activities as the performance of the application is explored.

The tree-map view labeled as (1) in Figure 17-36 provides an overview of the overall health of a graph. In the tree map, the area of each rectangle represents the total aggregated CPU time of the node and the color of each square indicates the concurrency observed during the execution of the node. The concurrency information is categorized as poor (red), ok (orange), good (green), and oversubscribed (blue).

Nodes with a large area that are marked as “poor” are hotspots and have an average concurrency between 0% and 25% of the hardware concurrency. These are therefore good candidates for optimization. The tree-map view also serves as an index into a large graph; clicking on a square will highlight the node in the graph and selecting this highlighted node will in turn mark tasks from all instances of this node in the timeline trace view.

The graph topology canvas is synchronized with other views in the tool. Selecting a node in the tree-map view, the timeline, or in a data analytics report will highlight the node in the canvas. This lets users quickly relate performance data to the graph structure.

One of the most important analytic reports provided by FGA is the list of critical paths in a graph. This feature is particularly useful when one has to analyze a large and complex graph. Computing the critical paths results in a list of nodes that form the critical paths as shown in the region labeled (2) in Figure 17-36. As we discussed in Chapter 3, an upper bound on speedup of dependency graphs can be quickly computed by dividing the aggregate total time spent by all nodes in a graph by the time spent on the longest critical path, T₁/T_∞. This upper bound can be used to set expectations on the potential speedup for an application expressed as a graph.

The timeline and concurrency view labeled as (3) in Figure 17-36 displays the raw traces in swim lanes mapped to software threads. Using this trace information, FGA computes additional derived data such as the average concurrency of each node and the concurrency histogram over time for the graph execution. Above the per-thread swim lanes, a histogram shows how many nodes are active at that point in time. This view lets users quickly identify time regions with low concurrency. Clicking on nodes in the timelines during these regions of low concurrency lets developers find the structures in their graph that lead to these bottlenecks.

In this chapter, we discussed a number of potential performance issues that can arise when using a flow graph. In this section, we briefly discuss how FGA can be used to explore these issues in a TBB-based application.