This is related to QPID-1479.

For starters I have changed the IoSender.java IoReceiver.java and AMQSession.java#Dispatcher to use the Thread factory to create the threads they require.
The ThreadFactory has two implimentations, the default being the java.lang.Threads.
The other is the RealtimeThreadFactory which uses reflection to create threads with a specific priority.
-Dqpid.thread_factory=<thread_factory_class> will decide which thread factory should be loaded.
-Dqpid.rt_thread_priority=<int> specifies the gloabl real time thread priority and defaults to 20.
You could also set individual thread priorities by adding the nessacery config+code changes.

I have also changed the Testkit and QpidBench to use the Thread factory so you could use them for testing/benchmarking work on RT JVMs.



git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@719628 13f79535-47bb-0310-9956-ffa450edef68

8 files changed, 153 insertions, 26 deletions

diff --git a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/perf/LatencyTest.java b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/perf/LatencyTest.java
index 35a2374fbc..4a4f3d124b 100644
--- a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/perf/LatencyTest.java
+++ b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/perf/LatencyTest.java
@@ -37,6 +37,7 @@ import javax.jms.MessageProducer;
 import javax.jms.TextMessage;
 
 import org.apache.qpid.testkit.MessageFactory;
+import org.apache.qpid.thread.Threading;
 
 /**
  * Latency test sends an x number of messages in warmup mode and wait for a confirmation
@@ -314,19 +315,36 @@ public class LatencyTest extends PerfBase implements MessageListener
 
     public static void main(String[] args)
     {
-        LatencyTest latencyTest = new LatencyTest();
-        latencyTest.test();
-        latencyTest.printToConsole();
-        if (System.getProperty("file") != null)
+        final LatencyTest latencyTest = new LatencyTest();        
+        Runnable r = new Runnable()
         {
-            try
+            public void run()
             {
-                latencyTest.writeToFile();
-            }
-            catch(Exception e)
-            {
-                e.printStackTrace();
+                latencyTest.test();
+                latencyTest.printToConsole();
+                if (System.getProperty("file") != null)
+                {
+                    try
+                    {
+                        latencyTest.writeToFile();
+                    }
+                    catch(Exception e)
+                    {
+                        e.printStackTrace();
+                    }
+                }
             }
+        };
+        
+        Thread t;
+        try
+        {
+            t = Threading.getThreadFactory().createThread(r);                      
+        }
+        catch(Exception e)
+        {
+            throw new Error("Error creating latency test thread",e);
         }
+        t.start(); 
     }
 }
\ No newline at end of file
diff --git a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfConsumer.java b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfConsumer.java
index cd12c7010d..9781a7e839 100644
--- a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfConsumer.java
+++ b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfConsumer.java
@@ -27,6 +27,8 @@ import javax.jms.MessageListener;
 import javax.jms.MessageProducer;
 import javax.jms.TextMessage;
 
+import org.apache.qpid.thread.Threading;
+
 /**
  * PerfConsumer will receive x no of messages in warmup mode.
  * Once it receives the Start message it will then signal the PerfProducer.
@@ -242,7 +244,24 @@ public class PerfConsumer extends PerfBase implements MessageListener
 
     public static void main(String[] args)
     {
-        PerfConsumer cons = new PerfConsumer();
-        cons.test();
+        final PerfConsumer cons = new PerfConsumer();
+        Runnable r = new Runnable()
+        {
+            public void run()
+            {
+                cons.test();
+            }
+        };
+        
+        Thread t;
+        try
+        {
+            t = Threading.getThreadFactory().createThread(r);                      
+        }
+        catch(Exception e)
+        {
+            throw new Error("Error creating consumer thread",e);
+        }
+        t.start(); 
     }
 }
\ No newline at end of file
diff --git a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfProducer.java b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfProducer.java
index 757b1bfcda..e9421d7f22 100644
--- a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfProducer.java
+++ b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/perf/PerfProducer.java
@@ -27,6 +27,7 @@ import javax.jms.MessageConsumer;
 import javax.jms.MessageProducer;
 
 import org.apache.qpid.testkit.MessageFactory;
+import org.apache.qpid.thread.Threading;
 
 /**
  * PerfProducer sends an x no of messages in warmup mode and wait for a confirmation
@@ -201,7 +202,24 @@ public class PerfProducer extends PerfBase
 
     public static void main(String[] args)
     {
-        PerfProducer prod = new PerfProducer();
-        prod.test();
+        final PerfProducer prod = new PerfProducer();
+        Runnable r = new Runnable()
+        {
+            public void run()
+            {
+                prod.test();
+            }
+        };
+        
+        Thread t;
+        try
+        {
+            t = Threading.getThreadFactory().createThread(r);                      
+        }
+        catch(Exception e)
+        {
+            throw new Error("Error creating producer thread",e);
+        }
+        t.start();            
     }
 }
\ No newline at end of file
diff --git a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedConsumer.java b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedConsumer.java
index a91d9e7e85..d5514873e6 100644
--- a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedConsumer.java
+++ b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedConsumer.java
@@ -29,6 +29,8 @@ import javax.jms.MessageProducer;
 import javax.jms.Session;
 import javax.jms.TextMessage;
 
+import org.apache.qpid.thread.Threading;
+
 /**
  * Test Description
  * ================
@@ -67,7 +69,7 @@ public class MultiThreadedConsumer extends BaseTest
             {
 
                 final Session session = con.createSession(false, Session.AUTO_ACKNOWLEDGE);
-                Thread t = new Thread(new Runnable()
+                Runnable r = new Runnable()
                 {
                     public void run()
                     {
@@ -131,7 +133,18 @@ public class MultiThreadedConsumer extends BaseTest
 
                     }
 
-                });
+                };
+                
+                Thread t;
+                try
+                {
+                    t = Threading.getThreadFactory().createThread(r);                      
+                }
+                catch(Exception e)
+                {
+                    throw new Error("Error creating consumer thread",e);
+                }
+                
                 t.setName("session-" + i);
                 t.start();
             } // for loop
diff --git a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedProducer.java b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedProducer.java
index 279e5ea0bf..1cf4ee28ca 100644
--- a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedProducer.java
+++ b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/MultiThreadedProducer.java
@@ -32,6 +32,7 @@ import javax.jms.TextMessage;
 
 import org.apache.qpid.client.AMQQueue;
 import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.thread.Threading;
 
 /**
  * Test Description
@@ -79,7 +80,7 @@ public class MultiThreadedProducer extends SimpleProducer
             for (int i = 0; i < session_count; i++)
             {
                 final Session session = con.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
-                Thread t = new Thread(new Runnable()
+                Runnable r = new Runnable()
                 {
                     private Random gen = new Random();
 
@@ -142,7 +143,16 @@ public class MultiThreadedProducer extends SimpleProducer
 
                     }
 
-                });
+                };
+                Thread t;
+                try
+                {
+                    t = Threading.getThreadFactory().createThread(r);                      
+                }
+                catch(Exception e)
+                {
+                    throw new Error("Error creating producer thread",e);
+                }
                 t.setName("session-" + i);
                 t.start();
 
diff --git a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/ResourceLeakTest.java b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/ResourceLeakTest.java
index c33f9ffbf2..1ae2c35970 100644
--- a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/ResourceLeakTest.java
+++ b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/ResourceLeakTest.java
@@ -30,6 +30,7 @@ import javax.jms.Session;
 import org.apache.qpid.client.AMQConnection;
 import org.apache.qpid.client.AMQQueue;
 import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.thread.Threading;
 
 /**
  * Test Description
@@ -131,8 +132,23 @@ public class ResourceLeakTest extends BaseTest
 
     public static void main(String[] args)
     {
-        ResourceLeakTest test = new ResourceLeakTest();
-        test.test();
+        final ResourceLeakTest test = new ResourceLeakTest();
+        Runnable r = new Runnable(){    
+            public void run()
+            {
+                test.test();
+            }
+        };    
+        
+        Thread t;
+        try
+        {
+            t = Threading.getThreadFactory().createThread(r);                      
+        }
+        catch(Exception e)
+        {
+            throw new Error("Error creating test thread",e);
+        }
     }
 
 }
diff --git a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleConsumer.java b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleConsumer.java
index b3eb97dafe..cd6d9013f8 100644
--- a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleConsumer.java
+++ b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleConsumer.java
@@ -29,6 +29,8 @@ import javax.jms.MessageProducer;
 import javax.jms.Session;
 import javax.jms.TextMessage;
 
+import org.apache.qpid.thread.Threading;
+
 /**
  * Test Description
  * ================
@@ -126,9 +128,24 @@ public class SimpleConsumer extends BaseTest
 
     public static void main(String[] args)
     {
-        SimpleConsumer test = new SimpleConsumer();
-        test.setUp();
-        test.test();
+        final SimpleConsumer test = new SimpleConsumer();
+        Runnable r = new Runnable(){    
+            public void run()
+            {
+                test.setUp();
+                test.test();
+            }
+        };    
+        
+        Thread t;
+        try
+        {
+            t = Threading.getThreadFactory().createThread(r);                      
+        }
+        catch(Exception e)
+        {
+            throw new Error("Error creating consumer thread",e);
+        }
     }
 
 }
diff --git a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleProducer.java b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleProducer.java
index 1080092536..805ce7ac29 100644
--- a/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleProducer.java
+++ b/qpid/java/testkit/src/main/java/org/apache/qpid/testkit/soak/SimpleProducer.java
@@ -33,6 +33,7 @@ import javax.jms.TextMessage;
 
 import org.apache.qpid.client.AMQQueue;
 import org.apache.qpid.framing.AMQShortString;
+import org.apache.qpid.thread.Threading;
 
 /**
  * Test Description
@@ -138,9 +139,24 @@ public class SimpleProducer extends BaseTest
 
     public static void main(String[] args)
     {
-        SimpleProducer test = new SimpleProducer();
-        test.setUp();
-        test.test();
+        final SimpleProducer test = new SimpleProducer();
+        Runnable r = new Runnable(){    
+            public void run()
+            {
+                test.setUp();
+                test.test();
+            }
+        };    
+        
+        Thread t;
+        try
+        {
+            t = Threading.getThreadFactory().createThread(r);                      
+        }
+        catch(Exception e)
+        {
+            throw new Error("Error creating producer thread",e);
+        }
     }
 
 }