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
@@ -42,9 +43,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)
@@ -183,11 +186,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;
 
@@ -232,10 +236,30 @@ internal void DispatchSignals ()
 
 		public void Iterate ()
 		{
-			Message msg = transport.ReadMessage ();
+			Iterate (new CancellationToken (false));
+		}
+
+		Task Task = null;
+		object TaskLock = new object ();
 
-			HandleMessage (msg);
-			DispatchSignals ();
+		public void Iterate (CancellationToken stopWaitToken)
+		{
+			lock (TaskLock) {
+				if (Task == null || Task.IsCompleted) {
+					Task = Task.Run ( () => {
+						var msg = transport.ReadMessage ();
+						HandleMessage (msg);
+						DispatchSignals ();
+					});
+				}
+			}
+
+			Task.Wait (stopWaitToken);
+		}
+
+		private static ManualResetEventSlim MakeNewEventToNextIterate ()
+		{
+			return new ManualResetEventSlim (true);
 		}
 
 		internal virtual void HandleMessage (Message msg)
@@ -251,21 +275,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;
 							}
 						}
@@ -391,17 +413,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 +439,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 +485,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/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;

src/Protocol/PendingCall.cs

@@ -7,23 +7,33 @@
 
 namespace DBus.Protocol
 {
-	public class PendingCall : IAsyncResult
+	public class PendingCall : IAsyncResult, IDisposable
 	{
-		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;
+		private CancellationTokenSource stopWait = new CancellationTokenSource();
+
 		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;
 			this.keepFDs = keepFDs;
 		}
 
+		public void Dispose()
+		{
+			stopWait.Dispose ();
+		}
+
 		internal bool KeepFDs
 		{
 			get {
@@ -33,36 +43,24 @@ 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) {
+					try {
+						conn.Iterate (stopWait.Token);
+					}
+					catch (OperationCanceledException) {
+					}
 				}
-
 				return reply;
-			} 
+			}
+
 			set {
 				if (reply != null)
 					throw new Exception ("Cannot handle reply more than once");
-
 				reply = value;
 
-				if (waitHandle != null)
-					waitHandle.Set ();
+				waitHandle.Set ();
+				
+				stopWait.Cancel ();
 
 				if (Completed != null)
 					Completed (reply);
@@ -84,9 +82,6 @@ object IAsyncResult.AsyncState {
 
 		WaitHandle IAsyncResult.AsyncWaitHandle {
 			get {
-				if (waitHandle == null)
-					waitHandle = new ManualResetEvent (false);
-
 				return waitHandle;
 			}
 		}