What are we going to do tomorrow night, Brain? : IDisposable, Threading

Thread Safe Library Update

Brian Rudolph — Wed, 02 Sep 2009 16:33:00 GMT

As always, my life revolves around threading. So I have a few classes that I use frequently for these purposes. Now, fortunately, .NET 4.0 will include much of this functionality in PFX, but for now, we have this.

ThreadSafeLibrary

AutoProcessingQueue: This is a queue that as you add things to it, will automatically begin dispatching on a seperate thread and begin calling a specified delegate to process the items.

Batch: This is a class that is similar to a queue except that it allows you to do work in batches. Set the max & minimum sizes, and start adding items to it. When it reaches a certain size it will attempt to close the batch and will fire an event that will allow you to wire in code for processing the data in the batch. It also has a timer that can be enabled if you'd prefer to allow the batch to be closed at a certain interval. This is great for database calls and bulk inserts. Say I want to send a series of messages, but i don't want to send them one at a time. I would like to send the messages when the batch reaches 100 items, or every 5 seconds. I use this class alot for messaging.

Dispatcher: This is the class that does the heavy lifting for the majority of things in this library. You give it a maximum number of threads to work with, and as you add workers to it's queue, it will begin spawning threads to process them(carefully re-using them whenever possible)

ResourcePool: The dispatcher uses this internally to store & manage it's threads. Though it can be used for any object type. One case where i use this frequently is to pool connections to webservices. To save the time required to open and close connections constantly, i set up a pool, give it a threshold(so as to not flood the service i'm calling) and ask it to get the resource for you via the delegate you provide.

ThreadSafeDictionary<K,V>: The classic threadsafe implementation of a dictionary, with several added methods to support atomic operations.

ThreadSafeQueue<T>: The classic Queue implementation of a dictionary, with several added methods to support atomic operations.

ThreadSafeHashSet<T>: The classic HashSet implementation of a dictionary, with several added methods to support atomic operations.

ThreadSafeStackList<T>: This is similar to a stack, without alot of the limitations. Multi-level peek, unwind, etc.

AsyncBlock:
This is a class that allows you to wrap async work into serial action for the purposes of simplicity. If you need have a function that needs to run numerous asynchronous calls, and then return only when they are all finished, this is your class.

You can use it like so:

     public class AsyncSample
    {
        private Dispatcher Dispatcher;


        public AsyncSample()
        {
            //init a dispatcher that uses 5 threads and doesn't carry execution context
            Dispatcher = new Dispatcher(5, false);
        }


        public void ExecuteParallel<T>(IList<T> work)
        {
            var workers = ConstructParallelWorkers(work);


            //the async block will pause this thread until all workers finish
            using (AsyncBlock block = new AsyncBlock(this.Dispatcher, 10))
            {
                foreach (var worker in workers)
                {
                    block.AddToQueue(worker);
                }
            }
        }


        private IList<AnyWorkerBee<T>> ConstructParallelWorkers<T>(IList<T> work)
        {
            var workers = new List<AnyWorkerBee<T>>();


            foreach (var workUnit in work)
                workers.Add(new AnyWorkerBee<T>() { WorkUnit = workUnit });


            return workers;
        }
    }

Inline Asynchronous UI Coding using IDisposable in C# .NET

Brian Rudolph — Mon, 15 Dec 2008 17:00:00 GMT

I recently watched this video on the PowerThreading library: http://blogs.msdn.com/charlie/archive/2008/12/03/jeff-richter-video-on-asynchronous-programming-and-his-power-threading-library.aspx

This is very interesting as Jeff Richter utilizes a strange behavior in the way the C# compiler works with IEnumerables to make a useful and easy-to-use async library. His primary goal was to allow a website\service yield processing during I/O operations, which can provide for better scalability in your app tier.

While this is quite interesting, I personally would avoid using strange compilation patterns to my advantage in an application. Specifically because you have no assurance that the compiler will always do this as the language and subsequent compilers evolve. What I was interested in was the mechanism for including asynchronous calls into code without using a complex series of callbacks.

Now, to be quite fair, Jeff's implementation was an attempt to limit thread spawning, and context switches. I am not going to account for this. I am more interested in parrallel processing.

Let me give you an example. Take for instance a series of objects that the UI is going to take cues from and make subsequent database calls to retrieve extra data for. A simple example is a treeview. If I have a bunch of nodes in a tree and the UI requires that I display the child node or items count for each of these nodes. What you would generally do is loop through the objects and call the DB to retrieve the data, and assign the UI components in one serial block of code. The problem with this is that the database is more than likely capable of servicing the request for numerous nodes simultaneously, but in the UI we are implementing this serially. If the tree structure is complex, and you are showing more than just a node count, it is plausible that each of these requests will take upward of 50ms. If you have several dozen nodes being displayed this can easily add seconds to your response time.

A serial block of code might look like this:

TreeNodeCollection nodes = new TreeNodeCollection();
foreach (TreeNode node in nodes)
{
     int objectCount = database.DoSomething((MyObject)node.Tag); //call the database
     node.Text = String.Format("{0} ({1})", node.Text, objectCount); //update the UI
 }

Simple and clean, but slow.

Why not do all of these calls asynchronously, wait for all of the results to filter in, and then update the UI? This should surely be faster than a serial operation. Let's say, we spawn Math.Min(nodes.Count, 10) threads to accomplish this task. So at a maximum, we will use 10 threads to do the work. If we have 20 nodes, it should take at the longest 2 times as long as the longest running call.

The usual issue with this is that it requires you to use several call-back methods, do complicated thread monitoring, and subsequently put the UI into a wait state until all of the external threads have finished. Unfortunately there just is no simple way to do it.

So I got to thinking, what construct can I use to signal an asynchronous block, that would prevent execution beyond it's bounds until all the threads were done. The answer, IDisposable. If I created an IDisposable object and wrapped it in a "using" I could prevent the Dispose method from exiting until all the threads have finished.

Utilizing the rest of my Threading objects, this should be fairly simple to accomplish, all I had to do was add a done flag to my workerbee:

public class AsyncBlock : IDisposable
{


        long timeoutMS = 0;
        Dispatcher dispatcher = null;
        bool started = false;

        public AsyncBlock() : this(10) { }

        public AsyncBlock(int timeoutSeconds) : this(timeoutSeconds, 5) { }

        public AsyncBlock(int timeoutSeconds, int threads) : this(new Dispatcher(threads), timeoutSeconds) { }

        public AsyncBlock(Dispatcher dispatcher, int timeoutSeconds)
        {
            this.dispatcher = dispatcher;
            this.timeoutMS = timeoutSeconds * 1000;
            this.WorkQueue = new List<WorkerBee>();
        }


        private IList<WorkerBee> WorkQueue
        {
            get;
            set;
        } 


       public IList<WorkerBee> GetWorkers()        
       {
            return new List<WorkerBee>(this.WorkQueue);
        } 


        public void AddToQueue(WorkerBee bee)
        {
            this.WorkQueue.Add(bee);
            this.dispatcher.AddToQueue(bee);
            started = true;
        }


        public void Start()
        {
            foreach (WorkerBee bee in this.WorkQueue)
                if (!bee.Done)
                    this.dispatcher.AddToQueue(bee);


            started = true;
        }


        public void Dispose()
        {
            if (!started && WorkQueue.Count != 0)
            {
                throw new Exception("Work was never started");
            }


            System.Diagnostics.Stopwatch watch = new System.Diagnostics.Stopwatch();
            watch.Start();
            bool done = false;

            while (!done)
            {
                done = true;
                foreach (WorkerBee bee in WorkQueue)
                {
                    if (!(done = done && bee.Done))
                        break;
                }
 


                if (watch.ElapsedMilliseconds > timeoutMS)
                    throw new TimeoutException("Timeout elapsed");
 
                if (!done)
                    Thread.Sleep(1); // wait 1ms to try again
            }
 
            watch.Stop();
        }
}


public class DatabaseWorkerBee<RefType, ReturnType> : AnyWorkerBee<Delegate>
{
        public delegate ReturnType DBDelegate();
        private DBDelegate dbDelegate;

        public ReturnType ReturnValue
        {
            get;
            set;
        } 


        public RefType ReferenceObject
        {
            get;
            set;
        } 


        public DatabaseWorkerBee(RefType refObject, DBDelegate del)
        {
            this.ReferenceObject = refObject;
            dbDelegate = del;
        } 


        protected override void WorkStart()
        {
            base.WorkStart();
 
            if (dbDelegate != null)
            {
                ReturnValue = dbDelegate();
            }
        }
    }

Alright, this is now a fairly simple implementation in the UI:

IList<WorkerBee> workers = null;


using (AsyncBlock block = new AsyncBlock(30, Math.Min(nodes.Count, 10))) // do these async with a 30 second timeout
{
      foreach (TreeNode o in nodes)
      {
          TreeNode node = o; //must declare this in the loop, or you get the same object in every worker, this is a documented closure bug in the framework
          block.AddToQueue(
              new DatabaseWorkerBee<TreeNode, int>( node, delegate { /* Do Database Call Here using the node object*/; }));
       }


      workers = block.GetWorkers()


}//using will not return until all workers have been processed.


foreach (WorkerBee worker in workers)
{
      DatabaseWorkerBee<TreeNode, int> dbworker = worker as DatabaseWorkerBee<TreeNode, int>;
      dbWorker.ReferenceObject.Text = String.Format("{0} ({1})", dbWorker.ReferenceObject.Text, dbWorker.ReturnValue); //update the UI
}

Woot! All of the nodes are executed async, and all of the values are returned properly. One thing to note, notice I am declaring a redundant TreeNode object inside the loop. This is necessary, since we are using an anonymous method inside of the loop. If we were to just reference the "o" in the iterator, we would get the same object passed to each anonymous method.

My Threading Objects can be found here: ThreadLib.zip

[Edit]
After further review and testing this code works quite well, except in the instance of windows authenticated database connections. Because we are spawning new threads, and your security context is thread based, all of the subsequent database calls were made using the applications default user. This isn't a problem in Windows UI's but in Web apps this is not good. It could be as simple as setting the new threads to the same security context as the spawning thread, but I'm not sure I want that to happen for every thread. I will work this out and do an update.

I have worked this security context issue out. If anyone is interested in it, email me.

Thread Safe Dictionary in .NET with ReaderWriterLockSlim

Brian Rudolph — Fri, 26 Sep 2008 15:14:00 GMT

Since MS has decided to not include a thread-safe dictionary in the .NET framework, many of us have been forced to implement our own. What MS has done now with .NET 3.5 is given us a newer locking mechanism which makes this task easier.

Previously, we had 2 lock mechanisms, neither of which worked very well for my uses. The first is the Lock statement. While this method is effective, it causes serious contention issues with simultaneous reads\writes. Since it is generally safe to execute multiple concurrent reads from a dictionary, this was a huge performance flaw as the Lock statement was forcing us to perform all actions against the object in a serial fashion.

So that left ReaderWriterLock, which would allow us to take out Reader & Writer locks independently. This seemed to be the answer. Unfortunately, after further testing and review, MS and the dev communities warned against this lock type as it could have unintended consequences in certain situations. The explanations for these I will not go in to, a simple Google search will answer those questions. For the most part, I don't have the mental capacity to understand, much less explain the failures. I will tell you that I have tested this method, and ran in to some issues, and subsequently abandoned the approach.

With .NET 3.5 comes the brand-spankin-new ReaderWriterLockSlim. ReaderWriterLockSlim is similar to the ReaderWriterLock in functionality, minus some of the drawbacks. Of course any new approach will bring on its own consequences, and if you do a little research, you will find that Slim isn't infallible either.

Many of our applications implement the Factory model. Factory models are great, they are extensible, and they can be extremely performant. With that said, any data driven application that uses the Factory model without a caching model, will most likely perform poorly. So, in our factories, we build in a caching mechanism. A very simple type and ID based cache that allows us to persist commonly used objects to in-memory cache for a specific amount of time. I won't go in to the implementation of this model here as every caching implementation has its own needs and wants.

At the heart of this and likely any caching model is inevitably a Dictionary or Dictionary like structure. MS programmers have spent countless hours fine-tuning the Dictionary class to make it as performant as possible. I challenge anyone to write a managed Dictionary that performs as well in such a wide variety of scenarios. I've tried and failed. Unfortunately, MS has not provided a thread-safe implementation. Generally when you run into the question of thread-safety, most people have no idea what you are talking about. The problem with writing truly thread-safe code that can work in any scenario is that you will sacrifice performance. If you tailor your thread-safe mechanisms to your particular problem, you can generally maintain a higher level of performance.

The reason we need thread-safety is not that we are specifically spawning threads. In IIS, worker processes are inherently multi-threaded. Therefore, using static objects in a web-farm generally requires some level of thread safety, otherwise you will get dreaded "An item with the same key has already been added" errors. This is because between the time you check to see if a key already exists in your dictionary and the time you insert it, some other thread may have inserted this key. This is especially the case in object factories, because you have multiple threads attempting to access/create objects and subsequently cache them.

This is the code I've written. Update: This code has been tested, and is working well. I will keep this post updated with any changes and test results. I also have not made this class utilize lock escalation, which ReaderWriterLockSlim supports. As always, please perform your own testing.

using System;
using System.Collections;
using System.Collections.Generic;
using System.Threading;


public interface IThreadSafeDictionary<TKey, TValue> : IDictionary<TKey, TValue>
{
    /// <summary>
    /// Merge is similar to the SQL merge or upsert statement.  
    /// </summary>
    /// <param name="key">Key to lookup</param>
    /// <param name="newValue">New Value</param>
    void MergeSafe(TKey key, TValue newValue);


    /// <summary>
    /// This is a blind remove. Prevents the need to check for existence first.
    /// </summary>
    /// <param name="key">Key to Remove</param>
    void RemoveSafe(TKey key);
}


[Serializable]
public class ThreadSafeDictionary<TKey, TValue> : IThreadSafeDictionary<TKey, TValue>
{
    //This is the internal dictionary that we are wrapping
    IDictionary<TKey, TValue> dict = new Dictionary<TKey, TValue>();


    [NonSerialized]
    ReaderWriterLockSlim dictionaryLock = Locks.GetLockInstance(LockRecursionPolicy.NoRecursion); //setup the lock;


    /// <summary>
    /// This is a blind remove. Prevents the need to check for existence first.
    /// </summary>
    /// <param name="key">Key to remove</param>
    public void RemoveSafe(TKey key)
    {
        using (new ReadLock(this.dictionaryLock))
        {
            if (this.dict.ContainsKey(key))
            {
                using (new WriteLock(this.dictionaryLock))
                {
                    this.dict.Remove(key);
                }
            }
        }
    }


    /// <summary>
    /// Merge does a blind remove, and then add.  Basically a blind Upsert.  
    /// </summary>
    /// <param name="key">Key to lookup</param>
    /// <param name="newValue">New Value</param>
    public void MergeSafe(TKey key, TValue newValue)
    {
        using (new WriteLock(this.dictionaryLock)) // take a writelock immediately since we will always be writing
        {
            if (this.dict.ContainsKey(key))
            {
                this.dict.Remove(key);
            }


            this.dict.Add(key, newValue);
        }
    }


    public virtual bool Remove(TKey key)
    {
        using (new WriteLock(this.dictionaryLock))
        {
            return this.dict.Remove(key);
        }
    }


    public virtual bool ContainsKey(TKey key)
    {
        using (new ReadOnlyLock(this.dictionaryLock))
        {
            return this.dict.ContainsKey(key);
        }
    }


    public virtual bool TryGetValue(TKey key, out TValue value)
    {
        using (new ReadOnlyLock(this.dictionaryLock))
        {
            return this.dict.TryGetValue(key, out value);
        }
    }


    public virtual TValue this[TKey key]
    {
        get
        {
            using (new ReadOnlyLock(this.dictionaryLock))
            {
                return this.dict[key];
            }
        }
        set
        {
            using (new WriteLock(this.dictionaryLock))
            {
                this.dict[key] = value;
            }
        }
    }


    public virtual ICollection<TKey> Keys
    {
        get
        {
            using (new ReadOnlyLock(this.dictionaryLock))
            {
                return new List<TKey>(this.dict.Keys);
            }
        }
    }


    public virtual ICollection<TValue> Values
    {
        get
        {
            using (new ReadOnlyLock(this.dictionaryLock))
            {
                return new List<TValue>(this.dict.Values);
            }
        }
    }


    public virtual void Clear()
    {
        using (new WriteLock(this.dictionaryLock))
        {
            this.dict.Clear();
        }
    }


    public virtual int Count
    {
        get
        {
            using (new ReadOnlyLock(this.dictionaryLock))
            {
                return this.dict.Count;
            }
        }
    }


    public virtual bool Contains(KeyValuePair<TKey, TValue> item)
    {
        using (new ReadOnlyLock(this.dictionaryLock))
        {
            return this.dict.Contains(item);
        }
    }


    public virtual void Add(KeyValuePair<TKey, TValue> item)
    {
        using (new WriteLock(this.dictionaryLock))
        {
            this.dict.Add(item);
        }
    }


    public virtual void Add(TKey key, TValue value)
    {
        using (new WriteLock(this.dictionaryLock))
        {
            this.dict.Add(key, value);
        }
    }


    public virtual bool Remove(KeyValuePair<TKey, TValue> item)
    {
        using (new WriteLock(this.dictionaryLock))
        {
            return this.dict.Remove(item);
        }
    }


    public virtual void CopyTo(KeyValuePair<TKey, TValue>[] array, int arrayIndex)
    {
        using (new ReadOnlyLock(this.dictionaryLock))
        {
            this.dict.CopyTo(array, arrayIndex);
        }
    }


    public virtual bool IsReadOnly
    {
        get
        {
            using (new ReadOnlyLock(this.dictionaryLock))
            {
                return this.dict.IsReadOnly;
            }
        }
    }


    public virtual IEnumerator<KeyValuePair<TKey, TValue>> GetEnumerator()
    {
        throw new NotSupportedException("Cannot enumerate a threadsafe dictionary.  Instead, enumerate the keys or values collection");
    }


    IEnumerator IEnumerable.GetEnumerator()
    {
        throw new NotSupportedException("Cannot enumerate a threadsafe dictionary.  Instead, enumerate the keys or values collection");
    }
}


public static class Locks
{
    public static void GetReadLock(ReaderWriterLockSlim locks)
    {
        bool lockAcquired = false;
        while (!lockAcquired)
            lockAcquired = locks.TryEnterUpgradeableReadLock(1);
    }


    public static void GetReadOnlyLock(ReaderWriterLockSlim locks)
    {
        bool lockAcquired = false;
        while (!lockAcquired)
            lockAcquired = locks.TryEnterReadLock(1);
    }


    public static void GetWriteLock(ReaderWriterLockSlim locks)
    {
        bool lockAcquired = false;
        while (!lockAcquired)
            lockAcquired = locks.TryEnterWriteLock(1);
    }


    public static void ReleaseReadOnlyLock(ReaderWriterLockSlim locks)
    {
        if (locks.IsReadLockHeld)
            locks.ExitReadLock();
    }


    public static void ReleaseReadLock(ReaderWriterLockSlim locks)
    {
        if (locks.IsUpgradeableReadLockHeld)
            locks.ExitUpgradeableReadLock();
    }


    public static void ReleaseWriteLock(ReaderWriterLockSlim locks)
    {
        if (locks.IsWriteLockHeld)
            locks.ExitWriteLock();
    }


    public static void ReleaseLock(ReaderWriterLockSlim locks)
    {
        ReleaseWriteLock(locks);
        ReleaseReadLock(locks);
        ReleaseReadOnlyLock(locks);
    }


    public static ReaderWriterLockSlim GetLockInstance()
    {
        return GetLockInstance(LockRecursionPolicy.SupportsRecursion);
    }


    public static ReaderWriterLockSlim GetLockInstance(LockRecursionPolicy recursionPolicy)
    {
        return new ReaderWriterLockSlim(recursionPolicy);
    }
}


public abstract class BaseLock : IDisposable
{
    protected ReaderWriterLockSlim _Locks;


    public BaseLock(ReaderWriterLockSlim locks)
    {
        _Locks = locks;
    }


    public abstract void Dispose();
}


public class ReadLock : BaseLock
{
    public ReadLock(ReaderWriterLockSlim locks)
        : base(locks)
    {
        Locks.GetReadLock(this._Locks);
    }


    public override void Dispose()
    {
        Locks.ReleaseReadLock(this._Locks);
    }
}


public class ReadOnlyLock : BaseLock
{
    public ReadOnlyLock(ReaderWriterLockSlim locks)
        : base(locks)
    {
        Locks.GetReadOnlyLock(this._Locks);
    }


    public override void Dispose()
    {
        Locks.ReleaseReadOnlyLock(this._Locks);
    }
}


public class WriteLock : BaseLock
{
    public WriteLock(ReaderWriterLockSlim locks)
        : base(locks)
    {
        Locks.GetWriteLock(this._Locks);
    }


    public override void Dispose()
    {
        Locks.ReleaseWriteLock(this._Locks);
    }
}

[Update]
Through some testing, I found the need to update the ThreadSafe Dictionary. I changed all of the locking to use IDisposables to ensure lock creation and destruction.