Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[client] shard queues to reduce contention #106

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
Open

[client] shard queues to reduce contention #106

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
12395b4
[shards] shard locks+queues.
truthbk Jan 31, 2020
784f099
Cache threadId for quicker lookup, use modulo
truthbk Mar 12, 2020
183d368
fix rebase collateral
truthbk Apr 9, 2020
bcc863f
[test] fixing test to account for sharding
truthbk Oct 29, 2020
d00bc54
[style] address cops
truthbk Oct 30, 2020
f3165a7
[uds] set better defaults for mac packet size
truthbk Nov 2, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions src/main/java/com/timgroup/statsd/BufferPool.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,10 @@ int getSize() {
return size;
}

int getBufferSize() {
return bufferSize;
}

int available() {
return pool.size();
}
Expand Down
23 changes: 14 additions & 9 deletions src/main/java/com/timgroup/statsd/NonBlockingStatsDClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,12 +83,14 @@ String tag() {
}
}

public static final int DEFAULT_MAX_PACKET_SIZE_BYTES = 1400;
public static final int DEFAULT_UDP_MAX_PACKET_SIZE_BYTES = 1432;
public static final int DEFAULT_UDS_MAX_PACKET_SIZE_BYTES = 8192;
public static final int DEFAULT_QUEUE_SIZE = 4096;
public static final int DEFAULT_POOL_SIZE = 512;
public static final int DEFAULT_PROCESSOR_WORKERS = 1;
public static final int DEFAULT_SENDER_WORKERS = 1;
public static final int DEFAULT_DOGSTATSD_PORT = 8125;
public static final int DEFAULT_LOCK_SHARD_GRAIN = 4;
public static final int SOCKET_TIMEOUT_MS = 100;
public static final int SOCKET_BUFFER_BYTES = -1;
public static final boolean DEFAULT_BLOCKING = false;
Expand Down Expand Up @@ -213,6 +215,9 @@ private static String format(ThreadLocal<NumberFormat> formatter, Number value)
* The number of processor worker threads assembling buffers for submission.
* @param senderWorkers
* The number of sender worker threads submitting buffers to the socket.
* @param lockShardGrain
* The granularity for the lock sharding - sharding is based of thread id
* so value should not be greater than the application thread count..
* @param blocking
* Blocking or non-blocking implementation for statsd message queue.
* @param enableTelemetry
Expand All @@ -230,7 +235,7 @@ public NonBlockingStatsDClient(final String prefix, final int queueSize, String[
final StatsDClientErrorHandler errorHandler, Callable<SocketAddress> addressLookup,
Callable<SocketAddress> telemetryAddressLookup, final int timeout, final int bufferSize,
final int maxPacketSizeBytes, String entityID, final int poolSize, final int processorWorkers,
final int senderWorkers, boolean blocking, final boolean enableTelemetry,
final int senderWorkers, final int lockShardGrain, boolean blocking, final boolean enableTelemetry,
final int telemetryFlushInterval, final int aggregationFlushInterval, final int aggregationShards)
throws StatsDClientException {
if ((prefix != null) && (!prefix.isEmpty())) {
Expand Down Expand Up @@ -288,7 +293,7 @@ public NonBlockingStatsDClient(final String prefix, final int queueSize, String[
}

statsDProcessor = createProcessor(queueSize, handler, maxPacketSizeBytes, poolSize,
processorWorkers, blocking, aggregationFlushInterval, aggregationShards);
processorWorkers, lockShardGrain, blocking, aggregationFlushInterval, aggregationShards);
telemetryStatsDProcessor = statsDProcessor;

Properties properties = new Properties();
Expand Down Expand Up @@ -320,7 +325,7 @@ public NonBlockingStatsDClient(final String prefix, final int queueSize, String[

// similar settings, but a single worker and non-blocking.
telemetryStatsDProcessor = createProcessor(queueSize, handler, maxPacketSizeBytes,
poolSize, 1, false, 0, aggregationShards);
poolSize, 1, 1, false, 0, aggregationShards);
}

this.telemetry = new Telemetry(telemetrytags, telemetryStatsDProcessor);
Expand Down Expand Up @@ -997,18 +1002,18 @@ public NonBlockingStatsDClient(final String prefix, final int queueSize, String[

this(prefix, queueSize, constantTags, errorHandler, addressLookup, addressLookup, timeout,
bufferSize, maxPacketSizeBytes, entityID, poolSize, processorWorkers, senderWorkers,
blocking, enableTelemetry, telemetryFlushInterval, 0, 0);
DEFAULT_LOCK_SHARD_GRAIN, blocking, enableTelemetry, telemetryFlushInterval, 0, 0);
}

protected StatsDProcessor createProcessor(final int queueSize, final StatsDClientErrorHandler handler,
final int maxPacketSizeBytes, final int bufferPoolSize, final int workers, final boolean blocking,
final int aggregationFlushInterval, final int aggregationShards) throws Exception {
final int maxPacketSizeBytes, final int bufferPoolSize, final int workers, final int lockShardGrain,
final boolean blocking, final int aggregationFlushInterval, final int aggregationShards) throws Exception {
if (blocking) {
return new StatsDBlockingProcessor(queueSize, handler, maxPacketSizeBytes, bufferPoolSize,
workers, aggregationFlushInterval, aggregationShards);
workers, lockShardGrain, aggregationFlushInterval, aggregationShards);
} else {
return new StatsDNonBlockingProcessor(queueSize, handler, maxPacketSizeBytes, bufferPoolSize,
workers, aggregationFlushInterval, aggregationShards);
workers, lockShardGrain, aggregationFlushInterval, aggregationShards);
}
}

Expand Down
22 changes: 18 additions & 4 deletions src/main/java/com/timgroup/statsd/NonBlockingStatsDClientBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,16 +16,17 @@ public class NonBlockingStatsDClientBuilder {
* See https://github.com/DataDog/java-dogstatsd-client/pull/17 for discussion.
*/

public int maxPacketSizeBytes = 0;
public int port = NonBlockingStatsDClient.DEFAULT_DOGSTATSD_PORT;
public int telemetryPort = NonBlockingStatsDClient.DEFAULT_DOGSTATSD_PORT;
public int queueSize = NonBlockingStatsDClient.DEFAULT_QUEUE_SIZE;
public int timeout = NonBlockingStatsDClient.SOCKET_TIMEOUT_MS;
public int bufferPoolSize = NonBlockingStatsDClient.DEFAULT_POOL_SIZE;
public int socketBufferSize = NonBlockingStatsDClient.SOCKET_BUFFER_BYTES;
public int maxPacketSizeBytes = NonBlockingStatsDClient.DEFAULT_MAX_PACKET_SIZE_BYTES;
public int processorWorkers = NonBlockingStatsDClient.DEFAULT_PROCESSOR_WORKERS;
public int senderWorkers = NonBlockingStatsDClient.DEFAULT_SENDER_WORKERS;
public boolean blocking = NonBlockingStatsDClient.DEFAULT_BLOCKING;
public int lockShardGrain = NonBlockingStatsDClient.DEFAULT_LOCK_SHARD_GRAIN;
public boolean enableTelemetry = NonBlockingStatsDClient.DEFAULT_ENABLE_TELEMETRY;
public boolean enableAggregation = NonBlockingStatsDClient.DEFAULT_ENABLE_AGGREGATION;
public int telemetryFlushInterval = Telemetry.DEFAULT_FLUSH_INTERVAL;
Expand Down Expand Up @@ -90,6 +91,11 @@ public NonBlockingStatsDClientBuilder senderWorkers(int val) {
return this;
}

public NonBlockingStatsDClientBuilder lockShardGrain(int val) {
lockShardGrain = val;
return this;
}

public NonBlockingStatsDClientBuilder blocking(boolean val) {
blocking = val;
return this;
Expand Down Expand Up @@ -165,13 +171,21 @@ public NonBlockingStatsDClientBuilder aggregationShards(int val) {
* @return the built NonBlockingStatsDClient.
*/
public NonBlockingStatsDClient build() throws StatsDClientException {

int packetSize = maxPacketSizeBytes;
Callable<SocketAddress> lookup = addressLookup;
Callable<SocketAddress> telemetryLookup = telemetryAddressLookup;

if (lookup == null) {
lookup = staticStatsDAddressResolution(hostname, port);
}

if (packetSize == 0) {
packetSize = (port == 0) ? NonBlockingStatsDClient.DEFAULT_UDS_MAX_PACKET_SIZE_BYTES :
NonBlockingStatsDClient.DEFAULT_UDP_MAX_PACKET_SIZE_BYTES;
}


if (telemetryLookup == null) {
if (telemetryHostname == null) {
telemetryLookup = lookup;
Expand All @@ -181,9 +195,9 @@ public NonBlockingStatsDClient build() throws StatsDClientException {
}

return new NonBlockingStatsDClient(prefix, queueSize, constantTags, errorHandler,
lookup, telemetryLookup, timeout, socketBufferSize, maxPacketSizeBytes,
entityID, bufferPoolSize, processorWorkers, senderWorkers, blocking,
enableTelemetry, telemetryFlushInterval,
lookup, telemetryLookup, timeout, socketBufferSize, packetSize,
entityID, bufferPoolSize, processorWorkers, senderWorkers, lockShardGrain,
blocking, enableTelemetry, telemetryFlushInterval,
(enableAggregation ? aggregationFlushInterval : 0), aggregationShards);
}

Expand Down
54 changes: 42 additions & 12 deletions src/main/java/com/timgroup/statsd/StatsDBlockingProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,10 +11,15 @@

public class StatsDBlockingProcessor extends StatsDProcessor {

private final BlockingQueue<Message> messages;
private final BlockingQueue<Message>[] messages;
private final BlockingQueue<Integer>[] processorWorkQueue;

private class ProcessingTask extends StatsDProcessor.ProcessingTask {

public ProcessingTask(int id) {
super(id);
}

@Override
public void run() {
ByteBuffer sendBuffer;
Expand All @@ -30,15 +35,17 @@ public void run() {

aggregator.start();

while (!((emptyHighPrio = highPrioMessages.isEmpty()) && (empty = messages.isEmpty()) && shutdown)) {
while (!((emptyHighPrio = highPrioMessages.isEmpty())
&& processorWorkQueue[this.processorQueueId].isEmpty() && shutdown)) {

try {

final Message message;
if (!emptyHighPrio) {
message = highPrioMessages.poll();
} else {
message = messages.poll(WAIT_SLEEP_MS, TimeUnit.MILLISECONDS);
final int messageQueueIdx = processorWorkQueue[this.processorQueueId].poll();
message = messages[messageQueueIdx].poll(WAIT_SLEEP_MS, TimeUnit.MILLISECONDS);
}

if (message != null) {
Expand Down Expand Up @@ -74,7 +81,7 @@ public void run() {
writeBuilderToSendBuffer(sendBuffer);
}

if (null == messages.peek()) {
if (null == processorWorkQueue[this.processorQueueId].peek()) {
outboundQueue.put(sendBuffer);
sendBuffer = bufferPool.borrow();
}
Expand All @@ -94,34 +101,57 @@ public void run() {
aggregator.stop();
endSignal.countDown();
}

}

StatsDBlockingProcessor(final int queueSize, final StatsDClientErrorHandler handler,
final int maxPacketSizeBytes, final int poolSize, final int workers,
final int maxPacketSizeBytes, final int poolSize, final int workers, final int lockShardGrain,
final int aggregatorFlushInterval, final int aggregatorShards) throws Exception {

super(queueSize, handler, maxPacketSizeBytes, poolSize, workers, aggregatorFlushInterval, aggregatorShards);
this.messages = new ArrayBlockingQueue<>(queueSize);
super(queueSize, handler, maxPacketSizeBytes, poolSize, workers,
lockShardGrain, aggregatorFlushInterval, aggregatorShards);

this.messages = new ArrayBlockingQueue[lockShardGrain];
for (int i = 0 ; i < lockShardGrain ; i++) {
this.messages[i] = new ArrayBlockingQueue<Message>(queueSize);
}

this.processorWorkQueue = new ArrayBlockingQueue[workers];
for (int i = 0 ; i < workers ; i++) {
this.processorWorkQueue[i] = new ArrayBlockingQueue<Integer>(queueSize);
}
}

@Override
protected ProcessingTask createProcessingTask() {
return new ProcessingTask();
protected ProcessingTask createProcessingTask(int id) {
return new ProcessingTask(id);
}

StatsDBlockingProcessor(final StatsDBlockingProcessor processor)
throws Exception {

super(processor);
this.messages = new ArrayBlockingQueue<>(processor.getQcapacity());

this.messages = new ArrayBlockingQueue[lockShardGrain];
for (int i = 0 ; i < lockShardGrain ; i++) {
this.messages[i] = new ArrayBlockingQueue<Message>(getQcapacity());
}

this.processorWorkQueue = new ArrayBlockingQueue[workers];
for (int i = 0 ; i < workers ; i++) {
this.processorWorkQueue[i] = new ArrayBlockingQueue<Integer>(getQcapacity());
}
}

@Override
protected boolean send(final Message message) {
try {
int threadId = getThreadId();
int shard = threadId % lockShardGrain;
int processQueue = threadId % workers;

if (!shutdown) {
messages.put(message);
messages[shard].put(message);
processorWorkQueue[processQueue].put(shard);
return true;
}
} catch (InterruptedException e) {
Expand Down
73 changes: 55 additions & 18 deletions src/main/java/com/timgroup/statsd/StatsDNonBlockingProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,16 +11,22 @@

public class StatsDNonBlockingProcessor extends StatsDProcessor {

private final Queue<Message> messages;
private final AtomicInteger qsize; // qSize will not reflect actual size, but a close estimate.
private final Queue<Message>[] messages;
private final Queue<Integer>[] processorWorkQueue;
private final AtomicInteger[] qsize; // qSize will not reflect actual size, but a close estimate.

private class ProcessingTask extends StatsDProcessor.ProcessingTask {

public ProcessingTask(int id) {
super(id);
}

@Override
public void run() {
ByteBuffer sendBuffer;
boolean empty = true;
boolean emptyHighPrio = true;
int messageQueueIdx = 0;

try {
sendBuffer = bufferPool.borrow();
Expand All @@ -31,7 +37,8 @@ public void run() {

aggregator.start();

while (!((emptyHighPrio = highPrioMessages.isEmpty()) && (empty = messages.isEmpty()) && shutdown)) {
while (!((emptyHighPrio = highPrioMessages.isEmpty())
&& (empty = processorWorkQueue[this.processorQueueId].isEmpty()) && shutdown)) {

try {

Expand All @@ -48,12 +55,14 @@ public void run() {
if (!emptyHighPrio) {
message = highPrioMessages.poll();
} else {
message = messages.poll();

messageQueueIdx = processorWorkQueue[this.processorQueueId].poll();
message = messages[messageQueueIdx].poll();
}

if (message != null) {

qsize.decrementAndGet();
qsize[messageQueueIdx].decrementAndGet();

// TODO: Aggregate and fix, there's some duplicate logic
if (aggregator.aggregateMessage(message)) {
Expand Down Expand Up @@ -88,7 +97,7 @@ public void run() {
writeBuilderToSendBuffer(sendBuffer);
}

if (null == messages.peek()) {
if (null == processorWorkQueue[this.processorQueueId].peek()) {
outboundQueue.put(sendBuffer);
sendBuffer = bufferPool.borrow();
}
Expand All @@ -112,31 +121,59 @@ public void run() {

StatsDNonBlockingProcessor(final int queueSize, final StatsDClientErrorHandler handler,
final int maxPacketSizeBytes, final int poolSize, final int workers,
final int aggregatorFlushInterval, final int aggregatorShards)
throws Exception {
final int lockShardGrain, final int aggregatorFlushInterval,
final int aggregatorShards) throws Exception {

super(queueSize, handler, maxPacketSizeBytes, poolSize, workers,
lockShardGrain, aggregatorFlushInterval, aggregatorShards);

this.qsize = new AtomicInteger[lockShardGrain];
this.messages = new ConcurrentLinkedQueue[lockShardGrain];
for (int i = 0 ; i < lockShardGrain ; i++) {
this.qsize[i] = new AtomicInteger();
this.messages[i] = new ConcurrentLinkedQueue<Message>();
this.qsize[i].set(0);
}

super(queueSize, handler, maxPacketSizeBytes, poolSize, workers, aggregatorFlushInterval, aggregatorShards);
this.qsize = new AtomicInteger(0);
this.messages = new ConcurrentLinkedQueue<>();
this.processorWorkQueue = new ConcurrentLinkedQueue[workers];
for (int i = 0 ; i < workers ; i++) {
this.processorWorkQueue[i] = new ConcurrentLinkedQueue<Integer>();
}
}

@Override
protected ProcessingTask createProcessingTask() {
return new ProcessingTask();
protected ProcessingTask createProcessingTask(int id) {
return new ProcessingTask(id);
}

StatsDNonBlockingProcessor(final StatsDNonBlockingProcessor processor) throws Exception {
super(processor);
this.qsize = new AtomicInteger(0);
this.messages = new ConcurrentLinkedQueue<>();

this.qsize = new AtomicInteger[lockShardGrain];
this.messages = new ConcurrentLinkedQueue[lockShardGrain];
for (int i = 0 ; i < lockShardGrain ; i++) {
this.qsize[i] = new AtomicInteger();
this.messages[i] = new ConcurrentLinkedQueue<Message>();
this.qsize[i].set(0);
}

this.processorWorkQueue = new ConcurrentLinkedQueue[workers];
for (int i = 0 ; i < workers ; i++) {
this.processorWorkQueue[i] = new ConcurrentLinkedQueue<Integer>();
}
}

@Override
protected boolean send(final Message message) {
if (!shutdown) {
if (qsize.get() < qcapacity) {
messages.offer(message);
qsize.incrementAndGet();
int threadId = getThreadId();
int shard = threadId % lockShardGrain;
int processQueue = threadId % workers;

if (qsize[shard].get() < qcapacity) {
messages[shard].offer(message);
qsize[shard].incrementAndGet();
processorWorkQueue[processQueue].offer(shard);
Comment on lines +175 to +176
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Here if I'm correct we are dispatching message to a sharded queue, and those shards are unrelated to processQueue. Did we try 1:1 (or 1:N) mapping between process queues and message queues ? IMHO (unsure) it should (a little bit) ease the concurrency within the JVM.

return true;
}
}
Expand Down
Loading