# Fast software pipelining for stream processing

I have a pipelined stream processing situation, where a number of streams is linked in the computational process S0 -> S1 -> S2 -> .... You can understand "stream" as a list of numbers.

In this process, S0 is initially given in totality. S1 is a mapped result from S0 according to function f; S2 is the result from S1 by the same f, so on and so on. f(S) works by traversing stream S linearly and produces the resulting stream as it goes. Hence, f can work on a stream "partially": that is, for example, we don't need the totality of stream S1 in order to begin producing S2. The processing S0->S1 can occur concurrently with S1->S2 in a pipelined fashion.

I attempted implementing this pipeline using C. In my code, I use an atomic variable to synchronize progress between each pipeline stage $$S_i\quad\overrightarrow{\small f_i}\quad S_{i+1}$$: basically, thread $$f_i$$ constantly polls progress from the previous pipeline $$f_{i-1}$$ to see if new tokens are written into $$S_i$$. If so, it continues its own work. Otherwise, it spins and waits. Of course, this assumes each $$f_i$$ is responsible for reporting its progress to $$f_{i+1}$$. It turns out that this synchronization is too costly. I wasn't able to gain any speedup using multiple threads. In fact, it makes the code much slower than a single-threaded implementation.

My question is, what is a better approach to implement this construction in order to actually take advantage of multi-threading? This seems to be very hard because the $$f$$ in my scenario is computational-bound. It seems to me highly unlikely to extract parallel performance unless there is a very low-cost synchronization scheme. Thanks in advance for any comments and advice.

• Are there latency requirements or do you care primarily about throughput?
– D.W.
Feb 15, 2021 at 23:41
• @D.W. The ultimate goal is to obtain the totality of the final stream $S_k$ as fast as possible, where $k$ is a (potentially very) large number. So I guess it's "throughput"?
– apen
Feb 15, 2021 at 23:45

Do the work in batches. Have a separate buffer for each stream that stores the unprocessed items. When you invoke $$f$$ for stream $$S_i$$, invoke 100 times on the 100 oldest items in the buffer for $$S_{i-1}$$, then remove those items from the buffer. This basically increases the granularity of operations, and should reduce the overhead from synchronization by about 100x. You can of course tune the number 100 to adjust throughput vs latency.