From 4e15957f7b5e21b459dd1fd2cce17d916d7635e8 Mon Sep 17 00:00:00 2001
From: Jan Niehusmann <jan@gondor.com>
Date: Sun, 19 Mar 2017 21:58:32 +0100
Subject: [PATCH] Avoid quadratic complexity in poll()

streaming::pipeline::server::Dispatch::poll() called poll on every
single entry of self.in_flight, and then removed at most one
element from that structure (via pop_front()).

So, the complexity is O(n^2) in the size of self.in_flight.

Calling poll() on the first entry of self.in_flight should be
sufficient, the other entries can be handled later. This reduces the
complexity to O(n).

This behavior can be significant if self.in_flight gets large, which can
happen if input is read very quickly. I triggered it with a simple
echo server fed from a unix domain socket:

$ time yes | head -n 1000 |  nc -U -N /tmp/sock >/dev/null
real	0m0.051s

$ time yes | head -n 10000 |  nc -U -N /tmp/sock >/dev/null
real	0m3.534s

$ time yes | head -n 20000 |  nc -U -N /tmp/sock >/dev/null
real	0m13.883s

With this patch, the runtime becomes linear and much faster:

$ time yes | head -n 1000 |  nc -U -N /tmp/sock >/dev/null
real	0m0.010s

$ time yes | head -n 10000 |  nc -U -N /tmp/sock >/dev/null
real	0m0.049s

$ time yes | head -n 20000 |  nc -U -N /tmp/sock >/dev/null
real	0m0.084s

$ time yes | head -n 100000 |  nc -U -N /tmp/sock >/dev/null
real	0m0.405s

$ time yes | head -n 1000000 |  nc -U -N /tmp/sock >/dev/null
real	0m3.738s
---
 src/streaming/pipeline/server.rs | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/src/streaming/pipeline/server.rs b/src/streaming/pipeline/server.rs
index 5a18f267..91548168 100644
--- a/src/streaming/pipeline/server.rs
+++ b/src/streaming/pipeline/server.rs
@@ -131,8 +131,9 @@ impl<P, T, B, S> super::advanced::Dispatch for Dispatch<S, T, P> where
     }
 
     fn poll(&mut self) -> Poll<Option<PipelineMessage<Self::In, Self::Stream, Self::Error>>, io::Error> {
-        for slot in self.in_flight.iter_mut() {
-            slot.poll();
+        match self.in_flight.front_mut() {
+            Some(slot) => slot.poll(),
+            _ => return Ok(Async::NotReady)
         }
 
         match self.in_flight.front() {