qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From raj...@apache.org
Subject svn commit: r734205 - in /qpid/trunk/qpid/java: client/src/main/java/org/apache/qpid/client/ common/src/main/java/org/apache/qpid/thread/ common/src/main/java/org/apache/qpid/transport/network/io/ common/src/test/java/org/apache/qpid/thread/
Date Tue, 13 Jan 2009 18:19:17 GMT
Author: rajith
Date: Tue Jan 13 10:19:00 2009
New Revision: 734205

URL: http://svn.apache.org/viewvc?rev=734205&view=rev
Log:
This is related to QPID-1479
This commit contains the core classes for adding the thread abstraction patch

Added:
    qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/
    qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/DefaultThreadFactory.java
    qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/RealtimeThreadFactory.java
    qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/ThreadFactory.java
    qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/Threading.java
    qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/thread/
    qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/thread/ThreadFactoryTest.java
Modified:
    qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
    qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java
    qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java

Modified: qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java?rev=734205&r1=734204&r2=734205&view=diff
==============================================================================
--- qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java (original)
+++ qpid/trunk/qpid/java/client/src/main/java/org/apache/qpid/client/AMQSession.java Tue Jan
13 10:19:00 2009
@@ -67,16 +67,26 @@
 import org.apache.qpid.client.failover.FailoverNoopSupport;
 import org.apache.qpid.client.failover.FailoverProtectedOperation;
 import org.apache.qpid.client.failover.FailoverRetrySupport;
-import org.apache.qpid.client.message.*;
+import org.apache.qpid.client.message.AMQMessageDelegateFactory;
+import org.apache.qpid.client.message.AbstractJMSMessage;
+import org.apache.qpid.client.message.CloseConsumerMessage;
+import org.apache.qpid.client.message.JMSBytesMessage;
+import org.apache.qpid.client.message.JMSMapMessage;
+import org.apache.qpid.client.message.JMSObjectMessage;
+import org.apache.qpid.client.message.JMSStreamMessage;
+import org.apache.qpid.client.message.JMSTextMessage;
+import org.apache.qpid.client.message.MessageFactoryRegistry;
+import org.apache.qpid.client.message.UnprocessedMessage;
 import org.apache.qpid.client.protocol.AMQProtocolHandler;
-import org.apache.qpid.client.util.FlowControllingBlockingQueue;
-import org.apache.qpid.client.state.AMQStateManager;
 import org.apache.qpid.client.state.AMQState;
+import org.apache.qpid.client.state.AMQStateManager;
+import org.apache.qpid.client.util.FlowControllingBlockingQueue;
 import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.FieldTable;
 import org.apache.qpid.framing.FieldTableFactory;
 import org.apache.qpid.framing.MethodRegistry;
 import org.apache.qpid.jms.Session;
+import org.apache.qpid.thread.Threading;
 import org.apache.qpid.url.AMQBindingURL;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -271,6 +281,8 @@
 
     /** Holds the dispatcher thread for this session. */
     protected Dispatcher _dispatcher;
+    
+    protected Thread _dispatcherThread;
 
     /** Holds the message factory factory for this session. */
     protected MessageFactoryRegistry _messageFactoryRegistry;
@@ -668,7 +680,7 @@
             if (_dispatcher != null)
             {
                 // Failover failed and ain't coming back. Knife the dispatcher.
-                _dispatcher.interrupt();
+                _dispatcherThread.interrupt();
             }
         }
 
@@ -1852,7 +1864,7 @@
     void startDispatcherIfNecessary()
     {
         //If we are the dispatcher then we don't need to check we are started
-        if (Thread.currentThread() == _dispatcher)
+        if (Thread.currentThread() == _dispatcherThread)
         {
             return;
         }
@@ -1883,9 +1895,23 @@
         if (_dispatcher == null)
         {
             _dispatcher = new Dispatcher();
-            _dispatcher.setDaemon(true);
+            try
+            {
+                _dispatcherThread = Threading.getThreadFactory().createThread(_dispatcher);
      
+                
+            }
+            catch(Exception e)
+            {
+                throw new Error("Error creating Dispatcher thread",e);
+            }            
+            _dispatcherThread.setName("Dispatcher-Channel-" + _channelId);
+            _dispatcherThread.setDaemon(true);
             _dispatcher.setConnectionStopped(initiallyStopped);
-            _dispatcher.start();
+            _dispatcherThread.start();
+            if (_dispatcherLogger.isInfoEnabled())
+            {
+                _dispatcherLogger.info(_dispatcherThread.getName() + " created");
+            }
         }
         else
         {
@@ -2606,7 +2632,7 @@
     private static final Logger _dispatcherLogger = LoggerFactory.getLogger("org.apache.qpid.client.AMQSession.Dispatcher");
 
     /** Responsible for decoding a message fragment and passing it to the appropriate message
consumer. */
-    class Dispatcher extends Thread
+    class Dispatcher implements Runnable
     {
 
         /** Track the 'stopped' state of the dispatcher, a session starts in the stopped
state. */
@@ -2615,21 +2641,14 @@
         private final Object _lock = new Object();
         private String dispatcherID = "" + System.identityHashCode(this);
 
-
-
         public Dispatcher()
         {
-            super("Dispatcher-Channel-" + _channelId);
-            if (_dispatcherLogger.isInfoEnabled())
-            {
-                _dispatcherLogger.info(getName() + " created");
-            }
         }
 
         public void close()
         {
             _closed.set(true);
-            interrupt();
+            _dispatcherThread.interrupt();
 
             // fixme awaitTermination
 
@@ -2708,7 +2727,7 @@
         {
             if (_dispatcherLogger.isInfoEnabled())
             {
-                _dispatcherLogger.info(getName() + " started");
+                _dispatcherLogger.info(_dispatcherThread.getName() + " started");
             }
 
             UnprocessedMessage message;
@@ -2771,7 +2790,7 @@
 
             if (_dispatcherLogger.isInfoEnabled())
             {
-                _dispatcherLogger.info(getName() + " thread terminating for channel " + _channelId);
+                _dispatcherLogger.info(_dispatcherThread.getName() + " thread terminating
for channel " + _channelId);
             }
         }
 

Added: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/DefaultThreadFactory.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/DefaultThreadFactory.java?rev=734205&view=auto
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/DefaultThreadFactory.java
(added)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/DefaultThreadFactory.java
Tue Jan 13 10:19:00 2009
@@ -0,0 +1,18 @@
+package org.apache.qpid.thread;
+
+public class DefaultThreadFactory implements ThreadFactory
+{
+
+    public Thread createThread(Runnable r)
+    {
+        return new Thread(r);
+    }
+
+    public Thread createThread(Runnable r, int priority)
+    {
+        Thread t = new Thread(r);
+        t.setPriority(priority);
+        return t;
+    }
+
+}

Added: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/RealtimeThreadFactory.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/RealtimeThreadFactory.java?rev=734205&view=auto
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/RealtimeThreadFactory.java
(added)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/RealtimeThreadFactory.java
Tue Jan 13 10:19:00 2009
@@ -0,0 +1,47 @@
+package org.apache.qpid.thread;
+
+import java.lang.reflect.Constructor;
+
+public class RealtimeThreadFactory implements ThreadFactory
+{
+    private Class threadClass;
+    private Constructor threadConstructor;
+    private Constructor priorityParameterConstructor;
+    private int defaultRTThreadPriority = 20;
+    
+    public RealtimeThreadFactory() throws Exception
+    {
+        defaultRTThreadPriority = Integer.getInteger("qpid.rt_thread_priority",20);
+        threadClass = Class.forName("javax.realtime.RealtimeThread");
+    
+        Class schedulingParametersClass = Class.forName("javax.realtime.SchedulingParameters");
+        Class releaseParametersClass = Class.forName("javax.realtime.ReleaseParameters");
+        Class memoryParametersClass = Class.forName("javax.realtime.MemoryParameters");
+        Class memoryAreaClass = Class.forName("javax.realtime.MemoryArea");
+        Class processingGroupParametersClass = Class.forName("javax.realtime.ProcessingGroupParameters");
+     
+        Class[] paramTypes = new Class[]{schedulingParametersClass,
+                                         releaseParametersClass, 
+                                         memoryParametersClass,
+                                         memoryAreaClass,
+                                         processingGroupParametersClass,
+                                         java.lang.Runnable.class};
+        
+        threadConstructor = threadClass.getConstructor(paramTypes);
+        
+        Class priorityParameterClass = Class.forName("javax.realtime.PriorityParameters");
+        priorityParameterConstructor = priorityParameterClass.getConstructor(new Class[]{int.class});
       
+    }
+
+    public Thread createThread(Runnable r) throws Exception
+    {
+        return createThread(r,defaultRTThreadPriority);
+    }
+
+    public Thread createThread(Runnable r, int priority) throws Exception
+    {
+        Object priorityParams = priorityParameterConstructor.newInstance(priority);
+        return (Thread)threadConstructor.newInstance(priorityParams,null,null,null,null,r);
+    }
+
+}

Added: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/ThreadFactory.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/ThreadFactory.java?rev=734205&view=auto
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/ThreadFactory.java (added)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/ThreadFactory.java Tue
Jan 13 10:19:00 2009
@@ -0,0 +1,7 @@
+package org.apache.qpid.thread;
+
+public interface ThreadFactory
+{
+    public Thread createThread(Runnable r) throws Exception;
+    public Thread createThread(Runnable r, int priority) throws Exception;
+}

Added: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/Threading.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/Threading.java?rev=734205&view=auto
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/Threading.java (added)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/thread/Threading.java Tue Jan
13 10:19:00 2009
@@ -0,0 +1,26 @@
+package org.apache.qpid.thread;
+
+public final class Threading
+{
+    private static ThreadFactory threadFactory;
+    
+    static {
+        try
+        {
+            Class threadFactoryClass = 
+                Class.forName(System.getProperty("qpid.thread_factory", 
+                                                 "org.apache.qpid.thread.DefaultThreadFactory"));
+            
+            threadFactory = (ThreadFactory)threadFactoryClass.newInstance();
+        }
+        catch(Exception e)
+        {
+            throw new Error("Error occured while loading thread factory",e);
+        }
+    }
+    
+    public static ThreadFactory getThreadFactory()
+    {
+        return threadFactory;
+    }
+}

Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java?rev=734205&r1=734204&r2=734205&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java
(original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoReceiver.java
Tue Jan 13 10:19:00 2009
@@ -20,6 +20,7 @@
  */
 package org.apache.qpid.transport.network.io;
 
+import org.apache.qpid.thread.Threading;
 import org.apache.qpid.transport.Receiver;
 import org.apache.qpid.transport.TransportException;
 import org.apache.qpid.transport.util.Logger;
@@ -36,7 +37,7 @@
  *
  */
 
-final class IoReceiver extends Thread
+final class IoReceiver implements Runnable
 {
 
     private static final Logger log = Logger.get(IoReceiver.class);
@@ -47,6 +48,7 @@
     private final Socket socket;
     private final long timeout;
     private final AtomicBoolean closed = new AtomicBoolean(false);
+    private final Thread receiverThread;
     private final boolean shutdownBroken =
         ((String) System.getProperties().get("os.name")).matches("(?i).*windows.*");
 
@@ -58,10 +60,18 @@
         this.bufferSize = bufferSize;
         this.socket = transport.getSocket();
         this.timeout = timeout;
-
-        setDaemon(true);
-        setName(String.format("IoReceiver - %s", socket.getRemoteSocketAddress()));
-        start();
+        
+        try
+        {
+            receiverThread = Threading.getThreadFactory().createThread(this);           
          
+        }
+        catch(Exception e)
+        {
+            throw new Error("Error creating IOReceiver thread",e);
+        }
+        receiverThread.setDaemon(true);
+        receiverThread.setName(String.format("IoReceiver - %s", socket.getRemoteSocketAddress()));
+        receiverThread.start();
     }
 
     void close(boolean block)
@@ -78,10 +88,10 @@
                 {
                     socket.shutdownInput();
                 }
-                if (block && Thread.currentThread() != this)
+                if (block && Thread.currentThread() != receiverThread)
                 {
-                    join(timeout);
-                    if (isAlive())
+                    receiverThread.join(timeout);
+                    if (receiverThread.isAlive())
                     {
                         throw new TransportException("join timed out");
                     }
@@ -133,6 +143,9 @@
                   t.getMessage().equalsIgnoreCase("socket closed") &&
                   closed.get()))
             {
+                log.error(t, "===========================================================");
+                log.error(t, "Exception");
+                log.error(t, "===========================================================");
                 receiver.exception(t);
             }
         }

Modified: qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java?rev=734205&r1=734204&r2=734205&view=diff
==============================================================================
--- qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java
(original)
+++ qpid/trunk/qpid/java/common/src/main/java/org/apache/qpid/transport/network/io/IoSender.java
Tue Jan 13 10:19:00 2009
@@ -24,6 +24,7 @@
 import java.nio.ByteBuffer;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.apache.qpid.thread.Threading;
 import org.apache.qpid.transport.Sender;
 import org.apache.qpid.transport.SenderException;
 import org.apache.qpid.transport.TransportException;
@@ -32,7 +33,7 @@
 import static org.apache.qpid.transport.util.Functions.*;
 
 
-public final class IoSender extends Thread implements Sender<ByteBuffer>
+public final class IoSender implements Runnable, Sender<ByteBuffer>
 {
 
     private static final Logger log = Logger.get(IoSender.class);
@@ -54,7 +55,8 @@
     private final Object notFull = new Object();
     private final Object notEmpty = new Object();
     private final AtomicBoolean closed = new AtomicBoolean(false);
-
+    private final Thread senderThread;
+    
     private volatile Throwable exception = null;
 
 
@@ -74,9 +76,18 @@
             throw new TransportException("Error getting output stream for socket", e);
         }
 
-        setDaemon(true);
-        setName(String.format("IoSender - %s", socket.getRemoteSocketAddress()));
-        start();
+        try
+        {
+            senderThread = Threading.getThreadFactory().createThread(this);             
        
+        }
+        catch(Exception e)
+        {
+            throw new Error("Error creating IOSender thread",e);
+        }
+        
+        senderThread.setDaemon(true);
+        senderThread.setName(String.format("IoSender - %s", socket.getRemoteSocketAddress()));
+        senderThread.start();
     }
 
     private static final int pof2(int n)
@@ -188,10 +199,10 @@
 
             try
             {
-                if (Thread.currentThread() != this)
+                if (Thread.currentThread() != senderThread)
                 {
-                    join(timeout);
-                    if (isAlive())
+                    senderThread.join(timeout);
+                    if (senderThread.isAlive())
                     {
                         throw new SenderException("join timed out");
                     }

Added: qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/thread/ThreadFactoryTest.java
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/thread/ThreadFactoryTest.java?rev=734205&view=auto
==============================================================================
--- qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/thread/ThreadFactoryTest.java
(added)
+++ qpid/trunk/qpid/java/common/src/test/java/org/apache/qpid/thread/ThreadFactoryTest.java
Tue Jan 13 10:19:00 2009
@@ -0,0 +1,46 @@
+package org.apache.qpid.thread;
+
+import junit.framework.TestCase;
+
+public class ThreadFactoryTest extends TestCase
+{
+    public void testThreadFactory()
+    {
+        Class threadFactoryClass = null;
+        try
+        {
+            threadFactoryClass = Class.forName(System.getProperty("qpid.thread_factory",
+                    "org.apache.qpid.thread.DefaultThreadFactory"));            
+        }
+        // If the thread factory class was wrong it will flagged way before it gets here.
+        catch(Exception e)
+        {            
+            fail("Invalid thread factory class");
+        }
+        
+        assertEquals(threadFactoryClass, Threading.getThreadFactory().getClass());
+    }
+    
+    public void testThreadCreate()
+    {
+        Runnable r = new Runnable(){
+          
+            public void run(){
+                
+            }            
+        };
+        
+        Thread t = null;
+        try
+        {
+            t = Threading.getThreadFactory().createThread(r,5);
+        }
+        catch(Exception e)
+        {
+            fail("Error creating thread using Qpid thread factory");
+        }
+        
+        assertNotNull(t);
+        assertEquals(5,t.getPriority());
+    }
+}



Mime
View raw message