2006.07478v1

## Streaming Computations with Region-Based State on SIMD Architectures Stephen Timcheck and Jeremy Buhler Washington University in St. Louis One Brookings Dr., St. Louis, MO 63130, USA { stimcheck,jbuhler } @wustl.edu Abstract. Streaming computations on massive data sets are an attractive candidate for parallelization, particularly when they exhibit independence (and hence data parallelism) between items in the stream. However, some streaming computations are stateful, which disrupts independence and can limit parallelism. In this work, we consider how to extract data parallelism from streaming computations with a common, limited form of statefulness. The stream is assumed to be divided into variably-sized regions , and items in the same region are processed in a common context of state. In general, the computation to be performed on a stream is also irregular , with each item potentially undergoing different, data-dependent processing. This work describes mechanisms to implement such computations efficiently on a SIMD-parallel architecture such as a GPU. We first develop a low-level protocol by which a data stream can be augmented with control signals that are delivered to each stage of a computation at precise points in the stream. We then describe an abstraction, enumeration and aggregation , by which an application developer can specify the behavior of a streaming application with region-based state. Finally, we study an implementation of our ideas as part of the MERCATOR system [5] for irregular streaming computations on GPUs, investigating how the frequency of region boundaries in a stream impacts SIMD occupancy and hence application performance. 1 Keywords: signal · control message · SIMD · irregular · streaming ## 1 Introduction Streaming computations are integral to high-impact applications such as biological sequence analysis [1], astrophysics [13], and decision cascades in machine learning [14]. These computations operate on a long stream of data items, each of which must be processed through a pipeline of computational stages. When items can be processed independently and identically, extracting data parallelism is straightforward, particularly on SIMD-parallel architectures such as GPUs. 1 Presented at 13th Int'l Wkshp. on Programmability and Architectures for Heterogeneous Multicores, Bologna, Italy, Jan 2020 . Copyright c © 2020 by Stephen Timcheck and Jeremy Buhler. All rights reserved. A greater challenge lies in supporting streaming computations whose behavior deviates from the above ideal. Deviations can occur in two ways. First, items may not be processed identically; in particular, some stages of the pipeline may produce a variable, data-dependent amount of output for each input item. Such computations - which include all the examples cited above as well as applications in, e.g., particle simulation [2] and network packet processing [10] are said to exhibit irregular data flow, which complicates the lock-step execution model of a SIMD-parallel processor. Our prior work on the MERCATOR system [5] describes a way to support such irregular computations on GPUs. A second deviation, which is the subject of the present work, arises when items in a stream cannot be processed independently; that is, the computation is stateful. To avoid the need for full serialization, we focus on a common scenario in which the input stream is divided into variably-sized regions . Items in one region are processed independently of each other but in a common context. For example, a stream of characters may be grouped into lines or network packets; a stream of edges in a graph may be grouped by their source vertex; or a stream of measurements may be grouped by a common time window or event trigger. Region boundaries are state-change events for the stream - items after a boundary must be processed differently than items before it. In this work, we investigate mechanisms to support streaming computations with region-based contextual state on SIMD-parallel platforms. Our contributions are threefold. First, we describe a low-level mechanism for precise delivery of control signals between pipeline stages of a streaming application. This mechanism, unlike those described in some prior work (e.g. [12]), supports irregular dataflow. Second, we use this mechanism to construct an abstraction, enumeration and aggregation , that lets application developers express region-based contextual state as part of a streaming application. Finally, we implement our designs in MERCATOR to investigate the performance implications of regional context, exposing a SIMD-specific performance tradeoff between alternate ways of implementing this behavior in applications. The two strategies we examine trade off between SIMD occupancy and representation overhead. The remainder of this paper is organized as follows. Section 2 describes the application and architectural models in which we formulate our work and considers related work in other streaming models. Section 3 describes our protocol for synchronizing control signals with a data stream. Section 4 describes the developer-facing abstraction of enumeration and aggregation, which we implement in terms of signals. Section 5 investigates the performance of our designs, while Section 6 concludes and considers future work. ## 2 Background and Related Work ## 2.1 Application Model A streaming application consists of a pipeline of compute nodes connected by fixed-sized data queues . A node consumes a stream of data items from its input queue and produces a stream of data items (perhaps of a different type) at its output that are queued for processing by the next node downstream in the pipeline. When a node is executed to consume one or more inputs, we say that the node fires . Each input data item consumed by a node causes it to produce zero or more outputs. The number of outputs may vary for each input consumed, up to some node-dependent maximum, and is not known prior to execution. Figure 1a shows a simple application pipeline with three nodes. While we mainly discuss linear pipelines in this work, our contributions also apply to tree-structured topologies like those in Figure 1b. Fig. 1. (a) Streaming computation pipeline with three compute nodes; (b) Pipeline with a tree topology. <details> <summary>Image 1 Details</summary>  ### Visual Description ## Diagram: Node-Based Structures A and B ### Overview The image presents two distinct node-based diagrams labeled **A** and **B**. Diagram A depicts a linear sequence of three nodes, while Diagram B illustrates a branched structure with shared and divergent connections. No numerical data, legends, or color-coded elements are present. ### Components/Axes - **Diagram A**: - Nodes: `n₁` → `n₂` → `n₃` (linear chain). - Arrows indicate unidirectional flow or dependency. - **Diagram B**: - Nodes: A central node (unlabeled) connects to three branches: - One branch leads to an unlabeled node. - Another branch leads to `n₃`. - A third branch leads to `n₄`. - Arrows suggest directional relationships, with `n₃` and `n₄` sharing a common predecessor. ### Detailed Analysis - **Diagram A**: - Sequential flow: `n₁` → `n₂` → `n₃`. - No feedback loops or alternative pathways. - **Diagram B**: - Branching at the central node: - One path terminates at an unlabeled node. - Two paths converge at `n₃` and `n₄`, respectively. - No explicit labels for the central node or the unlabeled terminal node. ### Key Observations 1. **Structural Contrast**: - Diagram A represents a strictly linear progression. - Diagram B introduces divergence, with `n₃` and `n₄` sharing a common origin but no direct connection between them. 2. **Ambiguity in Labels**: - The central node in Diagram B and the unlabeled terminal node lack identifiers, limiting interpretability. 3. **Arrow Consistency**: - All arrows in both diagrams point forward, suggesting unidirectional relationships. ### Interpretation - **Functional Implications**: - Diagram A could model a linear process (e.g., workflow, data pipeline). - Diagram B might represent a decision tree or network with parallel pathways, where `n₃` and `n₄` are outcomes of a shared decision node. - **Missing Context**: - The absence of labels for critical nodes (e.g., the central node in B) reduces clarity. - No indication of node weights, probabilities, or feedback mechanisms. - **Potential Use Cases**: - Diagram A: Suitable for systems requiring strict sequential execution. - Diagram B: Useful for modeling scenarios with branching logic (e.g., user journeys, risk assessment). ## Notes - No numerical data, legends, or color-coding present. - All textual elements are node labels (`n₁`, `n₂`, `n₃`, `n₄`) and diagram identifiers (A, B). - Arrows denote directionality but lack explicit semantics (e.g., causality, dependency). </details> We do not consider DAG-structured topologies because the semantics associated with convergent edges are complex under irregular dataflow, even in the absence of stateful execution [9]. Topologies with cycles have clearer streaming semantics, but in the presence of irregularity, items in the stream can be reordered if they take different numbers of trips around a cycle. For such topologies, Maintaining precisely ordered control boundaries in the stream requires aggressive reordering that is beyond the scope of this work. An application is provided with an initial stream of inputs to its source node . When the application executes, a global scheduler repeatedly chooses a node with one or more pending inputs to fire. Because of our architectural mapping below, we assume that only one node fires at a time and that a node cannot be preempted while firing. The scheduler continues to select and fire nodes until no node has any inputs remaining. A signal is a control message generated by a node for consumption by its downstream neighbor. When a node receives a signal it can change its state and may also generate additional signals to the next node downstream. Signals must be delivered precisely with respect to the stream of data items. Formally, suppose we have two successive nodes n 1 and n 2 in a pipeline. If n 1 emits a data item d , followed by a signal s , followed by another data item d ′ , then n 2 must receive s after processing d but before processing d ′ . ## 2.2 Target Architecture and Mapping Our work targets wide-SIMD multiprocessors. While many general-purpose processors support SIMD instructions, we realize our designs on NVIDIA GPUs due to their robust CUDA tool chain and their popularity as accelerators. A GPU may be viewed as a collection of SIMD processors sharing a common memory. Each processor has a fixed SIMD width w , which is the number of concurrent SIMD lanes that it can execute 2 . SIMD lanes execute computations 2 We treat the CUDA block size as the effective SIMD width, ignoring CUDA's virtualization of an underlying, smaller width (the warp size). in lock-step; hence, divergent behavior such as nonuniform branches, or lack of inputs to some SIMD lanes, causes some lanes to sit idle while others execute. Mapping a streaming computational pipeline onto a GPU entails moving the input stream from the host system to the GPU's memory, executing a GPU kernel to process the data through the pipeline, and finally transferring the output stream back to the host. Our work focuses on efficient processing on the GPU, rather than the orthogonal task of host/GPU data transfer. GPU runtimes offer little support for interprocessor synchronization other than via control transfer back to the host. To avoid excessive host-device control overhead, we therefore instantiate the pipeline separately on each processor of the GPU, then let all processors' pipelines compete to consume data from a common input stream. This approach requires atomic operations but no locking. The GPU kernel does not return control to the host until the entire stream is consumed. Each processor on the GPU independently implements the sequential, non-preemptively scheduled computation described above. However, a node now consumes not a single input per firing but rather a variably-sized ensemble of inputs, up to the processor's SIMD width, which are processed in parallel. Our performance goal is to maximize throughput, or equivalently to minimize time for the GPU to process an entire input stream. A secondary goal that promotes high throughput is to maximize SIMD occupancy , the fraction of SIMD lanes doing useful work at any step of the computation. In particular, we want the sizes of ensembles presented to each node to achieve the full SIMD width. However, we will show that region-based state can be in tension with the goal of maximizing SIMD occupancy, creating a challenge for performance optimization. ## 2.3 Related work Several systems and languages have been developed to express streaming dataflow computations. One of the most influential such systems is StreamIt [11], which implements the synchronous data flow (SDF) abstraction [8]. StreamIt assumes a fixed number of outputs per input to a compute node. This assumption allows StreamIt to offer a powerful abstraction, teleport messaging [12], in which signals can flow both forward and backward in a pipeline. StreamIt can also schedule a signal to be delivered to an arbitrary destination node at some precise future time, rather than forcing the signal to flow through the pipeline. Many capabilities of teleport messaging rely on the underlying SDF model. In contrast, our model does not assume a fixed number of outputs per input and so requires a different design to ensure precise signaling. Our work therefore extends precise signaling capabilities from regular to irregular streaming applications. Like StreamIt, we choose to send signals 'out of band,' in our case via parallel control edges, rather than attempt to enqueue them together with data items. Other streaming systems, such as Ptolemy [6] and Auto-Pipe [4], support a variety of dataflow semantics, including multiple, differently-typed dataflow edges between a pair of nodes. This support is in principle sufficient to implement a control channel for signals. However, except in restricted cases like SDF, the systems do not specify how multiple channels between the same two nodes are synchronized and so do not by themselves support precise signaling. Our work is influenced by a control messaging protocol developed by Li et al. [9]. That work, however, incurs additional complexity to support asynchronous streaming dataflow and to impose well-defined semantics on convergent dataflow edges in an DAG-structured irregular application. We preserve their idea of a credit protocol for synchronizing data and control streams but realize this idea in a way that is efficient for our target model and architecture. The idea of processing part of a stream of items in a common context is similar to facilities present in Apache Spark [16]. Spark streams consist of a sequence of RDDs [15], which are discrete data sets, processed as a unit, that may contain multiple elements. Our abstraction supports some operations semantically similar to Spark's but realizes them in the context of a single wide-SIMD processor rather than the multicore and distributed systems that Spark targets. Other frameworks supporting streaming-like behavior, such as CnC-CUDA [7], utilize an 'in-band' approach with control collections that mix control and data into a single stream. Control collections are analogous to our regions of items with a common context. Their implementation requires a tag for each item to track the region associated with it. In contrast, our implementation keeps region boundaries synchronized with the data stream without the need for tagging. We compare these two approaches in Section 5. ## 3 A Mechanism for Precise Signaling In this section, we describe how to synchronize data and control signals between two successive nodes in a streaming pipeline. The mechanism bears some similarities to control flow in networking. We state the correctness properties of our design but relegate their proofs to the appendix. ## 3.1 Credit Protocol for Synchronizing Signals Let n 1 and n 2 be successive nodes in a pipeline, connected by data queue Q . We add a separate, finite-sized signal queue S between the nodes, as in Figure 2a. Data items are moved from n 1 to n 2 on Q , while signals are moved on S . We must ensure that, although data and signals move on separate queues, their movement is synchronized so as to ensure precise signal delivery. For this purpose, we introduce a credit protocol between n 1 and n 2 . Each signal created by n 1 is assigned an non-negative integer amount of credit , which is transmitted along with the signal on S . Credit records a number of data items that n 2 must process before it can receive the signal. When n 1 emits a signal s , it uses two rules to set the credit associated with s . (1) If no signal is currently queued on S , then s gets an amount of credit equal to the number of data items queued on Q . (2) If one or more signals are queued on S , let s ′ be the signal at the tail of S . Then s gets an amount of credit equal to the number of data items emitted by n 1 since s ′ was enqueued. Node n 1 maintains a counter of emitted data items, which is reset each time it emits a signal, that is used to implement the second rule. The downstream node n 2 maintains a current credit counter , initially set to 0, that tracks the number of items that can safely be consumed before processing the next signal. Node n 2 uses the following two rules to determine whether to process data or a signal when it fires. (1) If no signal is queued on S , n 2 may freely consume any available data items on Q without regard to the counter. Otherwise (i.e., a signal is queued on S ), (2a) if the current credit counter is non-zero, n 2 may consume only a number of data items less than or equal to the value of this counter, which is decremented once for each data item consumed. (2b) If instead the current credit counter is 0, let s be the signal at the head of queue S . If s carries more than 0 credit, that credit is removed from s and added to the current credit counter. Otherwise, n 2 consumes s . Figure 2b illustrates how the credit carried in the signals and in the receiving node's credit counter synchronizes the two queues. Lemma 1. A signal s emitted by n 1 is received by n 2 precisely when n 2 has consumed all data items emitted by n 1 prior to s . ## 3.2 Scheduling Applications with Signals A firing of a node n proceeds in two phases: a data phase and a signal phase . In the data phase, n consumes as many queued data items as it can. The number of items consumed is limited to the minimum of three values: the number of queued items, the amount of space in n 's downstream queue, and (if a signal is pending for n ) the value of n 's current credit counter. Once n can consume no more data, if its current credit counter is 0, it enters the signal phase, in which it consumes as many queued signals as it can. Signal processing ends when no queued signals remain, or when n 's current credit counter becomes > 0 (and hence data must be consumed prior to the next queued signal). A node is fireable if it has either data or a signal pending, and if there is sufficient space in its output queue to hold any outputs from the firing. The maximum number of output data items per input item is known a priori for each node, so the scheduler can determine whether at least one data item can <details> <summary>Image 2 Details</summary>  ### Visual Description ## Diagram: System Interaction Model ### Overview The image contains two side-by-side diagrams (A and B) depicting a system with two nodes (`n₁` and `n₂`) connected by a bidirectional arrow labeled `Q`. Diagram B includes additional annotations (dots, squares, and labels) not present in Diagram A. ### Components/Axes - **Nodes**: - `n₁` (left circle) and `n₂` (right circle) in both diagrams. - **Arrows**: - Bidirectional arrow labeled `Q` connecting `n₁` and `n₂` in both diagrams. - **Diagram A Specific**: - Label `S` positioned below the `Q` arrow. - **Diagram B Specific**: - Three blue dots above the `Q` arrow, labeled `CCC=1`. - Two red squares below the `Q` arrow, labeled `2` and `0`. ### Detailed Analysis - **Diagram A**: - Minimalist representation of a system with two nodes (`n₁`, `n₂`) and a bidirectional interaction (`Q`). - `S` likely represents a state, variable, or process associated with the interaction. - **Diagram B**: - **Blue Dots (`CCC=1`)**: Three dots above the `Q` arrow, possibly indicating a count or category (e.g., "CCC" as a metric or classification). - **Red Squares (`2`, `0`)**: Positioned below the `Q` arrow, these could represent binary states (e.g., active/inactive) or counts (e.g., "2" active instances, "0" inactive). ### Key Observations 1. **Structural Similarity**: Both diagrams share the core structure of `n₁` ↔ `Q` ↔ `n₂`, suggesting a foundational relationship. 2. **Diagram B Annotations**: - `CCC=1` and the three blue dots imply a quantifiable or categorical attribute tied to the interaction. - The red squares (`2`, `0`) introduce binary or discrete states, contrasting with the continuous interaction in Diagram A. 3. **Positioning**: - In Diagram B, annotations (`CCC=1`, `2`, `0`) are spatially separated from the core nodes, suggesting they are secondary or contextual to the primary interaction (`Q`). ### Interpretation - **System Dynamics**: - Diagram A represents a basic bidirectional relationship (`Q`) between two nodes, with `S` as a modifier or state. - Diagram B expands this model by introducing **quantitative** (`CCC=1`) and **binary** (`2`, `0`) parameters, possibly reflecting real-world constraints or operational states. - **Potential Use Cases**: - The diagrams could model processes like data flow (`Q`), state transitions (`S`), or resource allocation (`CCC=1`, `2`, `0`). - The red squares (`2`, `0`) might indicate thresholds (e.g., "2" active nodes, "0" failures) or binary flags for system health. - **Anomalies**: - The absence of a legend for the blue dots and red squares in Diagram B leaves their exact meaning ambiguous. However, their placement suggests they are directly tied to the `Q` interaction. ### Conclusion The diagrams illustrate a hierarchical system where core interactions (`Q`) are modified by additional parameters (`S`, `CCC=1`, `2`, `0`). Diagram B’s annotations likely represent measurable or categorical extensions of the base model in Diagram A, emphasizing the system’s adaptability to external conditions or states. </details> S Fig. 2. (a) Two nodes with data and signal queues Q and S between them; (b) A possible state of the signal protocol, showing the credit associated with each signal (lower edge) and the current credit counter (right). n 2 may consume one data item from Q before the first signal, then another two before the second signal. be consumed given the available space on n 's downstream data queue. The maximum number of signals emitted per data item or signal input is also known a priori , so a similar determination can be made given n 's downstream signal queue. The scheduler repeatedly chooses some fireable node and fires it until no node has queued data or signals remaining. Lemma 2. Under the given firing/scheduling policy, an application pipeline always finishes execution in finite time and so cannot deadlock. ## 3.3 SIMD Extensions The above description assumes that nodes process input data items one at a time. However, on a SIMD-parallel processor, a node may process an ensemble of multiple items at once. Because a signal updates the state of the receiving node, we must ensure that items appearing before and after a signal in the stream are not processed in the same input ensemble . Hence, if a signal is queued for a node, the system must limit the size of the node's input ensemble to the value of its current credit counter. This requirement may adversely impact SIMD occupancy if signals occur frequently; we study its impact in Section 5. ## 4 Regional Context via Enumeration and Aggregation We now describe a developer-facing abstraction, enumeration and aggregation , that allows application developers to describe streaming computations in which regions of a stream must be processed in a common context. We have implemented this abstraction as an extension to our MERCATOR system, building on the work of the previous section. Our abstraction assumes that regions of a stream with a common context are represented as composite objects , similar to the RDDs of Apache Spark [16]. The actual input provided to the application is a stream of such objects. Each object may contain zero or more elements of a common data type. For example, an object could be a line of text whose elements are characters, or a vertex whose elements are its adjacent edges, or a list whose elements are numbers. Objects may contain different numbers of elements. At a given point in the application's pipeline, the developer may choose to 'open' the stream of composite objects to create a stream of all their elements. We call this opening process enumeration of the objects. The enumerated element stream becomes the input to the next node in the pipeline. In this and subsequent nodes, the developer may access the parent object that gave rise to an input item to obtain context needed for its processing. The opposite of enumeration is aggregation , which 'closes' the context associated with a parent object. The developer may choose to emit a stream of results derived from individual elements, stripped of their parent context, or to aggregate values computed from the elements of each parent object (e.g. by summing them) and emit a single result per parent. Either way, the stream of results continues down the pipeline. ## 4.1 Developer Interface To make these ideas concrete, consider the simple application whose topology is sketched in Figure 3. A stream of objects of type Blob , each containing a collection of numbers, flows from the source node. The Blobs' elements are enumerated, and node f does some computation on each number in the element stream, producing a (possibly shorter) output stream of numbers. These results are passed to node a , which sums the results from each Blob and sends a stream of per-Blob sums to a sink node. Fig. 3. A pipeline with enumeration and aggregation. The computation enumerates composite objects drawn from an input source, acts on their elements with a filtering node f , aggregates the filtered values in an accumulator node a , and writes the accumulated value from each object to an output sink. <details> <summary>Image 3 Details</summary>  ### Visual Description ## Flowchart: Data Flow Process ### Overview The image depicts a sequential data flow process represented as a flowchart. It consists of four primary components connected by directional arrows, with two intermediate processing steps labeled "enumerate" and "aggregate". The flow progresses from left to right, indicating a transformation pipeline. ### Components/Axes 1. **Nodes**: - **src**: Oval-shaped node on the far left, labeled "src" (source) - **f**: Oval-shaped node connected to "src", labeled "f" (function) - **a**: Oval-shaped node connected to "f", labeled "a" (aggregation) - **snk**: Oval-shaped node on the far right, labeled "snk" (sink) 2. **Processing Steps**: - **enumerate**: Blue vertical rectangle between "src" and "f", labeled "enumerate" - **aggregate**: Blue vertical rectangle between "f" and "a", labeled "aggregate" 3. **Connectors**: - Black arrows indicate directional flow between all components - Arrows maintain consistent thickness and style throughout 4. **Visual Style**: - White background - Black text for labels - Blue rectangles for processing steps with white text - Oval shapes for nodes with black borders ### Detailed Analysis - **src → enumerate → f**: Data originates at "src", undergoes enumeration processing, then passes through function "f" - **f → aggregate → a**: Processed data moves through aggregation step before reaching node "a" - **a → snk**: Final data transfer to the sink node "snk" - All components maintain equal vertical alignment, creating a straight horizontal flow ### Key Observations 1. The process follows a strict linear progression without branching or feedback loops 2. Processing steps are visually distinguished by blue rectangles, while nodes use oval shapes 3. No numerical values or quantitative data present in the diagram 4. Consistent use of black for structural elements (arrows, node borders, text) 5. Blue color specifically reserved for processing steps, creating visual hierarchy ### Interpretation This flowchart represents a typical data processing pipeline where: 1. Raw data ("src") is first enumerated (likely converted into a sequence/collection) 2. The enumerated data undergoes functional transformation ("f") 3. Results are aggregated before reaching intermediate node "a" 4. Final output is delivered to the sink ("snk") The diagram suggests a batch processing model where data is transformed through discrete stages. The use of "enumerate" and "aggregate" implies potential operations like: - Enumeration: Indexing or converting to a sequence - Aggregation: Summarizing or combining data elements The absence of feedback loops or parallel processing paths indicates a strictly sequential workflow, which might be optimized for simplicity or specific data characteristics requiring ordered processing. </details> The listing of Figure 4 specifies the application's topology. For each node, we specify the types of its input and output streams. The enumerate keyword at the input to node f indicates that Blobs are to be enumerated starting there; subsequent data types in the enumeration region of the pipeline are labeled from Blob , indicating that items are to be processed in the context of their parent Blobs. Aggregation occurs at the output of node a , where the aggregate keyword indicates that a produces (up to) one double value per parent object rather than per element. Figure 5 shows code to implement the application. For each node, there is a run() function that processes items in the node's input stream. Output from a node is generated via the push() function; because the application is irregular, not every input might produce an output. The listing shows several functions specific to enumeration and aggregation. findCount() is called once per parent object to determine how many elements it contains. The begin() and end() functions, which may be defined for each node receiving enumerated inputs, are executed before and after the region of the stream associated with each parent object, respectively. The parent object associated with a node's current input is accessible via getParent() . Enumeration produces a stream of sequential indices of elements in each parent object. However, the application developer is responsible for providing code to extract the elements from an object. This design allows MERCATOR to remain ignorant of how objects are organized internally. ## 4.2 Implementation The MERCATOR system takes in an application topology, as shown in Figure 4, and produces stubs for all the functions shown in Figure 5. The user then fills in the function bodies with the actual code of the application. We note that the code shown is CUDA, not C++ ; hence, the run() functions are actually called not with a single input but with a SIMD ensemble of items in multiple threads, which execute the function body in parallel for each item. (The ``` ``` Fig. 4. Application topology specification illustrating enumeration and aggregation. ``` ``` Fig. 5. Application code, with stubs generated from topology and developer-supplied function bodies. accumulation in node a would in practice be implemented atomically or with a SIMD-parallel reduction.) MERCATOR provides the runtime infrastructure needed to transfer data from one node to the next and to schedule nodes. The signaling mechanism of the previous section is key to enabling enumeration and aggregation. At the point of enumeration, the runtime generates a data stream of element indices together with signals indicating the start and end of each parent object's elements. Downstream nodes intercept these signals in order to update their current parent object and to call the begin() and end() stubs at the right times. Because data before and after a signal is never processed in the same SIMD ensemble, operations on different parent objects' elements always happen in separate calls to a node's run() function, and the result of getParent() is the same for all items in an ensemble. ## 5 Results We implemented both our precise signaling infrastructure and the enumeration and aggregation abstraction as extensions to our MERCATOR framework and studied their performance on several benchmark computations. All experiments were conducted on an NVIDIA GTX 1080Ti GPU (28 processors), using as many active blocks as could fit on the device and a SIMD width of 128 threads per block. Code was compiled using CUDA v10 under Linux. The abstraction penalty of the new features was verified to be negligible in MERCATOR applications that do not use them. Cost of Regional Context Abstraction To characterize the performance impact of regional context, we began with two simple benchmark computations. Each benchmark operates on a large array of integers in GPU memory, and each divides this array into a series of regions. The computation enumerates each region, sums its elements, and produces a stream of per-region sums. In the first benchmark, the regions are of uniform size; in the second, the size of each region is chosen uniformly at random between 0 and a specified maximum. Fig. 6. Execution time vs. region size for sum app with fixed-size regions. <details> <summary>Image 4 Details</summary>  ### Visual Description ## Line Graph: Execution Time vs. Region Size ### Overview The image depicts a line graph illustrating the relationship between **region size** (x-axis) and **execution time** (y-axis). The graph shows a sharp initial decline in execution time as region size increases, followed by a gradual plateau. Data points are represented by blue squares connected by a line. --- ### Components/Axes - **Y-Axis (Left)**: - Label: **Execution Time (s)** - Scale: 0 to 40 seconds (increments of 10). - **X-Axis (Bottom)**: - Label: **Region Size** - Scale: 256 to 2048 (increments of 256). - **Data Series**: - Blue squares connected by a line. - No explicit legend, but the color (blue) and marker (square) are consistent across all points. --- ### Detailed Analysis - **Data Points**: - Region Size 256: Execution Time ≈ 45s - Region Size 512: Execution Time ≈ 25s - Region Size 768: Execution Time ≈ 12s - Region Size 1024: Execution Time ≈ 8s - Region Size 1280: Execution Time ≈ 6s - Region Size 1536: Execution Time ≈ 5s - Region Size 1792: Execution Time ≈ 4.5s - Region Size 2048: Execution Time ≈ 4s - **Trends**: - **Initial Drop**: Execution time decreases sharply from 45s (256) to 25s (512), a 44% reduction. - **Gradual Decline**: From 512 to 2048, execution time decreases more slowly, stabilizing near 4s. --- ### Key Observations 1. **Outlier at 256**: The execution time at the smallest region size (256) is significantly higher than subsequent values, suggesting inefficiency at lower scales. 2. **Diminishing Returns**: Beyond 512, execution time reductions become marginal (e.g., 12s → 8s at 1024, then 6s → 5s at 1536). 3. **Stabilization**: Execution time plateaus near 4s for region sizes ≥1792, indicating a potential threshold for optimization. --- ### Interpretation The graph suggests that execution time is inversely proportional to region size up to a critical point (≈512), after which further increases in region size yield minimal performance gains. This could imply: - **Algorithmic Efficiency**: Larger regions may leverage parallelism or batch processing, reducing per-unit computation time. - **Resource Saturation**: Beyond a certain size, system resources (e.g., memory, CPU) may become the limiting factor, capping execution time improvements. - **Practical Threshold**: For applications requiring sub-5s execution, scaling beyond 1792 offers little benefit. The sharp initial drop highlights the importance of optimizing for mid-range region sizes, while the plateau suggests diminishing returns at scale. This pattern is critical for resource allocation and system design decisions. </details> Fig. 7. Execution time vs. max region size for sum app with variable regions. <details> <summary>Image 5 Details</summary>  ### Visual Description ## Line Graph: Execution Time vs. Max Region Size ### Overview The image depicts a line graph illustrating the relationship between execution time (in seconds) and maximum region size. The graph shows a sharp decline in execution time as the maximum region size increases, followed by a plateau effect at larger region sizes. ### Components/Axes - **Y-Axis (Left)**: Labeled "Execution Time (s)" with a scale from 0 to 100 seconds in increments of 20. - **X-Axis (Bottom)**: Labeled "Max Region Size" with values: 256, 512, 768, 1024, 1280, 1536, 1792, 2048. - **Data Series**: A single blue line with square markers connects data points at each x-axis value. - **Legend**: Not explicitly visible in the image, but the line is described as blue with square markers. ### Detailed Analysis - **Data Points**: - 256: ~100 seconds - 512: ~50 seconds - 768: ~20 seconds - 1024: ~10 seconds - 1280: ~8 seconds - 1536: ~7 seconds - 1792: ~6 seconds - 2048: ~5 seconds - **Trend**: The execution time decreases exponentially as the maximum region size increases. The steepest drop occurs between 256 and 512, after which the decline slows significantly, approaching a plateau near 5 seconds for region sizes ≥1024. ### Key Observations 1. **Exponential Decline**: Execution time halves when the region size doubles from 256 to 512, then continues to drop rapidly until 1024. 2. **Diminishing Returns**: Beyond 1024, execution time stabilizes, with only marginal improvements (e.g., 10s → 8s → 7s → 6s → 5s). 3. **Outlier**: The initial execution time at 256 (100s) is an outlier compared to the trend observed at larger region sizes. ### Interpretation The data suggests that increasing the maximum region size significantly improves performance up to a critical threshold (~1024). Beyond this point, further increases yield negligible benefits, indicating an optimal region size for efficiency. The sharp decline at smaller sizes may reflect computational overhead or inefficiencies in processing smaller regions, while larger regions likely leverage parallelism or batch processing to reduce execution time. The plateau effect implies a saturation point where system resources (e.g., memory, CPU) are fully utilized, limiting further optimization. </details> Figures 6 and 7 show the time to process an array of 512 million integers in each benchmark as a function of the region size (fixed in the first figure, maximum in the second). Focusing first on the test with fixed-sized regions, we see that execution time decreases sharply as the region size grows from 32 to the SIMD width of 128, then decreases more gradually for larger sizes. This decrease reflects the lower frequency of region boundary signals relative to the data stream as the region size increases. For region sizes on the order of several hundred of elements or more, the abstraction overhead is small relative to the total cost of execution. A second phenomenon observable in Figure 6 is that the overhead incurred by region boundaries changes non-monotonically with region size. In particular, overhead is locally minimized for region sizes equal to a multiple of the SIMD width and then jumps sharply for slightly larger sizes. This behavior reflects the impact of region boundaries on SIMD occupancy. Recall that signals prevent elements in two regions from being combined in the same SIMD ensemble, which is required to ensure that each element contributes only to its own region's sum. Region sizes that do not evenly divide the SIMD width therefore require that nodes run with non-full input ensembles at least once per region. This loss of SIMD occupancy appears as reduced application throughput. For region sizes less than the SIMD width, every ensemble becomes non-full, which explains the large performance impacts seen at region sizes below 128. Figure 7 shows a much-reduced impact of small size variations on throughput. Unlike the previous benchmark, but more typically of real-world irregular applications, the region size is not fixed. The sharp peaks of reduced throughput for worst-case region sizes are therefore smoothed out, but the dominant effect remains: larger region sizes incur less abstraction overhead. Comparison of Mechanisms for Communicating Context For our second experiment, we implemented a real-world application taken from the DIBS benchmark set [3], a suite of applications representative of data integration workloads. The application, which DIBS calls tstcsv-> csv but we refer to hereafter as 'taxi,' processes a sequence of lines of text, each of which contains a tag, a variablelength list of GPS locations specified as real-valued coordinate pairs, and other data. The goal is to parse each coordinate pair, swap the elements of the pair, and emit the pair together with the tag corresponding to its source line. Our initial implementation of the taxi application operates on the raw text in GPU memory. It takes as input a stream of line start indices and line lengths. For each line, the first stage of the application enumerates the line's individual characters as a stream, checks them in parallel, and retains only those character positions (identified by an open-brace character) that likely mark the start of a coordinate pair. The second stage verifies, again in parallel, that each open-brace indeed marks a coordinate pair and, if so, parses the pair's coordinates. Each line's tag is parsed once when the line is first enumerated and is then used to mark each parsed coordinate pair for that line. The first series of Figure 8 (square points) shows the execution time of the taxi app as a function of its input size. Larger file sizes were obtained by replicating the input file from DIBS multiple times. Mindful of the relationship between abstraction penalty and region size, we then investigated how the input data in the taxi app determined SIMD occupancy. Input lines have an average length of 1397 characters, so regions corresponding to each line in the stage 1 are large, and the penalty to occupancy is expected to be low. In contrast, lines contain on average only 45 coordinate pairs, less than the SIMD width, and so would be expected to incur a large penalty to occupancy in stage 2, whose region size is determined by the number of pairs per line. Indeed, we found that stage 1 was fired with full SIMD ensembles 91% of the time, while stage 2 had full ensembles only 9% of the time. When regional context changes frequently relative to the SIMD width, the occupancy cost to performance of our implementation may exceed the cost (mainly extra memory accesses) of replicating this context along with every data item. We therefore developed a second version of the taxi app that used enumeration to provide context in stage 1 but explicitly marked each open-brace with its line's tag before sending it to stage 2. The latter stage does not utilize the enumeration abstraction and so can process items from multiple lines in one ensemble, achieving essentially full SIMD occupancy. The second series in Figure 8 (triangular points) shows that improved occupancy results in lower total execution time. However, using the same strategy to tag each character of each line in stage 1, while it slightly improves occupancy by avoiding enumeration entirely, incurs substantially more overhead due to the much greater number of elements to be tagged per region. The third series (x points) shows that, at the largest input size tested, a pure tagging implementation is roughly 30% slower than one that judiciously uses either tagging or our design as appropriate for each stage. We conclude that the best way to provide regional context to streaming applications on a SIMD architecture depends strongly on the performance tradeoff between reduced SIMD occupancy and reduced representation overhead. Each stage of a pipeline may represent a different point in this tradeoff, and the highest-performing implementation (dense or sparse) for regional context may Fig. 8. Execution time vs input size for three versions of the taxi app. <details> <summary>Image 6 Details</summary>  ### Visual Description ## Line Graph: Execution Time vs. Input Size ### Overview The image is a line graph comparing execution time (in seconds) across three scenarios as input size increases from 437KB to 1728KB. The three scenarios are: 1. **Both stages enumerated** (blue squares) 2. **Stage 1 enumerated** (red triangles) 3. **Neither stage enumerated** (orange crosses) ### Components/Axes - **X-axis (Input size)**: Labeled "Input size (Kilobytes)" with values: 437, 864, 1.69K, 3.38K, 6.75K, 13.5K, 27K, 54K, 108K, 216K, 432K, 864K, 1728K. - **Y-axis (Execution Time)**: Labeled "Execution Time (s)" with a scale from 0 to 10. - **Legend**: Positioned in the top-right corner, with three entries: - Blue squares: "Both stages enumerated" - Red triangles: "Stage 1 enumerated" - Orange crosses: "Neither stage enumerated" ### Detailed Analysis 1. **Both stages enumerated (blue squares)**: - Starts at ~0.5s for 437KB. - Gradual increase to ~0.8s at 1.69K. - Accelerates sharply after 864K, reaching ~9s at 1728K. 2. **Stage 1 enumerated (red triangles)**: - Starts at ~0.5s for 437KB. - Slightly lower than blue line at all points (e.g., ~0.7s at 864K). - Reaches ~8s at 1728K. 3. **Neither stage enumerated (orange crosses)**: - Starts at ~0.5s for 437KB. - Gradual rise to ~3s at 108K. - Steep increase after 864K, peaking at ~10s at 1728K. ### Key Observations - All three scenarios show increasing execution time with larger input sizes. - **Both stages enumerated** (blue) and **Stage 1 enumerated** (red) exhibit similar trends, with blue consistently slightly higher. - **Neither stage enumerated** (orange) diverges significantly at larger inputs, showing the steepest rise after 864K. - At 1728K, execution times are: - Blue: ~9s - Red: ~8s - Orange: ~10s ### Interpretation The data suggests that enumerating both stages (blue) is the most efficient approach, while enumerating only Stage 1 (red) offers marginal improvements. **Not enumerating either stage** (orange) leads to significantly worse performance, particularly at larger input sizes. The sharp rise in the orange line after 864K indicates that skipping enumeration becomes increasingly costly as data grows. This highlights the importance of enumeration strategies in optimizing execution time for large datasets. </details> therefore vary between stages. Ultimately, this choice should be made transparently to the application developer based on profile-guided feedback. ## 6 Conclusion and Future Work We have described an abstraction, enumeration and aggregation, to support stateful streaming computation based on regional contexts. We presented an implementation of this abstraction for irregular streaming computations on SIMDparallel architectures such as GPUs. Our abstraction relies on a sparse implementation of precise signal delivery between computational stages. We characterized the cost of the abstraction on benchmark computations and demonstrated that the best strategy for realizing it may depend on the relationship between region size and the architecture's SIMD width. Future work will include more careful modeling and/or empirical measurement of the costs of alternative implementations, with an eye toward allowing the MERCATOR runtime to transparently choose between strategies based on the typical number of elements per region. Another direction for future work will investigate how to lower the abstraction penalty of precise signaling for SIMD occupancy. When the effects of a signal on a node's state are limited and well-defined (e.g. changing the parent object pointer), the node may be able to compute the correct state (pre- or post-signal) to expose to the item in each SIMD lane separately. Computing the correct state per item in each node, rather than storing it with items in the queues between nodes, would offer the same efficient representation of state as in our design while eliminating signals' cost to SIMD occupancy. ## Acknowledgments This work was supported by NSF CISE awards CNS-1763503 and CNS-1500173. ## References 1. Altschul, S., Gish, W., Miller, W., Myers, E., Lipman, D.: Basic local alignment search tool. J. Molecular Biology 215 (3), 403-10 (1990) 2. Barnes, J., Hut, P.: A hierarchical o(n log n) force-calculation algorithm. Nature 324 (6096), 446 (1986) 3. Cabrera, A.M., Faber, C.J., Cepeda, K., Derber, R., Epstein, C., Zheng, J., Cytron, R.K., Chamberlain, R.D.: DIBS: A data integration benchmark suite. In: 2018 ACM/SPEC Int'l Conf. Performance Engineering. pp. 25-28 (2018) 4. Chamberlain, R., Franklin, M., Tyson, E., Buckley, J., et al.: Auto-Pipe: streaming applications on architecturally diverse systems. Computer 43 (3), 42-49 (2010) 5. Cole, S., Buhler, J.: MERCATOR: A GPGPU framework for irregular streaming applications. In: 2017 Int'l Conf. High Performance Computing and Simulation. pp. 727-36 (2017) 6. Eker, J., Janneck, J., Lee, E.A., Liu, J., Ludvig, J., Sachs, S., , Xiong, Y.: Taming heterogeneity - the Ptolemy approach. Proc. IEEE 91 (1), 127-44 (2003) 7. Grossman, M., Simion Sbˆ ırlea, A., Budimli´ c, Z., Sarkar, V.: CnC-CUDA: Declarative programming for GPUs. In: Cooper, K., Mellor-Crummey, J., Sarkar, V. (eds.) Languages and Compilers for Parallel Computing. pp. 230-245. Springer Berlin (2011) 8. Lee, E., Messerschmitt, D.: Synchronous data flow. Proc. IEEE 75 (9), 1235-245 (1987) 9. Li, P., Agrawal, K., Buhler, J., Chamberlain, R.: Orchestrating safe streaming computations with precise control. In: 4th Int'l Workshop on Extreme Scale Computing Application Enablement - Modeling and Tools. pp. 1017-22 (2014) 10. Roesch, M., et al.: Snort: Lightweight intrusion detection for networks. In: Proc. 13th Systems Administration Conf. (LISA). pp. 229-238 (1999) 11. Thies, W., Karczmarek, M., Amaransinghe, S.: StreamIt: A language for streaming applications. In: 11th Int'l Conf. Compiler Construction. pp. 179-96 (2002) 12. Thies, W., Karczmarek, M., Sermulins, J., Rabbah, R., Amaransinghe, S.: Teleport messaging for distributed stream programs. In: 10th ACM SIGPLAN Symp. Principles and Practice of Parallel Programming. pp. 224-35 (2005) 13. Tyson, E., Buckley, J., Franklin, M., Chamberlain, R.D.: Acceleration of atmospheric Cherenkov telescope signal processing to real-time speed with the AutoPipe design system. Nuclear Instruments and Methods in Physics Research Sec. A: Accelerators, Spectrometers, Detectors and Associated Equipment 595 (2), 474-9 (2008) 14. Viola, P., Jones, M.: Rapid object detection using a boosted cascade of simple features. In: Proc. IEEE Comp. Soc. Conf. Computer Vision and Pattern Recognition (2001) 15. Zaharia, M., Chowdhury, M., Das, T., Dave, A., et al.: Resilient distributed datasets: A fault-tolerant abstraction for in-memory cluster computing. In: 9th USENIX Conf. Networked Systems Design and Implementation. p. 2 (2012) 16. Zaharia, M., Xin, R., Wendell, P., Das, T., Armbrust, M., Dave, A., Meng, X., Rosen, J., Venkataraman, S., Franklin, M., Ghodsi, A., Gonzalez, J., Shenker, S., Stoica, I.: Apache Spark: a unified engine for big data processing. Communications of the ACM 59 (11), 56-65 (2016) ## 7 Appendix: Proofs Omitted in Text ## 7.1 Proof of Lemma 1 Proof. We proceed by induction on the number of signals already on the signal queue S when s is emitted. Suppose S is empty when n 1 emits s . All items emitted by n 1 prior to s either have been consumed by n 2 or are present on Q . The protocol assigns s an amount of credit equal to the size of Q . This amount is then added to n 2 's current credit counter, which was previously 0. Finally, n 2 consumes s precisely when its current credit counter returns to 0, which happens once n 2 consumes the items that were present on Q when s was emitted. Now suppose s is emitted when S is not empty. s receives a number of credits equal to the number of items added to Q since the prior signal s ′ . We know inductively that n 2 consumes s ′ precisely when all data items emitted prior to s ′ have been consumed. At this point, the only items on Q must be those emitted after s ′ but before s , and n 2 's current credit counter is 0 since it just consumed a signal. Conclude that n 2 will transfer the credit in s to its current credit counter and will then consume exactly those data items emitted after s ′ but before s before it consumes s itself. ## 7.2 Proof of Lemma 2 Proof. Our proof of deadlock-freedom relies on the following two claims. Claim. A node cannot have a current credit counter > 0 without a pending data item. Proof of claim : Suppose that a node's current credit counter is > 0. The credit in the current counter was transferred from the signal s currently at the head of the signal queue; it cannot remain from prior signals because the node did not even check for s until its current credit counter last became 0. Hence, this credit was assigned to s to cover data items that were enqueued at the time that s was issued. Since not all credit has yet been consumed, at least one of these items is still enqueued. Claim. If a node n has either pending data or a pending signal, one of the following holds: (1) n can consume a data item; (2) n can consume a signal; (3) n is blocked due to insufficient space in its downstream queues. Proof of claim : The node either has credit or not. If it has credit, then by the previous claim it has pending data that can be consumed. If it has no credit but has a pending signal, then either the signal can be consumed, or credit can be transferred from the signal; in the latter case, there must again be data corresponding to this credit. If there is no credit and no pending signal, then there must be pending data, which can be consumed without credit in the absence of pending signals. In all cases, the node can consume some input unless its downstream queues lack sufficient space. We now proceed to prove the original lemma. If any node in the pipeline has pending data or a signal, then let n be the last such node. Either n 's downstream queues are empty (else its successor would have pending data or signals), or n has no successor, i.e., it is the last node in the pipeline, which has unbounded output space and cannot block. Hence, node n is not blocked on its downstream queues and so, by the second claim, can be fired to consume input. We conclude that the application terminates only when all nodes have exhausted their inputs.