Configurable thread priority for collider worker threads.

Author: Sergey Zubarev

src/main/java/org/jsl/collider/Collider.java

@@ -44,6 +44,7 @@ public abstract class Collider
 {
     public static class Config
     {
+        public int threadPriority;
         public int threadPoolThreads;
         public boolean useDirectBuffers;
         public ByteOrder byteOrder;
@@ -58,6 +59,7 @@ public static class Config
 
         public Config()
         {
+            threadPriority = Thread.NORM_PRIORITY;
             threadPoolThreads = 0; /* by default = number of cores */
             useDirectBuffers  = true;
             byteOrder = ByteOrder.nativeOrder();
@@ -66,7 +68,7 @@ public Config()
             socketRecvBufSize = 0; /* Use system default settings by default */
 
             forwardReadMaxSize     = (256 * 1024);
-            inputQueueBlockSize    = (32 * 1024);
+            inputQueueBlockSize    = (64 * 1024);
             inputQueueCacheMaxSize = 128;
             joinMessageMaxSize     = 0;
             datagramReadMinSize    = (2 * 1024);

src/main/java/org/jsl/collider/ColliderImpl.java

@@ -371,7 +371,7 @@ public ColliderImpl( Config config ) throws IOException
             threadPoolThreads = Runtime.getRuntime().availableProcessors();
         if (threadPoolThreads < 4)
             threadPoolThreads = 4;
-        m_threadPool = new ThreadPool( "CTP", threadPoolThreads );
+        m_threadPool = new ThreadPool("CTP", threadPoolThreads, config.threadPriority);
 
         if (config.inputQueueCacheMaxSize == 0)
             config.inputQueueCacheMaxSize = (threadPoolThreads * 3);
@@ -390,8 +390,11 @@ public ColliderImpl( Config config ) throws IOException
     public void run()
     {
         if (s_logger.isLoggable(Level.FINE))
-            s_logger.fine( "start" );
+            s_logger.fine("start");
 
+        final Thread currentThread = Thread.currentThread();
+        final int threadPriority = currentThread.getPriority();
+        currentThread.setPriority(getConfig().threadPriority);
         m_threadPool.start();
 
         final DummyRunnable dummyRunnable = new DummyRunnable();
@@ -493,14 +496,16 @@ public void run()
         }
 
         for (Map.Entry<Integer, RetainableDataBlockCache> me : m_dataBlockCache.entrySet())
-            me.getValue().clear( s_logger );
+            me.getValue().clear(s_logger);
         m_dataBlockCache.clear();
 
         if (m_joinPool != null)
-            m_joinPool.release( s_logger );
+            m_joinPool.release(s_logger);
 
         if (s_logger.isLoggable(Level.FINE))
-            s_logger.fine( "finish (" + statLoopIt + ", " + statLoopReadersG0 + ")." );
+            s_logger.fine("finish (" + statLoopIt + ", " + statLoopReadersG0 + ").");
+
+        currentThread.setPriority(threadPriority);
     }
 
     public void stop()

src/main/java/org/jsl/collider/ThreadPool.java

@@ -38,7 +38,7 @@ private class Worker extends Thread
     {
         private final int m_id;
 
-        public Worker( int id )
+        Worker(int id)
         {
             m_id = id;
         }
@@ -47,10 +47,10 @@ public void run()
         {
             final String name = m_name + "-" + getId();
             final int workerId = (1 << m_id);
-            setName( name );
+            setName(name);
 
             if (s_logger.isLoggable(Level.FINE))
-                s_logger.log( Level.FINE, name + ": started." );
+                s_logger.log(Level.FINE, name + ": started.");
 
             int parks = 0;
             int idx = 0;
@@ -148,7 +148,7 @@ private static class DummyRunnable extends Runnable
         public void runInThreadPool()
         {
             /* Should never be called */
-            assert( false );
+            assert(false);
         }
     }
 
@@ -172,7 +172,7 @@ public void runInThreadPool()
     private final AtomicReferenceArray<Runnable> m_tra;
     private volatile int m_state;
 
-    public ThreadPool( String name, int threads, int contentionFactor )
+    public ThreadPool(String name, int threads, int threadPriority, int contentionFactor)
     {
         /* Current implementation supports up to 29 worker threads,
          * should be enough for most cases for now.
@@ -181,7 +181,7 @@ public ThreadPool( String name, int threads, int contentionFactor )
         if (threads > 29)
             threads = 29;
 
-        assert( contentionFactor >= 1 );
+        assert(contentionFactor >= 1);
         if (contentionFactor < 1)
             contentionFactor = 1;
 
@@ -190,16 +190,25 @@ public ThreadPool( String name, int threads, int contentionFactor )
 
         m_thread = new Thread[threads];
         for (int idx=0; idx<threads; idx++)
-            m_thread[idx] = new Worker(idx);
+        {
+            final Worker worker = new Worker(idx);
+            worker.setPriority(threadPriority);
+            m_thread[idx] = worker;
+        }
 
         m_hra = new AtomicReferenceArray<Runnable>( contentionFactor * FS_PADDING );
         m_tra = new AtomicReferenceArray<Runnable>( contentionFactor * FS_PADDING );
         m_state = 0;
     }
 
-    public ThreadPool( String name, int threads )
+    public ThreadPool(String name, int threads)
+    {
+        this(name, threads, Thread.NORM_PRIORITY, 4);
+    }
+
+    public ThreadPool(String name, int threads, int threadPriority)
     {
-        this( name, threads, 4 );
+        this(name, threads, threadPriority, 4);
     }
 
     public void start()