Cache threadId for quicker lookup, use modulo

Author: truthbk

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

@@ -129,11 +129,9 @@ protected ProcessingTask createProcessingTask() {
     @Override
     protected boolean send(final Message message) {
         try {
-            long threadId = Thread.currentThread().getId();
-            // modulo reduction alternative to: long shard = threadID % this.lockShardGrain;
-            // ref: https://lemire.me/blog/2016/06/27/a-fast-alternative-to-the-modulo-reduction/
-            int shard = (int)((threadId * (long)this.lockShardGrain) >> 32);
-            int processQueue = (int)((threadId * (long)this.workers) >> 32);
+            int threadId = getThreadId();
+            int shard = threadId % lockShardGrain;
+            int processQueue = threadId % workers;
 
             if (!shutdown) {
                 messages[shard].put(message);

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

@@ -157,11 +157,9 @@ protected ProcessingTask createProcessingTask() {
     @Override
     protected boolean send(final Message message) {
         if (!shutdown) {
-            long threadId = Thread.currentThread().getId();
-            // modulo reduction alternative to: long shard = threadID % [shard]this.lockShardGrain;
-            // ref: https://lemire.me/blog/2016/06/27/a-fast-alternative-to-the-modulo-reduction/
-            int shard = (int)((threadId * (long)this.lockShardGrain) >> 32);
-            int processQueue = (int)((threadId * (long)this.workers) >> 32);
+            int threadId = getThreadId();
+            int shard = threadId % lockShardGrain;
+            int processQueue = threadId % workers;
 
             if (qsize[shard].get() < qcapacity) {
                 messages[shard].offer(message);

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

@@ -26,13 +26,22 @@ public abstract class StatsDProcessor implements Runnable {
     protected static final String MESSAGE_TOO_LONG = "Message longer than size of sendBuffer";
     protected static final int WAIT_SLEEP_MS = 10;  // 10 ms would be a 100HZ slice
 
+    // Atomic integer containing the next thread ID to be assigned
+    private static final AtomicInteger nextId = new AtomicInteger(0);
+
     protected final StatsDClientErrorHandler handler;
 
     protected final BufferPool bufferPool;
     protected final Queue<Message> highPrioMessages; // FIFO queue for high priority messages
     protected final BlockingQueue<ByteBuffer> outboundQueue; // FIFO queue with outbound buffers
     protected final ExecutorService executor;
     protected final CountDownLatch endSignal;
+    protected static final ThreadLocal<Integer> threadId = new ThreadLocal<Integer>() {
+        @Override
+        protected Integer initialValue() {
+            return nextId.getAndIncrement();
+        }
+    };
 
     protected final int workers;
     protected final int qcapacity;
@@ -154,6 +163,11 @@ public int getQcapacity() {
         return this.qcapacity;
     }
 
+    // Returns the current thread's unique ID, assigning it if necessary
+    public static int getThreadId() {
+        return threadId.get().intValue();
+    }
+
     @Override
     public void run() {