Fix race conditions is case of threaded Iterate() loop and methods call

Author: nvoronchev

src/BusException.cs

@@ -25,7 +25,7 @@ public override string Message
 		{
 			get
 			{
-				return ErrorName + ": " + ErrorMessage;
+				return ErrorName + ": " + ErrorMessage + Environment.NewLine + this.StackTrace;
 			}
 		}
 

src/Connection.cs

@@ -42,9 +42,11 @@ public class Connection
 		public delegate void MonitorEventHandler (Message msg);
 		public MonitorEventHandler Monitors; // subscribe yourself to this list of observers if you want to get notified about each incoming message
 
+		private ManualResetEventSlim iterateEvent = MakeNewEventToNextIterate ();
+		private readonly object iterateLocker = new object ();
+
 		protected Connection ()
 		{
-
 		}
 
 		internal Connection (Transport transport)
@@ -230,12 +232,38 @@ internal void DispatchSignals ()
 			}
 		}
 
-		public void Iterate ()
+		public void Iterate()
 		{
-			Message msg = transport.ReadMessage ();
+			bool lockAcquired = false;
+
+			lock (iterateLocker) {
+				if (iterateEvent.IsSet) {
+					iterateEvent.Reset ();
+					lockAcquired = true;
+				}
+			}
+
+			if (lockAcquired) {
+				// One therad acquired a semi-lock. This thread read message from the stream.
+				Message msg = transport.ReadMessage ();
+				HandleMessage (msg);
 
-			HandleMessage (msg);
-			DispatchSignals ();
+				lock (iterateLocker) {
+					iterateEvent.Set ();
+					iterateEvent = MakeNewEventToNextIterate ();
+				}
+
+				DispatchSignals ();
+			}
+			else {
+				// All other treads wait the massage will be read and handeled.
+				iterateEvent.Wait ();
+			}
+		}
+
+		private static ManualResetEventSlim MakeNewEventToNextIterate ()
+		{
+			return new ManualResetEventSlim (true);
 		}
 
 		internal virtual void HandleMessage (Message msg)
@@ -251,21 +279,18 @@ internal virtual void HandleMessage (Message msg)
 			try {
 
 				//TODO: Restrict messages to Local ObjectPath?
-
 				{
-					object field_value = msg.Header[FieldCode.ReplySerial];
+					object field_value = msg.Header [FieldCode.ReplySerial];
 					if (field_value != null) {
 						uint reply_serial = (uint)field_value;
-						PendingCall pending;
-
 						lock (pendingCalls) {
+							PendingCall pending;
 							if (pendingCalls.TryGetValue (reply_serial, out pending)) {
-								if (pendingCalls.Remove (reply_serial)) {
-									pending.Reply = msg;
-									if (pending.KeepFDs)
-										cleanupFDs = false; // caller is responsible for closing FDs
-								}
-
+								if (!pendingCalls.Remove (reply_serial))
+									return;
+								pending.Reply = msg;
+								if (pending.KeepFDs)
+									cleanupFDs = false; // caller is responsible for closing FDs
 								return;
 							}
 						}
@@ -391,17 +416,19 @@ internal void HandleMethodCall (MessageContainer method_call)
 				//this is messy and inefficient
 				List<string> linkNodes = new List<string> ();
 				int depth = method_call.Path.Decomposed.Length;
-				foreach (ObjectPath pth in registeredObjects.Keys) {
-					if (pth.Value == (method_call.Path.Value)) {
-						ExportObject exo = (ExportObject)registeredObjects[pth];
-						exo.WriteIntrospect (intro);
-					} else {
-						for (ObjectPath cur = pth ; cur != null ; cur = cur.Parent) {
-							if (cur.Value == method_call.Path.Value) {
-								string linkNode = pth.Decomposed[depth];
-								if (!linkNodes.Contains (linkNode)) {
-									intro.WriteNode (linkNode);
-									linkNodes.Add (linkNode);
+				lock (registeredObjects) {
+					foreach (ObjectPath pth in registeredObjects.Keys) {
+						if (pth.Value == (method_call.Path.Value)) {
+							ExportObject exo = (ExportObject) registeredObjects [pth];
+							exo.WriteIntrospect (intro);
+						} else {
+							for (ObjectPath cur = pth; cur != null; cur = cur.Parent) {
+								if (cur.Value == method_call.Path.Value) {
+									string linkNode = pth.Decomposed [depth];
+									if (!linkNodes.Contains (linkNode)) {
+										intro.WriteNode (linkNode);
+										linkNodes.Add (linkNode);
+									}
 								}
 							}
 						}
@@ -415,12 +442,14 @@ internal void HandleMethodCall (MessageContainer method_call)
 				return;
 			}
 
-			BusObject bo;
-			if (registeredObjects.TryGetValue (method_call.Path, out bo)) {
-				ExportObject eo = (ExportObject)bo;
-				eo.HandleMethodCall (method_call);
-			} else {
-				MaybeSendUnknownMethodError (method_call);
+			lock (registeredObjects) {
+				BusObject bo;
+				if (registeredObjects.TryGetValue(method_call.Path, out bo)) {
+					ExportObject eo = (ExportObject)bo;
+					eo.HandleMethodCall (method_call);
+				} else {
+					MaybeSendUnknownMethodError (method_call);
+				}
 			}
 		}
 
@@ -459,17 +488,19 @@ public void Register (ObjectPath path, object obj)
 			eo.Registered = true;
 
 			//TODO: implement some kind of tree data structure or internal object hierarchy. right now we are ignoring the name and putting all object paths in one namespace, which is bad
-			registeredObjects[path] = eo;
+			lock (registeredObjects)
+				registeredObjects[path] = eo;
 		}
 
 		public object Unregister (ObjectPath path)
 		{
 			BusObject bo;
 
-			if (!registeredObjects.TryGetValue (path, out bo))
-				throw new Exception ("Cannot unregister " + path + " as it isn't registered");
-
-			registeredObjects.Remove (path);
+			lock (registeredObjects) {
+				if (!registeredObjects.TryGetValue (path, out bo))
+					throw new Exception ("Cannot unregister " + path + " as it isn't registered");
+				registeredObjects.Remove (path);
+			}
 
 			ExportObject eo = (ExportObject)bo;
 			eo.Registered = false;

src/Protocol/PendingCall.cs

@@ -9,15 +9,19 @@ namespace DBus.Protocol
 {
 	public class PendingCall : IAsyncResult
 	{
-		Connection conn;
-		Message reply;
-		ManualResetEvent waitHandle;
-		bool completedSync;
-		bool keepFDs;
-		
+		private Connection conn;
+		private Message reply;
+		private ManualResetEvent waitHandle = new ManualResetEvent (false);
+		private bool completedSync = false;
+		private bool keepFDs;
+
 		public event Action<Message> Completed;
 
-		public PendingCall (Connection conn) : this (conn, false) {}
+		public PendingCall(Connection conn)
+			: this (conn, false)
+		{
+		}
+
 		public PendingCall (Connection conn, bool keepFDs)
 		{
 			this.conn = conn;
@@ -33,36 +37,17 @@ internal bool KeepFDs
 
 		public Message Reply {
 			get {
-				if (reply != null)
-					return reply;
-
-				if (Thread.CurrentThread == conn.mainThread) {
-					while (reply == null)
-						conn.HandleMessage (conn.Transport.ReadMessage ());
-
-					completedSync = true;
-
-					conn.DispatchSignals ();
-				} else {
-					if (waitHandle == null)
-						Interlocked.CompareExchange (ref waitHandle, new ManualResetEvent (false), null);
-
-					while (reply == null)
-						waitHandle.WaitOne ();
-
-					completedSync = false;
-				}
-
+				while (reply == null)
+					conn.Iterate ();
 				return reply;
-			} 
+			}
+
 			set {
 				if (reply != null)
 					throw new Exception ("Cannot handle reply more than once");
-
 				reply = value;
 
-				if (waitHandle != null)
-					waitHandle.Set ();
+				waitHandle.Set ();
 
 				if (Completed != null)
 					Completed (reply);
@@ -84,9 +69,6 @@ object IAsyncResult.AsyncState {
 
 		WaitHandle IAsyncResult.AsyncWaitHandle {
 			get {
-				if (waitHandle == null)
-					waitHandle = new ManualResetEvent (false);
-
 				return waitHandle;
 			}
 		}

src/Protocol/Transport.cs

@@ -14,6 +14,7 @@ namespace DBus.Transports
 	abstract class Transport
 	{
 		readonly object writeLock = new object ();
+		readonly object readLock = new object ();
 
 		[ThreadStatic]
 		static byte[] readBuffer;
@@ -154,12 +155,26 @@ protected void FireWakeUp ()
 				WakeUp (this, EventArgs.Empty);
 		}
 
+
+		internal virtual void WriteMessage(Message msg)
+		{
+			lock (writeLock)
+			{
+				msg.Header.GetHeaderDataToStream (stream);
+				if (msg.Body != null && msg.Body.Length != 0)
+					stream.Write (msg.Body, 0, msg.Body.Length);
+				stream.Flush ();
+			}
+		}
+
 		internal Message ReadMessage ()
 		{
 			Message msg;
 
 			try {
-				msg = ReadMessageReal ();
+				lock (readLock)
+					msg = ReadMessageReal ();
+
 				if (msg == null) {
 					if (connection.IsConnected) {
 						if (ProtocolInformation.Verbose)
@@ -173,30 +188,17 @@ internal Message ReadMessage ()
 				connection.IsConnected = false;
 				msg = null;
 			}
-         
+
 			if (connection != null && connection.Monitors != null)
 				connection.Monitors (msg);
 
 			return msg;
 		}
 
-		internal virtual int Read (byte[] buffer, int offset, int count, UnixFDArray fdArray)
-		{
-			int read = 0;
-			while (read < count) {
-				int nread = stream.Read (buffer, offset + read, count - read);
-				if (nread == 0)
-					break;
-				read += nread;
-			}
-
-			if (read > count)
-				throw new Exception ();
-
-			return read;
-		}
-
-		Message ReadMessageReal ()
+		/// <summary>
+		/// Thread unsafe! Safely called from <see cref="ReadMessage"/>.
+		/// </summary>
+		private Message ReadMessageReal ()
 		{
 			byte[] header = null;
 			byte[] body = null;
@@ -266,18 +268,24 @@ Message ReadMessageReal ()
 			}
 
 			Message msg = Message.FromReceivedBytes (Connection, header, body, fdArray);
-
 			return msg;
 		}
 
-		internal virtual void WriteMessage (Message msg)
+		internal virtual int Read (byte[] buffer, int offset, int count, UnixFDArray fdArray)
 		{
-			lock (writeLock) {
-				msg.Header.GetHeaderDataToStream (stream);
-				if (msg.Body != null && msg.Body.Length != 0)
-					stream.Write (msg.Body, 0, msg.Body.Length);
-				stream.Flush ();
+			int read = 0;
+			while (read < count)
+			{
+				int nread = stream.Read (buffer, offset + read, count - read);
+				if (nread == 0)
+					break;
+				read += nread;
 			}
+
+			if (read > count)
+				throw new Exception ();
+
+			return read;
 		}
 
 		// Returns true if then transport supports unix FDs, even when the