RX Extensions & Threading

Like a lot of people, I’ve been playing with Rx recently, the Reactive Extensions for .Net. To find out a bit more about what these are have a look here. One of the things I found interesting was the use of Rx as a fluent interface for creating sets of asynchronous tasks and then subscribing to the collective result with both a result handler and error handler.

I’ve worked up a little example here that shows some code performing some fairly standard asynchronous tasks. When called, my code needs to call a web service, query a database and fetch a value from some cache. It then needs to combine the results. Since these processes each take different and often varying amounts of time and I want to minimise the executing time of my code I’m going to do them in parallel.

I have shown 4 different methods of achieving this, using my own threads, the thread pool’s threads by way of BeginInvoke(), the Tasks namespace new in .NET 4.0 and Rx.

namespace RXThreadingExample
{
	using System;
	using System.Collections.Generic;
	using System.Diagnostics;
	using System.Linq;
	using System.Threading;
	using System.Threading.Tasks;

	public class Program
	{
		public static void Main()
		{
			const bool throwException = false;

			ClassicAsync(throwException);
			WaitHandles(throwException);
			Tasks(throwException);
			RXExtensions(throwException);

			Console.ReadKey(true);
		}

		private static void ClassicAsync(bool throwException)
		{
			var sw = Stopwatch.StartNew();
			var wsResult = 0;
			string dbResult = null;
			string cacheResult = null;

			var callWebService = new Thread(() => wsResult = CallWebService());
			var queryDB = new Thread(() => dbResult = QueryDB(throwException, "Async"));
			var fetchCacheItem = new Thread(() => cacheResult = FetchCacheItem());

			try
			{
				callWebService.Start();
				queryDB.Start();
				fetchCacheItem.Start();

				callWebService.Join();
				queryDB.Join();
				fetchCacheItem.Join();

				Console.WriteLine(dbResult, wsResult, cacheResult, sw.ElapsedMilliseconds);
			}
			catch (Exception ex)
			{
				Console.WriteLine("Exception : {0}", ex);
			}
		}

		private static void WaitHandles(bool throwException)
		{
			var sw = Stopwatch.StartNew();

			var waitHandles = new List<WaitHandle>();
			Func<int> callWebService = CallWebService;
			var wsAsyncResult = callWebService.BeginInvoke(null, null);			
			waitHandles.Add(wsAsyncResult.AsyncWaitHandle);

			Func<bool, string, string> queryDB = QueryDB;
			var dbAsyncResult = queryDB.BeginInvoke(throwException, "WaitHandles", null, null);
			waitHandles.Add(dbAsyncResult.AsyncWaitHandle);

			Func<string> queryLocalCache = FetchCacheItem;
			var cacheAsyncResult = queryLocalCache.BeginInvoke(null, null);
			waitHandles.Add(cacheAsyncResult.AsyncWaitHandle);

			try
			{
				WaitHandle.WaitAll(waitHandles.ToArray());

				var wsResult = callWebService.EndInvoke(wsAsyncResult);
				var dbResult = queryDB.EndInvoke(dbAsyncResult);
				var cacheResult = queryLocalCache.EndInvoke(cacheAsyncResult);

				Console.WriteLine(dbResult, wsResult, cacheResult, sw.ElapsedMilliseconds);
			}
			catch(Exception ex)
			{
				Console.WriteLine("Exception : {0}", ex);
			}
		}

		private static void Tasks(bool throwException)
		{
			var sw = Stopwatch.StartNew();
			var wsResult = 0;
			string dbResult = null;
			string cacheResult = null;

			var tasks = new List<Task>
				{
					new Task(() => wsResult = CallWebService()),
					new Task(() => dbResult = QueryDB(throwException, "Tasks")),
					new Task(() => cacheResult = FetchCacheItem())
				};

			try
			{
				tasks.ForEach(t => t.Start());
				Task.WaitAll(tasks.ToArray());
				Console.WriteLine(dbResult, wsResult, cacheResult, sw.ElapsedMilliseconds);
			}
			catch (Exception ex)
			{
				Console.WriteLine("Exception : {0}", ex);
			}

		}

		private static void RXExtensions(bool throwException)
		{
			var sw = Stopwatch.StartNew();
			Observable.Join(
				Observable.ToAsync<int>(CallWebService)()
					.And(Observable.ToAsync<bool, string, string>(QueryDB)(throwException, "RX"))
					.And(Observable.ToAsync<string>(FetchCacheItem)())
					.Then((wsResult, dbResult, cacheValue) =>
						new { WebServiceResult = wsResult, DatabaseResult = dbResult, CacheValue = cacheValue })
				).Subscribe(
					o => Console.WriteLine(o.DatabaseResult, o.WebServiceResult, o.CacheValue, sw.ElapsedMilliseconds),
					e => Console.WriteLine("Exception: {0}", e));
		}

		private static int CallWebService()
		{
			Thread.Sleep(500);
			return new Random().Next(1,33);
		}

		private static string QueryDB(bool throwException, string name)
		{
			Thread.Sleep(1500);
			if (throwException)
			{
				throw new Exception("You asked for it !");
			}
			return name + " can rescue {0} Chilean miners in {2} ms. {1}";
		}

		private static string FetchCacheItem()
		{
			return new[]{"Awesome!", "Cool!", "Amazing!", "Jinkies!"}[new Random().Next(0,4)];
		}
	}
}

I have to say, I like the syntactic sugar of Rx. In the tests I have run, however, using Task consistently produces the fastest results. More to follow, I think.

Advertisements

Leave a Reply

Fill in your details below or click an icon to log in:

WordPress.com Logo

You are commenting using your WordPress.com account. Log Out / Change )

Twitter picture

You are commenting using your Twitter account. Log Out / Change )

Facebook photo

You are commenting using your Facebook account. Log Out / Change )

Google+ photo

You are commenting using your Google+ account. Log Out / Change )

Connecting to %s

%d bloggers like this: