Reaching 200K events/sec

87 points by mattyb 13 years ago | 28 comments

aphyr 13 years ago |

If you're wondering about the workload, this is the trivial Riemann config this benchmark uses:

  (streams
    (rate 5 (comp prn float :metric))

Which means for each five-second interval, sum all the metrics of the events flowing through this stream, divide by the time elapsed, and print that rate to the console.

I'm using this setup to put the heaviest load possible on Riemann's client and TCP server while I optimize those layers--it's not meant to stress the internal stream library, the state index, or pubsub. When I start optimizing those components, I'll have more "real-world" numbers to report.

I should also explain that this particular post explores the high-throughput, high-latency range of the client spectrum. End-to-end TCP latencies (not counting wire time) for single-event messages are on the order of ~100 microseconds-1ms, with occasional spikes to ~30ms depending on JVM GC behavior.

spullara 13 years ago | |

You are likely doing something wrong. My in-memory redis server clone that also uses netty reaches 1.5M requests/s on an arguably heavier workload with a more complicated protocol. I'd love to help you optimize it if it matters.

aphyr 13 years ago | | |

Thanks for the offer! I, uh, certainly didn't mean to claim that Riemann is extraordinarily fast. Up until this point I've been working as hard as I can just to make the darn thing turn on and accomplish something useful. This is quite early in the optimization process, haha. :)

The principle bottleneck in this particular test is actually the client--Riemann itself spends ~94% of its time waiting on epoll in this test. The client is a total hack using Java OIO, calling flush() on every message, Nagle's algorithm disabled for low per-msg latencies... it's a wreck, haha. Part of next week's optimization push is replacing it with a Netty client and tuning TCP options for various workloads.

thebluesky 13 years ago | | |

Is the source available anywhere?

1.5M locally or over a network? What is the size of each message?

trekkin 13 years ago |

>> Throughput here is measured in messages, each containing 100 events, so master is processing 200,000–215,000 events/sec.

So in reality it is ~ 2k messages/sec. This is a rather poor throughput, as even off-the-shelf generic web servers (e.g. nginx) have the throughput an order of magnitude higher, and proprietary systems can reach 500k messages/sec over the network.

aphyr 13 years ago | |

Ah, I didn't intend for this post to reach Hacker News absent context. If you haven't been tracking Riemann, this post might not make sense. ;-)

Riemann is not an HTTP server, or anything analogous. It's an event processor, and reacts to incoming events by running them through an arbitrary set of functions. Events are the logical "requests" against the system, if you're thinking in HTTP terms. Messages are just a bundle of events for synchronous transport, and events can be repackaged in varying bundles of messages depending on latency/throughput requirements. The clients can do this for you.

For instance, the code which generated this benchmark looks like:

  (send client
    {:host "test"
     :service "drop tcp"
     :state "ok"
     :description "a benchmark"
     :metric 1
     :ttl 1
     :tags ["bench"]})

which is a synchronous call, returning when the event is acknowledged by the server. It's making that call 200,000 times a second (in various threads). The clients are doing all sorts of internal buffering and pipelining to make that possible--this particular test uses a batch size of 100 events/msg.

Take a look at http://riemann.io for more. :)

trekkin 13 years ago | | |

Well, I'm sure there are specific use cases where Riemann would be preferable to a generic web server. But for most developers in most situations, it is a no brainer to choose an HTTP-based protocol with off-the-shelf HTTPD server over a 10x slower proprietary system.

revelation 13 years ago |

TLDR: Burned by framework magic. Talk about side effects.

Ten layers (and probably buffers) traveled through until your data hits the wire. Layer x decides to change its IO model and your throughput takes a dive. It's exactly why there was a post recently about building an operating system just to run some network daemon.

aphyr 13 years ago | |

I dunno if I'd call it magic, per se. Netty is a big library and it has good reasons for behaving this way. I just don't think I came through the right documentation path is all.

askimto 13 years ago | |

What post was that?

dschiptsov 13 years ago |

  (defn execution-handler
    "Creates a new netty execution handler."
    []
    (ExecutionHandler.
      (OrderedMemoryAwareThreadPoolExecutor.
        16       ; Core pool size
        1048576  ; 1MB per channel queued
        10485760 ; 10MB total queued
        )))

It is a farce, isn't it?))

aphyr 13 years ago | |

Heh, I originally cargo-culted this line from somewhere in Netty's docs: https://github.com/aphyr/riemann/blob/732a6e8986f75c638c30b5...

After finally digging into how execution handlers worked, the puzzle started to unravel. Netty's docs are pretty good, but you have to understand what all the names mean before you can understand, well, any one part of the system. Bit tough to piece together, at least for my little brain. ;-)

dschiptsov 13 years ago | | |

All I'm trying to say is that switching to the prefix notation adding parenthesis does not make Java a Lisp. ;-)

I'm also not sure that JVM itself is a good idea, especially for serving content, but I do respect people who are trying nevertheless.)

bad_user 13 years ago |

Offtopic, but it's refreshing to read blogs with clean designs and readable texts, without commercials or widgets inviting to click on things and all that crap.

(defn execution-handler "Creates a new netty execution handler." [] (ExecutionHandler. (OrderedMemoryAwareThreadPoolExecutor. 16 ; Core pool size 1048576 ; 1MB per channel queued 10485760 ; 10MB total queued )))

(defn where-partition-clauses "Given expressions like (a (else b) c (else d)), returns [[a c] [b d]]" [exprs] (map vec ((juxt remove (comp (partial mapcat rest) filter)) (fn [expr] (when (list? expr) (= 'else (first expr)))) exprs)))