Improve Parallel Cache with Reactive Extensions & Unity Interception

Question

enter image description here

The requirement is to be able to cache method invocations made on a boundary layer (Services layer). I'm using Unity to inject the concrete implementation of the Service layer classes. The intercepting CacheCallHandler caches all the responses across this layer.

The twist is that this is a multi-threaded environment with multiple clients invoking the same boundary layer in the context of a single request. If a service layer call is already in "flight", it should wait for the original invokers return and use that.

I've used Reactive extensions to implement this.

/// <summary>
/// Intercepts the calls and tries to retrieve from the cache
/// </summary>
class CacheCallHandler : ICallHandler
{
 [Dependency]
 public ICache RequestCache { get; set; }
 public IMethodReturn Invoke(IMethodInvocation input, GetNextHandlerDelegate getNext)
 {
 IMethodReturn mesg = null;
 string cacheKey = CacheKeyGenerator.GetCacheKey(input);
 //create the task to retrieve the data
 var task = new Task<IMethodReturn>(() =>
 {
 return getNext()(input, getNext);
 });
 //make it observable
 var observableItem = task.ToObservable();
 //try to add it to the cache
 //we need to do this in the order of Add and then try to get, otherwise multiple thread might enter the same area
 if (RequestCache.TryAdd(cacheKey, observableItem))
 {
 //if the add succeeed, it means that we are responsible to starting this task
 task.Start();
 }
 else
 {
 if ( RequestCache.TryGetValue(cacheKey, out observableItem) )
 {
 //do nothing, the observable item is already updated with the requried reference
 }
 else
 {
 throw new CacheHandlerException("Could not add to cache AND could not retrieve from cache either. Something's wrong", input);
 }
 }
 //observe the return 
 if ( observableItem != null )
 mesg = observableItem.FirstOrDefault();
 if (mesg == null)
 throw new CacheHandlerException("Not return value found. this should not happen", input);
 return mesg;
 }
 /// <summary>
 /// Should always be the first to execute on the boundary
 /// </summary>
 public int Order
 {
 get { return 1; }
 set { ; }
 }
}

Answer 1

Personally i believe that there is no need for the reactive framework in this scenario. To handle the multithreaded nature of your problem i would first make sure that your RequestCache (Icache) uses the ConcurrentDictionary> and exposrts the method GetOrAdd

With this your code will look like this:

/// <summary>
/// Intercepts the calls and tries to retrieve from the cache
/// </summary>
class CacheCallHandler : ICallHandler
{
 [Dependency]
 public ICache RequestCache { get; set; }
 public IMethodReturn Invoke(IMethodInvocation input, GetNextHandlerDelegate getNext)
 {
 return InvokeAsync.Result;
 }
 public Task<IMethodReturn> InvokeAsync(IMethodInvocation input, GetNextHandlerDelegate getNext)
 {
 IMethodReturn mesg = null;
 string cacheKey = CacheKeyGenerator.GetCacheKey(input);
 //create the task to retrieve the data
 var task = RequestCache.GetOrAdd(
 cacheKey,
 key => new Task.Factory.StartNew(() => getNext()(input, getNext))
 );
 return task; 
 }
 /// <summary>
 /// Should always be the first to execute on the boundary
 /// </summary>
 public int Order
 {
 get { return 1; }
 set { ; }
 }
}

The asynchronous version offers the option of to blocking the calling thread when invoking it.