Fix race conditions is case of threaded Iterate() call

Author: nvoronchev

src/Bus.cs

@@ -59,11 +59,13 @@ public static Bus Starter
 				throw new ArgumentNullException ("address");
 
 			Bus bus;
-			if (buses.TryGetValue (address, out bus))
-				return bus;
+			lock (buses) {
+				if (buses.TryGetValue (address, out bus))
+					return bus;
 
-			bus = new Bus (address);
-			buses[address] = bus;
+				bus = new Bus (address);
+				buses[address] = bus;
+			}
 
 			return bus;
 		}

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

@@ -6,6 +6,7 @@
 using System.Collections.Generic;
 using System.IO;
 using System.Threading;
+using System.Threading.Tasks;
 using System.Reflection;
 
 namespace DBus
@@ -38,22 +39,23 @@ public class Connection
 		Dictionary<uint,PendingCall> pendingCalls = new Dictionary<uint,PendingCall> ();
 		Queue<Message> inbound = new Queue<Message> ();
 		Dictionary<ObjectPath,BusObject> registeredObjects = new Dictionary<ObjectPath,BusObject> ();
+		private readonly ReadMessageTask readMessageTask;
 
 		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
 
 		protected Connection ()
 		{
-
+			readMessageTask = new ReadMessageTask (this);
 		}
 
-		internal Connection (Transport transport)
+		internal Connection (Transport transport) : this ()
 		{
 			this.transport = transport;
 			transport.Connection = this;
 		}
 
-		internal Connection (string address)
+		internal Connection (string address) : this ()
 		{
 			OpenPrivate (address);
 			Authenticate ();
@@ -183,11 +185,12 @@ internal uint GenerateSerial ()
 
 		internal Message SendWithReplyAndBlock (Message msg, bool keepFDs)
 		{
-			PendingCall pending = SendWithReply (msg, keepFDs);
-			return pending.Reply;
+			using (PendingCall pending = SendWithPendingReply (msg, keepFDs)) {
+				return pending.Reply;
+			}
 		}
 
-		internal PendingCall SendWithReply (Message msg, bool keepFDs)
+		internal PendingCall SendWithPendingReply (Message msg, bool keepFDs)
 		{
 			msg.ReplyExpected = true;
 
@@ -215,27 +218,23 @@ internal virtual uint Send (Message msg)
 			return msg.Header.Serial;
 		}
 
-		//temporary hack
-		internal void DispatchSignals ()
+		public void Iterate ()
 		{
-			lock (inbound) {
-				while (inbound.Count != 0) {
-					Message msg = inbound.Dequeue ();
-					try {
-						HandleSignal (msg);
-					} finally {
-						msg.Dispose ();
-					}
-				}
-			}
+			Iterate (new CancellationToken (false));
 		}
 
-		public void Iterate ()
+		public void Iterate (CancellationToken stopWaitToken)
 		{
-			Message msg = transport.ReadMessage ();
-
-			HandleMessage (msg);
-			DispatchSignals ();
+			if (TryGetStoredSignalMessage (out Message inboundMsg)) {
+				try {
+					HandleSignal (inboundMsg);
+				} finally {
+					inboundMsg.Dispose ();
+				}
+			} else {
+				var msg = readMessageTask.MakeSureTaskRunAndWait (stopWaitToken);
+				HandleMessage (msg);
+			}
 		}
 
 		internal virtual void HandleMessage (Message msg)
@@ -251,21 +250,19 @@ 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;
 							}
 						}
@@ -285,8 +282,7 @@ internal virtual void HandleMessage (Message msg)
 						break;
 					case MessageType.Signal:
 						//HandleSignal (msg);
-						lock (inbound)
-							inbound.Enqueue (msg);
+						StoreInboundSignalMessage (msg); //temporary hack
 						cleanupFDs = false; // FDs are closed after signal is handled
 						break;
 					case MessageType.Error:
@@ -391,17 +387,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 +413,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 +459,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;
@@ -486,6 +488,25 @@ internal protected virtual void RemoveMatch (string rule)
 		{
 		}
 
+		private void StoreInboundSignalMessage (Message msg)
+		{
+			lock (inbound) {
+				inbound.Enqueue (msg);
+			}
+		}
+
+		private bool  TryGetStoredSignalMessage (out Message msg)
+		{
+			msg = null;
+			lock (inbound) {
+				if (inbound.Count != 0) {
+					msg = inbound.Dequeue ();
+					return true;
+				}
+			}
+			return false;
+		}
+
 		static UUID ReadMachineId (string fname)
 		{
 			byte[] data = File.ReadAllBytes (fname);
@@ -494,5 +515,31 @@ static UUID ReadMachineId (string fname)
 
 			return UUID.Parse (System.Text.Encoding.ASCII.GetString (data, 0, 32));
 		}
+	
+		private class ReadMessageTask
+		{
+			private readonly Connection ownerConnection;
+			private Task<Message> task = null;
+			private object taskLock = new object ();
+
+			public ReadMessageTask (Connection connection)
+			{
+				ownerConnection = connection;
+			}
+
+			public Message MakeSureTaskRunAndWait(CancellationToken stopWaitToken)
+			{
+				Task<Message> catchedTask = null;
+
+				lock (taskLock) {
+					if (task == null || task.IsCompleted) {
+						task = Task<Message>.Run (() => ownerConnection.transport.ReadMessage ());
+					}
+					catchedTask = task;
+				}
+				catchedTask.Wait (stopWaitToken);
+				return catchedTask.Result;
+			}
+		}
 	}
 }

src/ExportObject.cs

@@ -67,9 +67,11 @@ internal virtual void WriteIntrospect (Introspector intro)
 		internal static MethodCaller GetMCaller (MethodInfo mi)
 		{
 			MethodCaller mCaller;
-			if (!mCallers.TryGetValue (mi, out mCaller)) {
-				mCaller = TypeImplementer.GenCaller (mi);
-				mCallers[mi] = mCaller;
+			lock (mCallers) {
+				if (!mCallers.TryGetValue (mi, out mCaller)) {
+					mCaller = TypeImplementer.GenCaller (mi);
+					mCallers[mi] = mCaller;
+				}
 			}
 			return mCaller;
 		}
@@ -90,11 +92,7 @@ public virtual void HandleMethodCall (MessageContainer method_call)
 				return;
 			}
 
-			MethodCaller mCaller;
-			if (!mCallers.TryGetValue (mi, out mCaller)) {
-				mCaller = TypeImplementer.GenCaller (mi);
-				mCallers[mi] = mCaller;
-			}
+			MethodCaller mCaller = GetMCaller (mi);
 
 			Signature inSig, outSig;
 			bool hasDisposableList;