Skip to content

Commit f0f6bf8

Browse files
truthbktruthbk
committed
fix rebase collateral
1 parent 805c073 commit f0f6bf8

File tree

5 files changed

+35
-88
lines changed

5 files changed

+35
-88
lines changed

‎src/main/java/com/timgroup/statsd/NonBlockingStatsDClientBuilder.java

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -142,9 +142,9 @@ public NonBlockingStatsDClient build() throws StatsDClientException {
142142
enableTelemetry, telemetryFlushInterval);
143143
} else {
144144
return new NonBlockingStatsDClient(prefix, queueSize, constantTags, errorHandler,
145-
staticStatsDAddressResolution(hostname, port), timeout, socketBufferSize, maxPacketSizeBytes,
146-
entityID, bufferPoolSize, processorWorkers, senderWorkers, blocking, enableTelemetry,
147-
telemetryFlushInterval);
145+
staticStatsDAddressResolution(hostname, port), timeout, socketBufferSize,
146+
maxPacketSizeBytes, entityID, bufferPoolSize, processorWorkers, senderWorkers,
147+
lockShardGrain, blocking, enableTelemetry, telemetryFlushInterval);
148148
}
149149
}
150150

‎src/main/java/com/timgroup/statsd/StatsDBlockingProcessor.java

Lines changed: 7 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,10 @@ public class StatsDBlockingProcessor extends StatsDProcessor {
1616

1717
private class ProcessingTask extends StatsDProcessor.ProcessingTask {
1818

19+
public ProcessingTask(int id) {
20+
super(id);
21+
}
22+
1923
@Override
2024
public void run() {
2125
boolean empty;
@@ -63,7 +67,7 @@ public void run() {
6367
writeBuilderToSendBuffer(sendBuffer);
6468
}
6569

66-
if (null == messages.peek()) {
70+
if (null == processorWorkQueue[this.processorQueueId].peek()) {
6771
outboundQueue.put(sendBuffer);
6872
sendBuffer = bufferPool.borrow();
6973
}
@@ -82,7 +86,6 @@ public void run() {
8286
builder.trimToSize();
8387
endSignal.countDown();
8488
}
85-
8689
}
8790

8891
StatsDBlockingProcessor(final int queueSize, final StatsDClientErrorHandler handler,
@@ -100,13 +103,11 @@ public void run() {
100103
for (int i = 0 ; i < workers ; i++) {
101104
this.processorWorkQueue[i] = new ArrayBlockingQueue<Integer>(queueSize);
102105
}
103-
104-
super(queueSize, handler, maxPacketSizeBytes, poolSize, workers);
105106
}
106107

107108
@Override
108-
protected ProcessingTask createProcessingTask() {
109-
return new ProcessingTask();
109+
protected ProcessingTask createProcessingTask(int id) {
110+
return new ProcessingTask(id);
110111
}
111112

112113
@Override

‎src/main/java/com/timgroup/statsd/StatsDNonBlockingProcessor.java

Lines changed: 7 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -17,13 +17,13 @@ public class StatsDNonBlockingProcessor extends StatsDProcessor {
1717
private final int qcapacity;
1818
private final AtomicInteger[] qsize; // qSize will not reflect actual size, but a close estimate.
1919

20-
private class ProcessingTask implements Runnable {
21-
private final int processorQueueId;
20+
private class ProcessingTask extends StatsDProcessor.ProcessingTask {
2221

2322
public ProcessingTask(int id) {
24-
this.processorQueueId = id;
23+
super(id);
2524
}
2625

26+
@Override
2727
public void run() {
2828
boolean empty;
2929
ByteBuffer sendBuffer;
@@ -48,68 +48,9 @@ public void run() {
4848
}
4949

5050
final int messageQueueIdx = processorWorkQueue[this.processorQueueId].poll();
51-
final String message = messages[messageQueueIdx].poll();
51+
final Message message = messages[messageQueueIdx].poll();
5252
if (message != null) {
5353
qsize[messageQueueIdx].decrementAndGet();
54-
final byte[] data = message.getBytes(MESSAGE_CHARSET);
55-
if (sendBuffer.capacity() < data.length) {
56-
throw new InvalidMessageException(MESSAGE_TOO_LONG, message);
57-
}
58-
if (sendBuffer.remaining() < (data.length + 1)) {
59-
outboundQueue.put(sendBuffer);
60-
sendBuffer = bufferPool.borrow();
61-
}
62-
if (sendBuffer.position() > 0) {
63-
sendBuffer.put((byte) '\n');
64-
}
65-
sendBuffer.put(data);
66-
if (null == processorWorkQueue[this.processorQueueId].peek()) {
67-
outboundQueue.put(sendBuffer);
68-
sendBuffer = bufferPool.borrow();
69-
}
70-
}
71-
} catch (final InterruptedException e) {
72-
if (shutdown) {
73-
endSignal.countDown();
74-
return;
75-
}
76-
} catch (final Exception e) {
77-
handler.handle(e);
78-
}
79-
}
80-
endSignal.countDown();
81-
}
82-
}
83-
84-
private class ProcessingTask extends StatsDProcessor.ProcessingTask {
85-
86-
@Override
87-
public void run() {
88-
boolean empty;
89-
ByteBuffer sendBuffer;
90-
91-
try {
92-
sendBuffer = bufferPool.borrow();
93-
} catch (final InterruptedException e) {
94-
handler.handle(e);
95-
return;
96-
}
97-
98-
while (!((empty = messages.isEmpty()) && shutdown)) {
99-
100-
try {
101-
if (empty) {
102-
Thread.sleep(WAIT_SLEEP_MS);
103-
continue;
104-
}
105-
106-
if (Thread.interrupted()) {
107-
return;
108-
}
109-
final Message message = messages.poll();
110-
if (message != null) {
111-
112-
qsize.decrementAndGet();
11354
builder.setLength(0);
11455

11556
message.writeTo(builder);
@@ -137,7 +78,7 @@ public void run() {
13778
writeBuilderToSendBuffer(sendBuffer);
13879
}
13980

140-
if (null == messages.peek()) {
81+
if (null == processorWorkQueue[this.processorQueueId].peek()) {
14182
outboundQueue.put(sendBuffer);
14283
sendBuffer = bufferPool.borrow();
14384
}
@@ -179,8 +120,8 @@ public void run() {
179120
}
180121

181122
@Override
182-
protected ProcessingTask createProcessingTask() {
183-
return new ProcessingTask();
123+
protected ProcessingTask createProcessingTask(int id) {
124+
return new ProcessingTask(id);
184125
}
185126

186127
@Override

‎src/main/java/com/timgroup/statsd/StatsDProcessor.java

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -94,7 +94,7 @@ protected void writeBuilderToSendBuffer(ByteBuffer sendBuffer) {
9494
this.endSignal = new CountDownLatch(workers);
9595
}
9696

97-
protected abstract ProcessingTask createProcessingTask();
97+
protected abstract ProcessingTask createProcessingTask(int id);
9898

9999
protected abstract boolean send(final Message message);
100100

@@ -115,7 +115,7 @@ public static int getThreadId() {
115115
public void run() {
116116

117117
for (int i = 0 ; i < workers ; i++) {
118-
executor.submit(createProcessingTask());
118+
executor.submit(createProcessingTask(i));
119119
}
120120

121121
boolean done = false;

‎src/test/java/com/timgroup/statsd/TelemetryTest.java

Lines changed: 16 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -31,11 +31,16 @@ public static class FakeProcessor extends StatsDProcessor {
3131
public final List<Message> messages = new ArrayList<>();
3232

3333
FakeProcessor(final StatsDClientErrorHandler handler) throws Exception {
34-
super(0, handler, 0, 1, 1);
34+
super(0, handler, 0, 1, 1, 1);
3535
}
3636

3737

3838
private class FakeProcessingTask extends StatsDProcessor.ProcessingTask {
39+
40+
public FakeProcessingTask(int id) {
41+
super(id);
42+
}
43+
3944
@Override
4045
public void run() {}
4146
}
@@ -50,8 +55,8 @@ public boolean send(final Message msg) {
5055
public void run(){}
5156

5257
@Override
53-
protected ProcessingTask createProcessingTask() {
54-
return new FakeProcessingTask();
58+
protected ProcessingTask createProcessingTask(int id) {
59+
return new FakeProcessingTask(id);
5560
}
5661

5762
public List<Message> getMessages() {
@@ -82,12 +87,12 @@ public StatsDNonBlockingTelemetry(final String prefix, final int queueSize, Stri
8287
final StatsDClientErrorHandler errorHandler, Callable<SocketAddress> addressLookup,
8388
final int timeout, final int bufferSize, final int maxPacketSizeBytes,
8489
String entityID, final int poolSize, final int processorWorkers,
85-
final int senderWorkers, boolean blocking, final boolean enableTelemetry,
86-
final int telemetryFlushInterval)
90+
final int senderWorkers, final int lockShardGrain, boolean blocking,
91+
final boolean enableTelemetry, final int telemetryFlushInterval)
8792
throws StatsDClientException {
8893
super(prefix, queueSize, constantTags, errorHandler, addressLookup, timeout, bufferSize,
89-
maxPacketSizeBytes, entityID, poolSize, processorWorkers, senderWorkers, blocking,
90-
enableTelemetry, telemetryFlushInterval);
94+
maxPacketSizeBytes, entityID, poolSize, processorWorkers, senderWorkers, lockShardGrain,
95+
blocking, enableTelemetry, telemetryFlushInterval);
9196
}
9297
};
9398

@@ -98,13 +103,13 @@ public StatsDNonBlockingTelemetry build() throws StatsDClientException {
98103
if (addressLookup != null) {
99104
return new StatsDNonBlockingTelemetry(prefix, queueSize, constantTags, errorHandler,
100105
addressLookup, timeout, socketBufferSize, maxPacketSizeBytes, entityID,
101-
bufferPoolSize, processorWorkers, senderWorkers, blocking, enableTelemetry,
102-
telemetryFlushInterval);
106+
bufferPoolSize, processorWorkers, senderWorkers, lockShardGrain, blocking,
107+
enableTelemetry, telemetryFlushInterval);
103108
} else {
104109
return new StatsDNonBlockingTelemetry(prefix, queueSize, constantTags, errorHandler,
105110
staticStatsDAddressResolution(hostname, port), timeout, socketBufferSize, maxPacketSizeBytes,
106-
entityID, bufferPoolSize, processorWorkers, senderWorkers, blocking, enableTelemetry,
107-
telemetryFlushInterval);
111+
entityID, bufferPoolSize, processorWorkers, senderWorkers, lockShardGrain, blocking,
112+
enableTelemetry, telemetryFlushInterval);
108113
}
109114
}
110115
}

0 commit comments

Comments
 (0)