Various bug fixes and XML comments (see README for 1.0.1)

haneytron · haneytron · commit 77ff16870eed · 2013-11-12T23:38:43.000-05:00
diff --git a/README.md b/README.md
@@ -1,6 +1,7 @@
-SIMPLSOCKETS 1.0.0
+SIMPLSOCKETS 1.0.1
 ===========
 
+
 A spinoff library of Dache that provides highly efficient, scalable, simple socket communication.
 
 http://www.getdache.net
@@ -12,6 +13,23 @@ VERSION HISTORY
 ============================================
 
 
+1.0.1
+------------------
+
+
+-Fixed logic error in BlockingQueue constructor where _queue wasn't actually assigned
+
+-Fixed logic error in Pool where resetItemMethod was not always called when Popping an item
+
+-Fixed atomicity of Error event so that it is raised exactly once on disconnection regardless of multithreaded use
+
+-On error, communication methods now exit gracefully after Error event is raised (no bubbled exceptions)
+
+-Exposed CurrentlyConnectedClients property on ISimplSocketServer
+
+-Added XML comments to a few classes
+
+
 1.0.0
 ------------------
 
diff --git a/SimplSockets/BlockingQueue.cs b/SimplSockets/BlockingQueue.cs
@@ -4,21 +4,38 @@
 
 namespace SimplSockets
 {
+    /// <summary>
+    /// A queue that wraps a regular generic queue but when empty will block Dequeue threads until an item is available.
+    /// </summary>
+    /// <typeparam name="T">The type of the object contained in the queue.</typeparam>
     internal sealed class BlockingQueue<T>
     {
+        // The underlying queue
         Queue<T> _queue = null;
+        // The semaphore used for blocking
         Semaphore _semaphore = new Semaphore(0, Int32.MaxValue);
 
+        /// <summary>
+        /// The constructor.
+        /// </summary>
         public BlockingQueue()
         {
-            new Queue<T>();
+            _queue = new Queue<T>();
         }
 
+        /// <summary>
+        /// The constructor.
+        /// </summary>
+        /// <param name="capacity">Sets the initial queue capacity.</param>
         public BlockingQueue(int capacity)
         {
             _queue = new Queue<T>(capacity);
         }
 
+        /// <summary>
+        /// Enqueues an item.
+        /// </summary>
+        /// <param name="item">An item.</param>
         public void Enqueue(T item)
         {
             lock (_queue)
@@ -29,6 +46,10 @@ public void Enqueue(T item)
             _semaphore.Release();
         }
 
+        /// <summary>
+        /// Dequeues an item. Will block if the queue is empty until an item becomes available.
+        /// </summary>
+        /// <returns>An item.</returns>
         public T Dequeue()
         {
             _semaphore.WaitOne();
diff --git a/SimplSockets/ISimplSocketServer.cs b/SimplSockets/ISimplSocketServer.cs
@@ -26,6 +26,11 @@ public interface ISimplSocketServer : IDisposable
         /// </summary>
         void Close();
 
+        /// <summary>
+        /// Gets the currently connected client count.
+        /// </summary>
+        int CurrentlyConnectedClients { get; }
+
         /// <summary>
         /// An event that is fired whenever a message is received. Hook into this to process messages and potentially call Reply to send a response.
         /// </summary>
diff --git a/SimplSockets/Pool.cs b/SimplSockets/Pool.cs
@@ -3,13 +3,27 @@
 
 namespace SimplSockets
 {
+    /// <summary>
+    /// A pool of objects that can be reused to manage memory efficiently.
+    /// </summary>
+    /// <typeparam name="T">The type of object that is pooled.</typeparam>
     internal sealed class Pool<T> where T : class
     {
+        // The queue that holds the items
         private readonly Queue<T> _queue = null;
+        // The initial pool count
         private readonly int _initialPoolCount = 0;
+        // The method that creates a new item
         private readonly Func<T> _newItemMethod = null;
+        // The method that resets an item's state
         private readonly Action<T> _resetItemMethod = null;
 
+        /// <summary>
+        /// The constructor.
+        /// </summary>
+        /// <param name="poolCount">The count of items in the pool.</param>
+        /// <param name="newItemMethod">The method that creates a new item.</param>
+        /// <param name="resetItemMethod">The method that resets an item's state.</param>
         public Pool(int poolCount, Func<T> newItemMethod, Action<T> resetItemMethod)
         {
             _queue = new Queue<T>(poolCount);
@@ -18,6 +32,10 @@ public Pool(int poolCount, Func<T> newItemMethod, Action<T> resetItemMethod)
             _resetItemMethod = resetItemMethod;
         }
 
+        /// <summary>
+        /// Pushes an item into the pool for later re-use.
+        /// </summary>
+        /// <param name="item">The item.</param>
         public void Push(T item)
         {
             // Limit queue size
@@ -32,10 +50,15 @@ public void Push(T item)
             }
         }
 
+        /// <summary>
+        /// Pops an item out of the pool for use. The item will have its state reset.
+        /// </summary>
+        /// <returns></returns>
         public T Pop()
         {
             T result = null;
 
+            // Cheap check
             if (_queue.Count == 0)
             {
                 result = _newItemMethod();
@@ -48,9 +71,15 @@ public T Pop()
 
             lock (_queue)
             {
+                // Double lock check
                 if (_queue.Count == 0)
                 {
-                    return _newItemMethod();
+                    result = _newItemMethod();
+                    if (_resetItemMethod != null)
+                    {
+                        _resetItemMethod(result);
+                    }
+                    return result;
                 }
 
                 result = _queue.Dequeue();
diff --git a/SimplSockets/SimplSocket.cs b/SimplSockets/SimplSocket.cs
@@ -181,6 +181,11 @@ public void Connect(EndPoint endPoint)
             {
                 HandleCommunicationError(_socket, ex);
             }
+            catch (ObjectDisposedException)
+            {
+                // If disposed, handle communication error was already done and we're just catching up on other threads. Supress it.
+                return;
+            }
 
             // Get a message state from the pool
             var messageState = _messageStatePool.Pop();
@@ -222,7 +227,7 @@ public void Listen(EndPoint localEndpoint)
 
             // Create socket
             _socket = _socketFunc();
-            
+
             try
             {
                 _socket.Bind(localEndpoint);
@@ -234,7 +239,10 @@ public void Listen(EndPoint localEndpoint)
             catch (SocketException ex)
             {
                 HandleCommunicationError(_socket, ex);
-                return;
+            }
+            catch (ObjectDisposedException)
+            {
+                // If disposed, handle communication error was already done and we're just catching up on other threads. Supress it.
             }
         }
 
@@ -288,6 +296,11 @@ public void Send(byte[] message)
             {
                 HandleCommunicationError(_socket, ex);
             }
+            catch (ObjectDisposedException)
+            {
+                // If disposed, handle communication error was already done and we're just catching up on other threads. Supress it.
+                return;
+            }
         }
 
         /// <summary>
@@ -324,6 +337,11 @@ public byte[] SendReceive(byte[] message)
                 HandleCommunicationError(_socket, ex);
                 return null;
             }
+            catch (ObjectDisposedException)
+            {
+                // If disposed, handle communication error was already done and we're just catching up on other threads. Supress it.
+                return null;
+            }
 
             // Wait for our message to go ahead from the receive callback
             multiplexerData.ManualResetEvent.WaitOne();
@@ -370,6 +388,11 @@ public void Reply(byte[] message, ReceivedMessage receivedMessage)
                 HandleCommunicationError(receivedMessage.Socket, ex);
                 return;
             }
+            catch (ObjectDisposedException)
+            {
+                // If disposed, handle communication error was already done and we're just catching up on other threads. Supress it.
+                return;
+            }
 
             // Put received message back in the pool
             _receiveMessagePool.Push(receivedMessage);
@@ -409,6 +432,12 @@ private void AcceptCallback(IAsyncResult asyncResult)
                 HandleCommunicationError(_socket, ex);
                 return;
             }
+            catch (ObjectDisposedException)
+            {
+                // If disposed, handle communication error was already done and we're just catching up on other threads. Supress it.
+                return;
+            }
+
             // Turn on or off Nagle algorithm
             handler.NoDelay = !_useNagleAlgorithm;
 
@@ -422,6 +451,11 @@ private void AcceptCallback(IAsyncResult asyncResult)
                 HandleCommunicationError(_socket, ex);
                 return;
             }
+            catch (ObjectDisposedException)
+            {
+                // If disposed, handle communication error was already done and we're just catching up on other threads. Supress it.
+                return;
+            }
 
             // Do not proceed until we have room to do so
             _maxConnectionsSemaphore.WaitOne();
@@ -442,6 +476,11 @@ private void AcceptCallback(IAsyncResult asyncResult)
                 HandleCommunicationError(handler, ex);
                 return;
             }
+            catch (ObjectDisposedException)
+            {
+                // If disposed, handle communication error was already done and we're just catching up on other threads. Supress it.
+                return;
+            }
 
             // Create receive queue for this client
             _receiveBufferQueueLock.EnterWriteLock();
@@ -473,6 +512,11 @@ private void SendCallback(IAsyncResult asyncResult)
                 HandleCommunicationError(socket, ex);
                 return;
             }
+            catch (ObjectDisposedException)
+            {
+                // If disposed, handle communication error was already done and we're just catching up on other threads. Supress it.
+                return;
+            }
         }
 
         private void ReceiveCallback(IAsyncResult asyncResult)
@@ -493,6 +537,11 @@ private void ReceiveCallback(IAsyncResult asyncResult)
                 HandleCommunicationError(messageState.Handler, ex);
                 return;
             }
+            catch (ObjectDisposedException)
+            {
+                // If disposed, handle communication error was already done and we're just catching up on other threads. Supress it.
+                return;
+            }
 
             if (bytesRead > 0)
             {
@@ -522,7 +571,10 @@ private void ReceiveCallback(IAsyncResult asyncResult)
                 catch (SocketException ex)
                 {
                     HandleCommunicationError(messageState.Handler, ex);
-                    return;
+                }
+                catch (ObjectDisposedException)
+                {
+                    // If disposed, handle communication error was already done and we're just catching up on other threads. Supress it.
                 }
             }
         }
@@ -664,17 +716,26 @@ private void ProcessReceivedMessage(MessageState messageState)
         /// <param name="ex">The exception that the socket raised.</param>
         private void HandleCommunicationError(Socket socket, Exception ex)
         {
-            // Close the socket
-            try
+            lock (socket)
             {
-                socket.Shutdown(SocketShutdown.Both);
-            }
-            catch
-            {
-                // Ignore
-            }
+                // Close the socket
+                try
+                {
+                    socket.Shutdown(SocketShutdown.Both);
+                }
+                catch (ObjectDisposedException)
+                {
+                    // Socket was already closed/disposed, so return out to prevent raising the Error event multiple times
+                    // This is most likely to happen when an error occurs during heavily multithreaded use
+                    return;
+                }
+                catch
+                {
+                    // Ignore
+                }
 
-            socket.Close();
+                socket.Close();
+            }
 
             // Release all multiplexer clients by signalling them
             _clientMultiplexerLock.EnterReadLock();

Original file line number	Diff line number	Diff line change
`@@ -4,21 +4,38 @@`
`4`	`4`
`5`	`5`	`namespace SimplSockets`
`6`	`6`	`{`
	`7`	`+ /// <summary>`
	`8`	`+ /// A queue that wraps a regular generic queue but when empty will block Dequeue threads until an item is available.`
	`9`	`+ /// </summary>`
	`10`	`+ /// <typeparam name="T">The type of the object contained in the queue.</typeparam>`
`7`	`11`	`internal sealed class BlockingQueue<T>`
`8`	`12`	`{`
	`13`	`+ // The underlying queue`
`9`	`14`	`Queue<T> _queue = null;`
	`15`	`+ // The semaphore used for blocking`
`10`	`16`	`Semaphore _semaphore = new Semaphore(0, Int32.MaxValue);`
`11`	`17`
	`18`	`+ /// <summary>`
	`19`	`+ /// The constructor.`
	`20`	`+ /// </summary>`
`12`	`21`	`public BlockingQueue()`
`13`	`22`	`{`
`14`		`- new Queue<T>();`
	`23`	`+ _queue = new Queue<T>();`
`15`	`24`	`}`
`16`	`25`
	`26`	`+ /// <summary>`
	`27`	`+ /// The constructor.`
	`28`	`+ /// </summary>`
	`29`	`+ /// <param name="capacity">Sets the initial queue capacity.</param>`
`17`	`30`	`public BlockingQueue(int capacity)`
`18`	`31`	`{`
`19`	`32`	`_queue = new Queue<T>(capacity);`
`20`	`33`	`}`
`21`	`34`
	`35`	`+ /// <summary>`
	`36`	`+ /// Enqueues an item.`
	`37`	`+ /// </summary>`
	`38`	`+ /// <param name="item">An item.</param>`
`22`	`39`	`public void Enqueue(T item)`
`23`	`40`	`{`
`24`	`41`	`lock (_queue)`
`@@ -29,6 +46,10 @@ public void Enqueue(T item)`
`29`	`46`	`_semaphore.Release();`
`30`	`47`	`}`
`31`	`48`
	`49`	`+ /// <summary>`
	`50`	`+ /// Dequeues an item. Will block if the queue is empty until an item becomes available.`
	`51`	`+ /// </summary>`
	`52`	`+ /// <returns>An item.</returns>`
`32`	`53`	`public T Dequeue()`
`33`	`54`	`{`
`34`	`55`	`_semaphore.WaitOne();`
Original file line number	Diff line number	Diff line change
`@@ -3,13 +3,27 @@`
`3`	`3`
`4`	`4`	`namespace SimplSockets`
`5`	`5`	`{`
	`6`	`+ /// <summary>`
	`7`	`+ /// A pool of objects that can be reused to manage memory efficiently.`
	`8`	`+ /// </summary>`
	`9`	`+ /// <typeparam name="T">The type of object that is pooled.</typeparam>`
`6`	`10`	`internal sealed class Pool<T> where T : class`
`7`	`11`	`{`
	`12`	`+ // The queue that holds the items`
`8`	`13`	`private readonly Queue<T> _queue = null;`
	`14`	`+ // The initial pool count`
`9`	`15`	`private readonly int _initialPoolCount = 0;`
	`16`	`+ // The method that creates a new item`
`10`	`17`	`private readonly Func<T> _newItemMethod = null;`
	`18`	`+ // The method that resets an item's state`
`11`	`19`	`private readonly Action<T> _resetItemMethod = null;`
`12`	`20`
	`21`	`+ /// <summary>`
	`22`	`+ /// The constructor.`
	`23`	`+ /// </summary>`
	`24`	`+ /// <param name="poolCount">The count of items in the pool.</param>`
	`25`	`+ /// <param name="newItemMethod">The method that creates a new item.</param>`
	`26`	`+ /// <param name="resetItemMethod">The method that resets an item's state.</param>`
`13`	`27`	`public Pool(int poolCount, Func<T> newItemMethod, Action<T> resetItemMethod)`
`14`	`28`	`{`
`15`	`29`	`_queue = new Queue<T>(poolCount);`
`@@ -18,6 +32,10 @@ public Pool(int poolCount, Func<T> newItemMethod, Action<T> resetItemMethod)`
`18`	`32`	`_resetItemMethod = resetItemMethod;`
`19`	`33`	`}`
`20`	`34`
	`35`	`+ /// <summary>`
	`36`	`+ /// Pushes an item into the pool for later re-use.`
	`37`	`+ /// </summary>`
	`38`	`+ /// <param name="item">The item.</param>`
`21`	`39`	`public void Push(T item)`
`22`	`40`	`{`
`23`	`41`	`// Limit queue size`
`@@ -32,10 +50,15 @@ public void Push(T item)`
`32`	`50`	`}`
`33`	`51`	`}`
`34`	`52`
	`53`	`+ /// <summary>`
	`54`	`+ /// Pops an item out of the pool for use. The item will have its state reset.`
	`55`	`+ /// </summary>`
	`56`	`+ /// <returns></returns>`
`35`	`57`	`public T Pop()`
`36`	`58`	`{`
`37`	`59`	`T result = null;`
`38`	`60`
	`61`	`+ // Cheap check`
`39`	`62`	`if (_queue.Count == 0)`
`40`	`63`	`{`
`41`	`64`	`result = _newItemMethod();`
`@@ -48,9 +71,15 @@ public T Pop()`
`48`	`71`
`49`	`72`	`lock (_queue)`
`50`	`73`	`{`
	`74`	`+ // Double lock check`
`51`	`75`	`if (_queue.Count == 0)`
`52`	`76`	`{`
`53`		`- return _newItemMethod();`
	`77`	`+ result = _newItemMethod();`
	`78`	`+ if (_resetItemMethod != null)`
	`79`	`+ {`
	`80`	`+ _resetItemMethod(result);`
	`81`	`+ }`
	`82`	`+ return result;`
`54`	`83`	`}`
`55`	`84`
`56`	`85`	`result = _queue.Dequeue();`