Cache threadId for quicker lookup, use modulo

Author: truthbk

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

@@ -92,11 +92,9 @@ public void run() {
     @Override
     boolean send(final String 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

@@ -102,11 +102,9 @@ public void run() {
     @Override
     boolean send(final String 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

@@ -17,12 +17,21 @@ 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 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 lockShardGrain;
@@ -54,6 +63,11 @@ public BlockingQueue<ByteBuffer> getOutboundQueue() {
         return this.outboundQueue;
     }
 
+    // Returns the current thread's unique ID, assigning it if necessary
+    public static int getThreadId() {
+        return threadId.get().intValue();
+    }
+
     @Override
     public abstract void run();