Weighted Matching in a Poly-Streaming Model

Data-intensive computations arise in data science, machine learning, and science and engineering disciplines. These datasets are often massive, generated dynamically, and, when stored, kept in distributed formats on disks, making them amenable to processing as multiple data streams. The modular feature of these datasets can be exploited by streaming algorithms designed for tightly-coupled shared-memory and distributed-memory multiprocessors to efficiently solve large problem instances that offline algorithms cannot handle due to their high memory requirements. However, the design of parallel algorithms that process multiple data streams concurrently has not yet received much attention.

Current multicore shared-memory processors consist of up to a few hundred cores, organized hierarchically to share caches and memory controllers. These cores compute in parallel to achieve speedups over serial execution. With multiple memory controllers, I/O operations can also proceed in parallel, and this feature can be used to process multiple data streams concurrently. These I/O capabilities and the limitations of offline algorithms motivate a model of computation, illustrated in Figure 1 and discussed next.

The streaming model of computation allows $o\left(N\right)$ space for a data stream of size $N$ [2, 13]. For graphs, the semi-streaming model permits $𝒪\left(n\cdot \text{polylog }n\right)$ space for a graph with $n$ vertices and an edge stream of arbitrary length [8]. Building on these space-constrained models, we introduce the poly-streaming model. The key aspects of our model are as follows.

We consider $k$ data streams that collectively contain $N$ items. An algorithm has access to $k$ (abstract) processors, and is allowed $𝒪\left(f(k)\cdot M_1\right)$ total space, where $M_1$ is either $o\left(N\right)$ or the space permitted to a single-stream algorithm. In each pass, each stream is assigned to one of the processors, and each processor independently reads one item at a time from its stream and processes it. Processors may communicate as needed, either via shared or remote memory access. Algorithms are assessed on several metrics: space complexity, number of passes, per-item processing time, total runtime, communication cost, and solution quality.

In the poly-streaming model, we address the problem of approximating a maximum weight matching (MWM) in an edge-weighted graph, where the goal is to find a set of vertex-disjoint edges with maximum total weight. We design an algorithm for approximating an MWM when the graph is presented as multiple edge streams. Our design builds on the algorithm of [20] and adds support for handling multiple streams concurrently. We also generalize the design to NUMA (non-uniform memory access) multiprocessor architectures.

We summarize our contributions to the MWM problem as follows. Let $L_{max}$ and $L_{min}$ denote the maximum and minimum lengths of the input streams, respectively, and let $n$ denote the number of vertices in a graph $G$. For any realization of the CREW PRAM model (such as in Figure 1), we have the following result.

In NUMA architectures, memory access costs depend on a processor’s proximity to the target memory. We generalize the algorithm in Theorem 1 to account for these cost differences. In particular, we show that when $k$ processors are partitioned into $r$ groups, each with its own shared local memory, the total number of global memory accesses across all groups is $\widetilde{𝒪}\left(r\cdot n\right)$. This generalization preserves all other performance guarantees from Theorem 1, except that the $\widetilde{𝒪}\left(k+n\right)$ space bound becomes $\widetilde{𝒪}\left(k+r\cdot n\right)$. These results are formalized in Theorem 18 in Section 4. This design gives a memory-efficient algorithm for the NUMA shared memory multiprocessors, on which we report empirical results.

We have evaluated our algorithm on a NUMA machine using graphs with billions to trillions of edges. For most of these graphs, our algorithm uses space that is orders of magnitude smaller than that required by offline algorithms. For example, storing the largest graph in our evaluation would require more than $91,600\mathrm{GB}$ ($\approx 90\mathrm{TB}$), whereas our algorithm used less than $1\mathrm{GB}$. Offline matching algorithms typically require even more memory to accommodate their auxiliary data structures.

We employ approximate dual variables that correspond to a linear programming relaxation of MWM to obtain a posteriori upper bounds on the weights of optimal matchings. These bounds allow us to compare the weight of a matching produced by our algorithm with the optimal weight. Thus, we show that our algorithm produces matchings whose weights significantly exceed the approximation guarantee.

For $k=128$, our algorithm achieves runtime speedups of $16$–$83$ across all graphs in our evaluation, on a NUMA machine with only $8$ memory controllers. This is significant scaling for a poly-streaming algorithm, given that $8$ memory controllers are not sufficient to serve the concurrent and random access requests of $128$ processors without delays. Nevertheless, these speedups demonstrate the effectiveness of our design, which accounts for a processor’s proximity to the target memory. A metric less influenced by memory latency suggests that the algorithm would achieve even better speedups on architectures with more efficient memory access.

Note that Theorem 1 and Theorem 18 both guarantee $\widetilde{𝒪}\left(L_{max}+n\right)$ runtime. For $L_{max}=\mathrm{\Omega }\left(n\right)$, this is tight up to polylogarithmic factors. However, by using $\widetilde{𝒪}\left(k\cdot n\right)$ space and $𝒪\left(1\right)$ per-edge processing time, we can even achieve $\widetilde{𝒪}\left(L_{max}+n/k\right)$ runtime, which becomes polylogarithmic for large values of $k$ (see the arXiv version).

For a graph $G=(V,E)$, let $n:=|V|$ and $m:=|E|$ denote the number of vertices and edges, respectively. We denote an edge $e:=\{u,v\}$ by the unordered pair of its endpoints. For a weighted graph, let $w_e$ denote the weight of edge $e$, and for any subset $A\subseteq E$, define $w(A):={\sum }_{e\in A}{w}_e$. For $\mathrm{\ell }\in [k]$, let $E^{\mathrm{\ell }}$ be the set of edges received in the $\mathrm{\ell }$th stream. Define $L_{max}:={\max }_{\mathrm{\ell }\in [k]}|E^{\mathrm{\ell }}|$ and $L_{min}:={\min }_{\mathrm{\ell }\in [k]}|E^{\mathrm{\ell }}|$.

A matching $ℳ$ in a graph is a set of edges that do not share endpoints. A maximum weight matching (MWM) ${ℳ}^{\ast }$ is a matching with maximum total weight; that is, $w\left({ℳ}^{\ast }\right)\ge w\left(ℳ\right)$ for all matchings $ℳ\subseteq E$.

A $\rho$-approximation algorithm computes a solution whose value is within a factor $\rho$ of optimal. The factor $\rho$ is called the (worst-case) approximation ratio. We assume $\rho \ge 1$ for both maximization and minimization problems. Thus, for maximization, a $\rho$-approximation guarantees a solution whose value is at least $\frac{1}{\rho }$ times the optimal.

We use the linear programming (LP) relaxation of the MWM problem, and its dual, shown in Figure 2. In the primal LP, each variable $x_e$ is $1$ if edge $e$ is in the matching and $0$ otherwise. Each $y_u$ is a dual variable, and $\delta (u)$ denotes the set of edges incident on a vertex $u$. Let ${\{x_e\}}_{e\in E}$ and ${\{y_u\}}_{u\in V}$ be feasible solutions to the primal and dual LPs, respectively. By weak LP duality, we have ${\sum }_{e\in E}{w}_e{x}_e\le {\sum }_{u\in V}{y}_u$. If ${\{x_e\}}_{e\in E}$ is an optimal solution to the primal LP, then $w\left({ℳ}^{\ast }\right)\le {\sum }_{e\in E}{w}_e{x}_e\le {\sum }_{u\in V}{y}_u$. The first inequality holds because the primal LP is a relaxation of the MWM problem.

In this section, we present the design and analyses of our algorithm in Theorem 1 that assumes a uniform memory access cost.

In this section, we extend the algorithm from Section 3 to account for the non-uniform memory access (NUMA) costs present in real-world machines.

In a poly-streaming algorithm, each processor may receive an arbitrary subset of the input, making it difficult to maintain memory access locality. Modern shared-memory machines, as illustrated in Figure 1, have non-uniform memory access costs and far fewer memory controllers than processors [12]. As a result, memory systems with such limitations would struggle to handle the high volume of concurrent, random memory access requests generated by poly-streaming algorithms, leading to significant delays.

We now describe a generalization of the algorithm from Section 3 that localizes a significant portion of each processor’s memory access to its near memory. This generalization applies to both edge-processing strategies introduced in Section 3.1. We focus on the non-deferrable strategy. (The deferrable strategy generalizes in the same way, following the same relationship between the two strategies as in the specialized case.)

The runtime of Process-Edge is dominated by the time to access the dual variables ${\{{\alpha }_u\}}_{u\in V}$. By assigning a dedicated stack to each processor, we have substantially localized accesses associated with edges in that stack. However, since a large fraction of edges is typically not included in the stacks, the runtime remains dominated by accesses to dual variables associated with these discarded edges. We therefore describe an algorithm that localizes these accesses to memory near the processor.

To localize accesses to the dual variables ${\{{\alpha }_u\}}_{u\in V}$, we observe that these variables increase monotonically during the streaming phase. This observation motivates a design in which a subset of processors maintains local copies of the variables and can discard a substantial number of edges without synchronizing with the global copy. When a processor includes an edge in its stack, it increments the corresponding dual variables in the global copy by the gain of the edge and synchronizes its local copy accordingly. As a result, some local copies may lag behind the global copy, but they can be synchronized when needed.

A general scheme for allocating dual variables is as follows. The set of $k$ processors is partitioned into $r$ groups. For simplicity, we assume that $k$ is a multiple of $r$, so each group contains exactly $k/r$ processors. For $r>1$, in addition to a global copy of dual variables, we maintain $r$ local copies ${\{{\alpha }_u^j\}}_{u\in V}$, one for each $j\in [r]$. Group $j$ consists of the processors $\{\mathrm{\ell }\in [k]\mid \lfloor \mathrm{\ell }/(k/r)\rfloor =j\}$, and uses ${\{{\alpha }_u^j\}}_{u\in V}$ as its local copy of the dual variables. Algorithm PS-MWM is the special case $r=1$, where all processors operate using only the global copy of the dual variables.

Algorithm PS-MWM-LD, along with its subroutine Process-Edge-LD, incorporates local dual variables in addition to global ones. In Step 2, processors in each group $j\in [r]$ collectively initialize their group’s local copies of dual variables and locks, followed by initializing a group lock in Step 3. All other steps of the algorithm are identical to those in PS-MWM.

In the subroutine Process-Edge-LD, Step 5 implements the non-deferable strategy. Steps 1–3 and Step 6 enforce the localization of access to dual variables. Steps 2–3 ensure that, at any given time, each global dual variable is accessed by at most one processor per group; we refer to this processor as the delegate of the group for that variable. Thus, a processor executing Steps 4–6 serves as a delegate of its group for that time. In Step 6, after completing updates to the global variables, the delegate synchronizes its group’s local copy in $𝒪\left(1\right)$ time. As a result, the waiting time on a local variable in Step 2 is bounded by the total time spent by the corresponding delegates, up to constant factors.

The delegates in each group handle vertex-disjoint edges, so concurrent executions of Step 6 would have been safe. However, the lock in Step 4 ensures that at most one delegate per group executes Step 5. Regardless of these design choices, the behavior of the delegates executing Step 5 concurrently mirrors that of processors competing for exclusive access to global dual variables in PS-MWM.

The following lemma highlights the benefit of using Algorithm PS-MWM-LD.

In contrast to the result in Lemma 17, the streaming phase of Algorithm PS-MWM accesses global variables $\mathrm{\Omega }\left(m\right)$ times or up to $𝒪\left(m+k\cdot n\log n\right)$ times.

Algorithm PS-MWM-LD, together with the generalization of the deferrable strategy, leads to the following result (a proof is included in the arXiv version).

This section summarizes our evaluation results for Algorithm PS-MWM-LD. Detailed datasets, experimental setup, and additional comparisons (including with PS-MWM) are provided in the arXiv version.

While numerous studies have focused on optimizing time (in parallel computing) or space (in streaming algorithms) in isolation, the poly-streaming model offers a practically relevant paradigm for exploring how time and space can be jointly optimized. It fills a gap by providing a formal framework for analyzing algorithmic design choices and their associated time-space trade-offs. Our study of matchings illustrates both the benefits of this paradigm and its practical relevance.

The simplicity of our matching algorithm and its generalization reflect our choice to adopt the design of [20]. We believe this principle will inspire the development of other poly-streaming algorithms. To this end, we note that [20] has also motivated simple algorithms for related problems, such as matchings with submodular objectives [16], $b$-matchings [14], and collections of disjoint matchings [9].