Using Parallel Extension Extras with SqlConnections

Introduction

The Parallel Extensions Extras (PEE) library, readily available from NuGet, brings a number of useful classes, which greatly simplify parallel information processing. I hope for this tip to be first of many to come, highlighting the simplicity of parallel programming, using Task Parallel Library (TPL) and PEE.

Background

Parallel code execution is one of the challenges, we as developers and architects, have to deal with on a daily basis. The introduction of TPL was a much welcomed addition to the .NET Framework. No longer did I have to rely on handwritten background thread processor with many synchronization scenarios. TaskFactory, Parallel and PLINQ made most of my old code obsolete. Recently, I re-discovered the Parallel Extension Extras, to which, sadly, I did not pay enough attention previously.

ObjectPool<T>

An object pool is a mechanism/pattern to avoid the repeated creation and destruction of objects. When code is done with an object, rather than allowing it to be garbage collected (and finalized if it’s finalizable), you put the object back into a special collection known as an object pool. Then, when you need an object, rather than always creating one, you ask the pool for one: if it has one, it gives it to you, otherwise it creates one and gives it to you. In many situations where creation and destruction is expensive, and where many objects are needed but where only a few at a time are needed, this can result in significant performance gains.

Object pools are just as relevant in multi-threaded scenarios as they are in single-threaded scenarios, but of course when dealing with multiple threads, you need to synchronize correctly (unless a separate pool is maintained per thread, in which case you’re trading synchronization cost for potentially creating more objects than you otherwise would). ParallelExtensionsExtras contains a simple ObjectPool<T> implementation in the ObjectPool.cs file, built on top of IProducerConsumerCollection<T>.

(a quote from Parallel Programming with .NET Blog)

Now, the example below is quite simple, but demonstrates a concept, how to safely use objects that might not be otherwise thread safe, in this example, the SqlConnection class.

First start by creating a Console Application solution, than install the Parallel Extensions Extras package from NuGet.

Install-Package ParallelExtensionsExtras  

Then copy and paste this code into your Program.cs file.

//
namespace ConsoleApplication1
{
    using System;
    using System.Collections.Concurrent;
    using System.Collections.Generic;
    using System.Data;
    using System.Data.SqlClient;
    using System.Diagnostics;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;

    internal class Program
    {
        #region Static Fields

        private static readonly ConcurrentDictionary<long, DateTime> DatesRetrieved = new ConcurrentDictionary<long, DateTime>();

        private static readonly ObjectPool<SqlConnection> ObjectPool = new ObjectPool<SqlConnection>(
            delegate
            {
                Interlocked.Increment(ref numberOfInstances);
                return new SqlConnection(SqlConnectionStringBuilder.ConnectionString);
            });

        private static readonly SqlConnectionStringBuilder SqlConnectionStringBuilder = new SqlConnectionStringBuilder
                                                                                        {
                                                                                            DataSource = "server",
                                                                                            UserID = "user",
                                                                                            Password = "password",
                                                                                            IntegratedSecurity = false,
                                                                                            InitialCatalog = "database"
                                                                                        };

        private static volatile Int32 numberOfInstances;

        #endregion

        #region Methods

        private static IEnumerable<IDataRecord> GetRows(string sql)
        {
            SqlConnection connection = ObjectPool.GetObject();
            using (var command = new SqlCommand(sql, connection))
            {
                if (connection.State == ConnectionState.Closed)
                {
                    connection.Open();
                }
                using (SqlDataReader reader = command.ExecuteReader())
                {
                    foreach (IDataRecord record in reader)
                    {
                        yield return record;
                    }
                }
            }
            ObjectPool.PutObject(connection);
        }

        private static void Main(string[] args)
        {
            var r = new Random();
            var sw = new Stopwatch();
            Action timeStamp = () => Console.WriteLine("{0}ms elapsed", sw.ElapsedMilliseconds);
            sw.Start();
            List<long> allPolicies = GetRows("SELECT ID FROM RECORDS").Select(s => Convert.ToInt64(s[0])).ToList();
            timeStamp.Invoke();
            Parallel.For(
                (long)0,
                1000,
                i =>
                {
                    IDataRecord row = GetRows("SELECT * FROM RECORDS WHERE ID=" + allPolicies[r.Next(0, allPolicies.Count)]).Single();
                    long id = Convert.ToInt64(row["ID"]);
                    DateTime fm = Convert.ToDateTime(row["DATE_MODIFIED"]);
                    DatesRetrieved[id] = fm;
                });
            Console.WriteLine("A total of {0} sql connection objects was created", numberOfInstances);
            Console.WriteLine("A total of {0} unique records was retrieved (randomly)", DatesRetrieved.Count);
            timeStamp.Invoke();
        }

        #endregion
    }
}
// 

How Does It Work?

First, we create some helper objects, to track how many instances of SqlConnection have we actually created and how many unique rows have been retrieved for the 1000 iterations. This is just to demonstrate the concept of some data processing.

Next, we create an ObjectPool<SqlConnection> and provide it with a generator function, which will create a new SqlConnection object every time one is not already available. This is where the simplicity of ObjectPool<T> really shines. Initially, there will be no objects. As the pool gets hit from multiple threads in GetRows, new objects will be created, and at the end of GetRows, put back into the pool. Next time around, thanks to the internal implementation of ProducerConsumer pattern, the object will be removed from the pool, for each thread which requires it. Feel free to experiment with commenting out the ObjectPull.Put() method call, which will demonstrate that for each call a new SqlConnection will be created. Additionally, you could enclose the method's body in a try/catch block, and only return pooled object back to the pool, if it is not in a broken state. This becomes especially important, when you start using ObjectPool<T> with WCF services, which client might go into a faulted state for many reasons.

Once you execute the above program, you will get results similar to mine:

690ms elapsed 
A total of 9 sql connection objects was created
A total of 999 unique records was retrieved (randomly)
1570ms elapsed     

One more thing worth mentioning ... When you execute any method, which returns an IEnumerable<T>, you must pay close attention to how you retrieve your data. For example, if you have moved the LINQ Single() method call to each of the variable assignments, the database connection would be created or retrieved twice as many times, because the "header" of the GetRow method, before getting to yield return statement, would have to be executed for each of the two separate calls.

History

• 10/1/2013 - Initial version posted