You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by jg...@apache.org on 2008/09/17 20:33:05 UTC
svn commit: r696392 - in
/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp:
Connection.cs DispatchingThread.cs Session.cs Transport/Tcp/TcpTransport.cs
Author: jgomes
Date: Wed Sep 17 11:33:05 2008
New Revision: 696392
URL: http://svn.apache.org/viewvc?rev=696392&view=rev
Log:
Rolled back most of the changes made to the threading. Kept those that were definite improvements, and made changes to the existing threading based on the contribution. Now the performance is back to what it should be with improved robustness.
Fixes [AMQNET-112]. (See https://issues.apache.org/activemq/browse/AMQNET-112)
Modified:
activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Connection.cs
activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/DispatchingThread.cs
activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Session.cs
activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Tcp/TcpTransport.cs
Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Connection.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Connection.cs?rev=696392&r1=696391&r2=696392&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Connection.cs (original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Connection.cs Wed Sep 17 11:33:05 2008
@@ -36,19 +36,20 @@
private TimeSpan requestTimeout;
private BrokerInfo brokerInfo; // from broker
private WireFormatInfo brokerWireFormatInfo; // from broker
- private readonly IList sessions = new ArrayList();
+ private readonly IList sessions = ArrayList.Synchronized(new ArrayList());
/// <summary>
/// Private object used for synchronization, instead of public "this"
/// </summary>
private readonly object myLock = new object();
private bool asyncSend = false;
private bool asyncClose = true;
- private volatile bool closed;
- private volatile bool triedConnect;
+ private bool connected = false;
+ private bool closed = false;
+ private bool closing = false;
private long sessionCounter = 0;
private long temporaryDestinationCounter = 0;
private long localTransactionCounter;
- private bool started = false;
+ private readonly AtomicBoolean started = new AtomicBoolean(false);
private bool disposed = false;
public Connection(Uri connectionUri, ITransport transport, ConnectionInfo info)
@@ -111,11 +112,10 @@
public void Start()
{
CheckConnected();
- lock(myLock)
+ if(started.CompareAndSet(false, true))
{
- if(!started)
+ lock(sessions.SyncRoot)
{
- started = true;
foreach(Session session in sessions)
{
session.StartAsyncDelivery();
@@ -130,7 +130,7 @@
/// </summary>
public bool IsStarted
{
- get { return started; }
+ get { return started.Value; }
}
/// <summary>
@@ -140,11 +140,10 @@
public void Stop()
{
CheckConnected();
- lock(myLock)
+ if(started.CompareAndSet(true, false))
{
- if(started)
+ lock(sessions.SyncRoot)
{
- started = false;
foreach(Session session in sessions)
{
session.StopAsyncDelivery();
@@ -174,22 +173,20 @@
System.Collections.Specialized.StringDictionary map = URISupport.ParseQuery(this.brokerUri.Query);
URISupport.SetProperties(session, map, "session.");
- lock(myLock)
+ if(IsStarted)
{
- if(IsStarted)
- {
- session.StartAsyncDelivery();
- }
-
- sessions.Add(session);
+ session.StartAsyncDelivery();
}
+
+ sessions.Add(session);
return session;
}
public void RemoveSession(Session session)
{
DisposeOf(session.SessionId);
- lock(myLock)
+
+ if(!this.closing)
{
sessions.Remove(session);
}
@@ -197,41 +194,22 @@
public void Close()
{
- if(closed)
- {
- return;
- }
-
- //
- // Do a first-run close of sessions after brief synchronization
- //
- IList sessionsCopy;
- lock(myLock)
- {
- sessionsCopy = new ArrayList(sessions);
- }
-
- foreach(Session session in sessionsCopy)
- {
- session.Close();
- }
-
lock(myLock)
{
- if(closed)
+ if(this.closed)
{
return;
}
try
{
- //
- // Copy again for safe enumeration. Always assume
- // that closing a session modifies the sessions list.
- //
- foreach(Session session in new ArrayList(sessions))
+ this.closing = true;
+ lock(sessions.SyncRoot)
{
- session.Close();
+ foreach(Session session in sessions)
+ {
+ session.Close();
+ }
}
sessions.Clear();
@@ -245,8 +223,9 @@
}
finally
{
- closed = true;
- transport = null;
+ this.transport = null;
+ this.closed = true;
+ this.closing = false;
}
}
}
@@ -314,7 +293,7 @@
get { return info.ClientId; }
set
{
- if(triedConnect)
+ if(connected)
{
throw new NMSException("You cannot change the ClientId once the Connection is connected");
}
@@ -397,9 +376,7 @@
/// </summary>
public String CreateTemporaryDestinationName()
{
- return info.ConnectionId.Value
- + ":"
- + Interlocked.Increment(ref temporaryDestinationCounter);
+ return info.ConnectionId.Value + ":" + Interlocked.Increment(ref temporaryDestinationCounter);
}
/// <summary>
@@ -420,30 +397,14 @@
throw new ConnectionClosedException();
}
- if(triedConnect)
- {
- return;
- }
-
- lock(myLock)
+ if(!connected)
{
- if(closed)
- {
- throw new ConnectionClosedException();
- }
-
- if(triedConnect)
- {
- return;
- }
-
- // Set this in advance, to short-circuit SyncRequest's call to this method
- triedConnect = true;
-
+ connected = true;
// now lets send the connection and see if we get an ack/nak
if(null == SyncRequest(info))
{
closed = true;
+ connected = false;
throw new ConnectionClosedException();
}
}
@@ -475,12 +436,9 @@
else if(command is ShutdownInfo)
{
//ShutdownInfo info = (ShutdownInfo)command;
- lock(myLock)
+ if(!closing && !closed)
{
- if(!closed)
- {
- OnException(commandTransport, new NMSException("Broker closed this connection."));
- }
+ OnException(commandTransport, new NMSException("Broker closed this connection."));
}
}
else
@@ -493,7 +451,7 @@
{
bool dispatched = false;
- lock(myLock)
+ lock(sessions.SyncRoot)
{
foreach(Session session in sessions)
{
@@ -569,6 +527,5 @@
answer.SessionId = sessionId;
return answer;
}
-
}
}
Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/DispatchingThread.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/DispatchingThread.cs?rev=696392&r1=696391&r2=696392&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/DispatchingThread.cs (original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/DispatchingThread.cs Wed Sep 17 11:33:05 2008
@@ -27,7 +27,7 @@
public delegate void ExceptionHandler(Exception exception);
private readonly AutoResetEvent m_event = new AutoResetEvent(false);
- private readonly ManualResetEvent m_stopEvent = new ManualResetEvent(false);
+ private bool m_bStopFlag = false;
private Thread m_thread = null;
private readonly DispatchFunction m_dispatchFunc;
private event ExceptionHandler m_exceptionListener;
@@ -37,18 +37,7 @@
m_dispatchFunc = dispatchFunc;
}
- public bool IsStarted
- {
- get
- {
- lock(this)
- {
- return (null != m_thread);
- }
- }
- }
-
- // TODO can't use EventWaitHandle on MONO 1.0
+ // TODO can't use EventWaitHandle on MONO 1.0
public AutoResetEvent EventHandle
{
get { return m_event; }
@@ -72,9 +61,10 @@
{
if (m_thread == null)
{
- m_stopEvent.Reset();
+ m_bStopFlag = false;
m_thread = new Thread(new ThreadStart(MyThreadFunc));
m_thread.IsBackground = true;
+ m_event.Set();
Tracer.Info("Starting dispatcher thread for session");
m_thread.Start();
}
@@ -90,12 +80,16 @@
internal void Stop(int timeoutMilliseconds)
{
Tracer.Info("Stopping dispatcher thread for session");
- Thread localThread;
+ Thread localThread = null;
lock (this)
{
localThread = m_thread;
m_thread = null;
- m_stopEvent.Set();
+ if (!m_bStopFlag)
+ {
+ m_bStopFlag = true;
+ m_event.Set();
+ }
}
if(localThread!=null)
{
@@ -111,43 +105,49 @@
private void MyThreadFunc()
{
Tracer.Info("Dispatcher thread started");
-
- //
- // Put m_stopEvent first so it is preferred if both are signaled
- //
- WaitHandle[] signals = new WaitHandle[] {
- m_stopEvent,
- m_event
- };
- const int kStopEventOffset = 0;
-
try
{
while (true) // loop forever (well, at least until we've been asked to stop)
{
- try
+ lock (this)
{
- m_dispatchFunc();
+ if (m_bStopFlag)
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ break;
+
+
}
- catch(ThreadAbortException)
+
+ try
{
- // Throw for handling down below
- throw;
+ m_dispatchFunc();
}
- catch(Exception ex)
+ catch (Exception ex)
{
if(m_exceptionListener != null)
{
m_exceptionListener(ex);
}
}
-
- int sigOffset = WaitHandle.WaitAny(signals);
- if(kStopEventOffset == sigOffset)
- {
- break;
- }
- // otherwise, continue the loop
+ m_event.WaitOne();
}
Tracer.Info("Dispatcher thread stopped");
}
Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Session.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Session.cs?rev=696392&r1=696391&r2=696392&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Session.cs (original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Session.cs Wed Sep 17 11:33:05 2008
@@ -28,21 +28,22 @@
/// </summary>
public class Session : ISession
{
- private long consumerCounter;
- private readonly IDictionary consumers = new Hashtable();
- private readonly IDictionary producers = new Hashtable();
- private readonly DispatchingThread dispatchingThread;
/// <summary>
/// Private object used for synchronization, instead of public "this"
/// </summary>
private readonly object myLock = new object();
+ private long consumerCounter;
+ private readonly IDictionary consumers = Hashtable.Synchronized(new Hashtable());
+ private readonly IDictionary producers = Hashtable.Synchronized(new Hashtable());
+ private readonly DispatchingThread dispatchingThread;
private DispatchingThread.ExceptionHandler dispatchingThread_ExceptionHandler;
private readonly SessionInfo info;
private long producerCounter;
+ internal bool startedAsyncDelivery = false;
private bool disposed = false;
- private volatile bool closed = false;
-
- private const int kMAX_STOP_ASYNC_MS = 30000;
+ private bool closed = false;
+ private bool closing = false;
+ private TimeSpan MAX_THREAD_WAIT = TimeSpan.FromMilliseconds(30000);
public Session(Connection connection, SessionInfo info, AcknowledgementMode acknowledgementMode)
{
@@ -153,60 +154,37 @@
public void Close()
{
- if(closed)
- {
- return;
- }
-
- //
- // Do a first-run close of consumer and producers after
- // brief synchronization
- //
- IList consumersCopy;
- IList producersCopy;
-
- lock(myLock)
- {
- consumersCopy = new ArrayList(consumers.Values);
- producersCopy = new ArrayList(producers.Values);
- }
-
- foreach(MessageConsumer consumer in consumersCopy)
- {
- consumer.Close();
- }
-
- foreach(MessageProducer producer in producersCopy)
- {
- producer.Close();
- }
-
lock(myLock)
{
- if(closed)
+ if(this.closed)
{
return;
}
try
{
+ this.closing = true;
StopAsyncDelivery();
- // Copy again for safe enumeration
- foreach(MessageConsumer consumer in new ArrayList(consumers.Values))
+ Connection.RemoveSession(this);
+ lock(consumers.SyncRoot)
{
- consumer.Close();
+ foreach(MessageConsumer consumer in consumers.Values)
+ {
+ consumer.Close();
+ }
}
consumers.Clear();
- // Copy again for safe enumeration
- foreach(MessageProducer producer in new ArrayList(producers.Values))
+ lock(producers.SyncRoot)
{
- producer.Close();
+ foreach(MessageProducer producer in producers.Values)
+ {
+ producer.Close();
+ }
}
producers.Clear();
- Connection.RemoveSession(this);
}
- catch (Exception ex)
+ catch(Exception ex)
{
Tracer.ErrorFormat("Error during session close: {0}", ex);
}
@@ -214,6 +192,7 @@
{
this.connection = null;
this.closed = true;
+ this.closing = false;
}
}
}
@@ -227,24 +206,19 @@
{
ProducerInfo command = CreateProducerInfo(destination);
ProducerId producerId = command.ProducerId;
- MessageProducer producer = new MessageProducer(this, command);
- lock(myLock)
- {
- producers[producerId] = producer;
- }
+ MessageProducer producer = null;
try
{
+ producer = new MessageProducer(this, command);
+ producers[producerId] = producer;
this.DoSend(command);
}
catch(Exception)
{
- //
- // DoSend failed. No need to call MessageProducer.Close
- //
- lock(myLock)
+ if(producer != null)
{
- producers.Remove(producerId);
+ producer.Close();
}
throw;
@@ -272,31 +246,23 @@
ConsumerId consumerId = command.ConsumerId;
MessageConsumer consumer = null;
- consumer = new MessageConsumer(this, command, this.AcknowledgementMode);
- // let's register the consumer first in case we start dispatching messages immediately
- lock(myLock)
- {
- consumers[consumerId] = consumer;
- }
-
try
{
+ consumer = new MessageConsumer(this, command, this.AcknowledgementMode);
+ // lets register the consumer first in case we start dispatching messages immediately
+ consumers[consumerId] = consumer;
this.DoSend(command);
+ return consumer;
}
catch(Exception)
{
- //
- // DoSend failed. No need to call MessageProducer.Close
- //
- lock(myLock)
+ if(consumer != null)
{
- consumers.Remove(consumerId);
+ consumer.Close();
}
throw;
}
-
- return consumer;
}
public IMessageConsumer CreateDurableConsumer(ITopic destination, string name, string selector, bool noLocal)
@@ -307,25 +273,18 @@
command.NoLocal = noLocal;
MessageConsumer consumer = null;
- consumer = new MessageConsumer(this, command, this.AcknowledgementMode);
- // let's register the consumer first in case we start dispatching messages immediately
- lock(myLock)
- {
- consumers[consumerId] = consumer;
- }
-
try
{
+ consumer = new MessageConsumer(this, command, this.AcknowledgementMode);
+ // lets register the consumer first in case we start dispatching messages immediately
+ consumers[consumerId] = consumer;
this.DoSend(command);
}
catch(Exception)
{
- //
- // DoSend failed; no need to call MessageConsumer.Close
- //
- lock(myLock)
+ if(consumer != null)
{
- consumers.Remove(consumerId);
+ consumer.Close();
}
throw;
@@ -336,14 +295,9 @@
public void DeleteDurableConsumer(string name)
{
- IConnection conn = this.Connection;
RemoveSubscriptionInfo command = new RemoveSubscriptionInfo();
- if(null != conn)
- {
- command.ConnectionId = Connection.ConnectionId;
- command.ClientId = Connection.ClientId;
- }
-
+ command.ConnectionId = Connection.ConnectionId;
+ command.ClientId = Connection.ClientId;
command.SubcriptionName = name;
this.DoSend(command);
}
@@ -441,7 +395,7 @@
this.TransactionContext.Rollback();
// lets ensure all the consumers redeliver any rolled back messages
- lock(myLock)
+ lock(consumers.SyncRoot)
{
foreach(MessageConsumer consumer in consumers.Values)
{
@@ -538,7 +492,7 @@
public void DisposeOf(ConsumerId objectId)
{
Connection.DisposeOf(objectId);
- lock(myLock)
+ if(!this.closing)
{
consumers.Remove(objectId);
}
@@ -547,7 +501,7 @@
public void DisposeOf(ProducerId objectId)
{
Connection.DisposeOf(objectId);
- lock(myLock)
+ if(!this.closing)
{
producers.Remove(objectId);
}
@@ -556,11 +510,7 @@
public bool DispatchMessage(ConsumerId consumerId, Message message)
{
bool dispatched = false;
- MessageConsumer consumer;
- lock(myLock)
- {
- consumer = (MessageConsumer) consumers[consumerId];
- }
+ MessageConsumer consumer = (MessageConsumer) consumers[consumerId];
if(consumer != null)
{
@@ -579,7 +529,7 @@
{
// lets iterate through each consumer created by this session
// ensuring that they have all pending messages dispatched
- lock(myLock)
+ lock(consumers.SyncRoot)
{
foreach(MessageConsumer consumer in consumers.Values)
{
@@ -588,30 +538,6 @@
}
}
- /// <summary>
- /// Returns a copy of the current consumers in a thread safe way to avoid concurrency
- /// problems if the consumers are changed in another thread
- /// </summary>
- protected ICollection GetConsumers()
- {
- lock(myLock)
- {
- return new ArrayList(consumers.Values);
- }
- }
-
- /// <summary>
- /// Returns a copy of the current consumers in a thread safe way to avoid concurrency
- /// problems if the consumers are changed in another thread
- /// </summary>
- protected ICollection GetProducers()
- {
- lock(myLock)
- {
- return new ArrayList(producers.Values);
- }
- }
-
protected virtual ConsumerInfo CreateConsumerInfo(IDestination destination, string selector)
{
ConsumerInfo answer = new ConsumerInfo();
@@ -669,19 +595,21 @@
internal void StopAsyncDelivery()
{
- if(dispatchingThread.IsStarted)
+ if(startedAsyncDelivery)
{
this.dispatchingThread.ExceptionListener -= this.dispatchingThread_ExceptionHandler;
- dispatchingThread.Stop(kMAX_STOP_ASYNC_MS);
+ dispatchingThread.Stop((int) MAX_THREAD_WAIT.TotalMilliseconds);
+ startedAsyncDelivery = false;
}
}
internal void StartAsyncDelivery()
{
- if(!dispatchingThread.IsStarted)
+ if(!startedAsyncDelivery)
{
this.dispatchingThread.ExceptionListener += this.dispatchingThread_ExceptionHandler;
dispatchingThread.Start();
+ startedAsyncDelivery = true;
}
}
Modified: activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Tcp/TcpTransport.cs
URL: http://svn.apache.org/viewvc/activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Tcp/TcpTransport.cs?rev=696392&r1=696391&r2=696392&view=diff
==============================================================================
--- activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Tcp/TcpTransport.cs (original)
+++ activemq/activemq-dotnet/Apache.NMS.ActiveMQ/trunk/src/main/csharp/Transport/Tcp/TcpTransport.cs Wed Sep 17 11:33:05 2008
@@ -25,7 +25,6 @@
namespace Apache.NMS.ActiveMQ.Transport.Tcp
{
-
/// <summary>
/// An implementation of ITransport that uses sockets to communicate with the broker
/// </summary>
@@ -38,7 +37,7 @@
private BinaryWriter socketWriter;
private Thread readThread;
private bool started;
- private volatile bool closed;
+ private AtomicBoolean closed = new AtomicBoolean(false);
private volatile bool seenShutdown;
private TimeSpan maxWait = TimeSpan.FromMilliseconds(Timeout.Infinite);
@@ -46,7 +45,6 @@
private ExceptionHandler exceptionHandler;
private TimeSpan MAX_THREAD_WAIT = TimeSpan.FromMilliseconds(30000);
-
public TcpTransport(Socket socket, IWireFormat wireformat)
{
this.socket = socket;
@@ -114,7 +112,7 @@
{
try
{
- if(closed)
+ if(closed.Value)
{
throw new InvalidOperationException("Error writing to broker. Transport connection is closed.");
}
@@ -129,6 +127,7 @@
}
catch(Exception ex)
{
+ Monitor.Exit(myLock);
if (command.ResponseRequired)
{
// Make sure that something higher up doesn't get blocked.
@@ -183,17 +182,10 @@
public void Close()
{
- if(!closed)
+ if(closed.CompareAndSet(false, true))
{
lock(myLock)
{
- if(closed)
- {
- return;
- }
-
- closed = true;
-
try
{
socket.Shutdown(SocketShutdown.Both);
@@ -300,7 +292,7 @@
// An exception in the command handler may not be fatal to the transport, so
// these are simply reported to the exceptionHandler.
//
- while(!closed)
+ while(!closed.Value)
{
Command command = null;
@@ -311,11 +303,14 @@
catch(Exception ex)
{
command = null;
- if(!closed && !seenShutdown)
+ if(!closed.Value)
{
// Close the socket as there's little that can be done with this transport now.
Close();
- this.exceptionHandler(this, ex);
+ if(!seenShutdown)
+ {
+ this.exceptionHandler(this, ex);
+ }
}
break;