[sender] enable buffer pool + concurrent sending threads #95
Conversation
| import java.nio.channels.DatagramChannel; | ||
| import java.nio.charset.Charset; | ||
| import java.util.concurrent.Callable; | ||
|
|
There was a problem hiding this comment.
Note:
import java.io.IOException;
import java.net.SocketAddress;
import java.nio.channels.DatagramChannel;
import java.util.concurrent.Callable;
not used
| @@ -0,0 +1,38 @@ | |||
| package com.timgroup.statsd; | |||
|
|
|||
| import java.io.IOException; | |||
| pool.put(buffer); | ||
| } | ||
|
|
||
| int available() { |
There was a problem hiding this comment.
I still like it as a class method, so I think I'm going to keep it unless you strongly disagree, good catch.
| boolean send(final String message) { | ||
| if (!shutdown) { | ||
| if (qSize.get() < qCapacity) { | ||
| messages.offer(message); |
There was a problem hiding this comment.
Returned value of boolean java.util.Queue.offer(String e) might be checked.
There was a problem hiding this comment.
This has now been moved to StatsDNonBlockingProcessor, there we use a ConcurrentLinkedQueue which never returns false on offer() because it is an unbounded queue.
| } | ||
| } catch (final InterruptedException e) { | ||
| if (shutdown) { | ||
| return; |
There was a problem hiding this comment.
Do we also want to call handler.handle(e); as in line 76?
There was a problem hiding this comment.
It's a good question, I'm not sure, we're going to gracefully shutdown, and handle doesn't do much, but it's a good question. 🤔
There was a problem hiding this comment.
Let me think about it, I'll address in #94 if I decide we should handle these errors differently.
truthbk
force-pushed
the
jaime/buffer_pool
branch
from
ca83be3
to
cc97650
Compare
truthbk
force-pushed
the
jaime/buffer_pool
branch
from
cc97650
to
921a0bc
Compare
|
CI is green, merging to the big larger #94, final changes and fixes will be made there. Thank you for the review. |
This PR adds a buffer pool to the sender. The original code used a single pre-allocated buffer we reused to submit batched dogstatsd metrics. While efficient from a memory perspective, this approach implied we had to stop processing dogstatsd metrics while waiting on IO for every packet submission.
This PR aims to decouple actual packet transmission from dogstatsd message processing and packet assembling. To achieve this we introduce a pool of preallocated buffers and an additional queue, message are processed and batched into these buffers - blocking, if no buffers are available (further improvements are possible here, where we could entertain different strategies to deal with buffer unavailability), once a packet has been filled or no more messages are waiting, the buffer is queued in an outgoing queue. A workers thread (or threads) then will dequeue and write these buffers to the socket. The benefits of this approach, other than the high-level idea explained above, is that it should allow for better usage of the IOwait slice, as the OS kernel will likely do the right thing when scheduling pending work helping us increase the throughput.
TL;DR:
Note: the max perf test is useful for benchmarking but might be worth removing or modifying before merging - or disabling in the CI. It also conflicts currently with a similar test in
jaime/perf, we can address that later.Note: this should be merged onto
jaime/perf(so before #94)