Tukukkk

Often when reading data from a sequence (IEnumerable<T>) in multi-threaded code, a consistent snapshot needs to be taken inside a lock before using it in a longer operation. I'd like to simplify that by creating an API that does it within the enumerator implementation.

So in the manner of LINQ extension methods, I have created a Synchronized method that takes the source sequence and the lock object and provides synchronised enumerating of the sequence. When enumerating is started, the lock is acquired and held until enumerating is finished by disposing of the enumerator. Since there cannot be a lock statement across multiple methods (MoveNext and Dispose), I'm calling the Monitor.Enter and Monitor.Exit methods instead.

Here's how it looks:

var localList = sharedList   // the shared collection

    .Synchronized(syncObj)   // synchronised iteration (might also use sharedList)

    .ToList();               // take the snapshot (or ToDictionary or ToWhatever)

I've also written a small test program to see if accessing a shared list works across multiple threads. I'm not too sure this effectively tests it but it can be made to work and fail (partially) by using or not using my new method. I've never managed to get a corrupted List or LinkedList though while not using locks. Maybe I need a more complex/fragile collection structure.

Now what I'm interested in is to know whether this kind of using locks is reliable (no unsynchronised access and no deadlocks), "good style", and as fast as it can be. (In this order.) Are there any obscure scenarios when this method would fail? I'm looking to use it everywhere it fits, without paying too much attention on use case restrictions.

First part of the code is my extension method and the private IEnumerable/IEnumerator class it uses.

EnumerableExtensions.cs

using System;

using System.Collections;

using System.Collections.Generic;

using System.Threading;



namespace LinqLockedTest

{

    public static class EnumerableExtensions

    {

        /// <summary>

        /// Acquires an exclusive lock on the specified object before iterating the sequence and

        /// releases the lock when disposing of the <see cref="IEnumerator{T}"/>.

        /// </summary>

        /// <typeparam name="TSource">The type of objects to enumerate.</typeparam>

        /// <param name="source">The <see cref="IEnumerable{T}"/> to synchronize.</param>

        /// <param name="syncObject">The object on which to acquire the monitor lock.</param>

        /// <returns>An <see cref="IEnumerable{T}"/> that provides a synchronized enumerator.</returns>

        /// <exception cref="ArgumentNullException"><paramref name="source"/> or <paramref name="syncObject"/> is null.</exception>

        public static IEnumerable<TSource> Synchronized<TSource>(this IEnumerable<TSource> source, object syncObject)

        {

            if (source == null)

                throw new ArgumentNullException(nameof(source));

            if (syncObject == null)

                throw new ArgumentNullException(nameof(syncObject));



            return new SynchronizedIterator<TSource>(source, syncObject);

        }



        private class SynchronizedIterator<TSource> : IEnumerable<TSource>, IEnumerator<TSource>

        {

            private readonly IEnumerable<TSource> source;

            private readonly object syncObject;



            private IEnumerator<TSource> sourceEnumerator;

            private bool lockTaken;



            public SynchronizedIterator(IEnumerable<TSource> source, object syncObject)

            {

                this.source = source;

                this.syncObject = syncObject;

            }



            #region IEnumerable<TSource> members



            public IEnumerator<TSource> GetEnumerator() => this;



            IEnumerator IEnumerable.GetEnumerator() => GetEnumerator();



            #endregion IEnumerable<TSource> members



            #region IEnumerator<TSource> members



            public TSource Current => sourceEnumerator.Current;



            object IEnumerator.Current => Current;



            public void Dispose()

            {

                if (sourceEnumerator != null)

                {

                    sourceEnumerator.Dispose();

                    sourceEnumerator = null;

                    if (lockTaken)

                    {

                        Monitor.Exit(syncObject);

                    }

                }

            }



            public bool MoveNext()

            {

                if (sourceEnumerator == null)

                {

                    Monitor.Enter(syncObject, ref lockTaken);

                    sourceEnumerator = source.GetEnumerator();

                }

                if (sourceEnumerator.MoveNext())

                {

                    return true;

                }

                Dispose();

                return false;

            }



            public void Reset() => sourceEnumerator?.Reset();



            #endregion IEnumerator<TSource> members

        }

    }

}

Second part is the test program. I've made a .NET Core console project in Visual Studio 2017 to write it. The entire solution should be compatible with .NET Core 2.1 and .NET Standard 2.0.

The interesting call of my extension method is at the beginning of the ReadList method. Alternative traditional code is also shown for reference.

The test can be made to fail by not calling the Synchronized method but using the list directly in foreach.

When you run this code, it starts three threads: modifying the list, reading the list, and showing log messages. They run forever doing their thing. Press any key to quit. Press Pause (Break in English?) then Enter to have a break and inspect the output. I guess it would be good to have at least 3 CPU cores available for this.

Program.cs

using System;

using System.Collections.Concurrent;

using System.Collections.Generic;

using System.Linq;

using System.Threading;

using System.Threading.Tasks;



namespace LinqLockedTest

{

    internal class Program

    {

        private static List<int> list = new List<int>();

        private static ConcurrentQueue<string> log = new ConcurrentQueue<string>();



        private static void Main(string args)

        {

            Task.Run(() => ModifyList());

            Task.Run(() => ReadList());

            Task.Run(() => ShowLog());



            log.Enqueue("Test is running. Press any key to quit.n");

            Console.ReadKey(true);

        }



        /// <summary>

        /// Modifies the shared list by adding sequential numbers at the end and then

        /// removing them from the start.

        /// </summary>

        private static void ModifyList()

        {

            while (true)

            {

                log.Enqueue("===== Adding items =====");

                for (int i = 0; i < 100000; i++)

                {

                    lock (list)

                    {

                        list.Add(i);

                    }

                    // Adding is much faster than removing, so wait a little longer here

                    Thread.Yield();

                }

                log.Enqueue("===== Removing items =====");

                for (int i = 0; i < 100000; i++)

                {

                    lock (list)

                    {

                        list.RemoveAt(0);

                    }

                }

            }

        }



        /// <summary>

        /// Reads from the shared list, checks the result for consistency and writes

        /// occasional log messages about the progress.

        /// </summary>

        private static void ReadList()

        {

            int iter = 0;

            while (true)

            {

                try

                {

                    // This is the code to be tested:

                    // (Could also be used directly in foreach but then locks for the

                    // entire foreach loop instead of just while taking the snapshot

                    // with ToList.)

                    var localList = list.Synchronized(list).ToList();



                    // The equivalent traditional code:

                    //List<int> localList;

                    //lock (list)

                    //{

                    //  localList = list.ToList();

                    //}



                    int prev = -1;

                    int count = 0;

                    foreach (int i in localList)

                    {

                        if (prev != -1)

                        {

                            if (i <= prev)

                                log.Enqueue($"Error: Found {i} after {prev}");

                        }

                        prev = i;

                        count++;

                    }

                    if (iter % 50 == 0)

                        log.Enqueue($"{count} items up to {prev}");

                }

                catch (Exception ex)

                {

                    log.Enqueue($"Error: {ex.Message}");

                }

                iter++;

            }

        }



        /// <summary>

        /// Shows the log messages in the console window. This is a separate task because

        /// the queue blocks the enqueuing thread much shorter than console output.

        /// </summary>

        private static void ShowLog()

        {

            while (true)

            {

                if (log.TryDequeue(out string message))

                {

                    Console.WriteLine(message);

                }

            }

        }

    }

}

Your code is clear and easy to follow. Stylistically, I'd say don't use regions but that's personal preference.

That said, you do have a bug:

int arr = new {1,2,3};

object locker = new object();

var enumerator = arr.Synchronized(locker);

enumerator.ToList();

enumerator.ToList(); // BOOM

You should be able to enumerate the same enumerable multiple times - it shouldn't result in a runtime exception.

Writing a test program to check your code works as you expect is good but I would get in the habit of writing unit tests instead. They are just as easy to run and have the added benefit of also listing your expected behaviours.

Now, as for the actual idea behind this code, I'd suggest you take a look at the concurrent collections available in System.Collections.Concurrent and use them instead. You have used ConcurrentQueue so I know you know about them ;) Their approach to GetEnumerator is to return a snapshot (copy) of the collection at that time.

At the moment, you require the writers and readers to share the same synchronisation object which might not always be feasible. The other downside is that all other readers and writers are blocked until the reader is finished (and they dispose the enumerator correctly). That's too much to assume in my opinion. It's better to use a collection that is designed for the job than to try synchronise over the top of another collection externally.

I don't think this is a good idea. A well-designed API should be easy to use correctly, and difficult to use incorrectly. Let's look at a few ways in which this method could be used incorrectly:

Holding locks briefly:

foreach (var item in items.Synchronized(items))

{

    // do (expensive) work here

}

Locks should be kept only as long as necessary, but this holds a lock until all work has been done. You know this is not how Synchronized is meant to be used, but that's not all that obvious to Marvin the Maintenance programmer and Joe the Junior dev.

Verifying lock order:

var syncedItems = items.Synchronized(items);    // Is the lock obtained here...

var array = syncedItems.ToArray();              // ...or here?

With the more simple lock variant, it's obvious where and how long a lock is being held. With the above code, it's not so easy. And let's say Marvin needs to add another lock, which - if both locks are needed - must only be obtained after items:

var syncedItems = items.Synchronized(items);

lock (somethingElse)

{

    var array = syncedItems.ToArray();

}

Synchronized is called outside (and before) the somethingElse lock, so this appears to be safe, but it is not. Concurrency is difficult enough, and the more complex your code is the more difficult it is to verify the correctness of your code.

Multiple iterations:

RobH already pointed this out:

var syncedItems = items.Synchronized(items);

if (syncedItems.Any())                  // A lock is obtained here...

{

    var array = syncedItems.ToArray();  // ...and an attempt is made here as well.

}

This fails with an ArgumentException. Well, at least it fails, instead of obtaining a lock multiple times and potentially providing a different snapshot each time...

Interaction with broken enumeration:

IEnumerable<T> CustomIteration(IEnumerable<T> items)

{

    var e = items.GetEnumerator();

    while (e.MoveNext())

        yield e.Current;

}

This code is broken - it doesn't dispose the enumerator. But that often doesn't cause problems, so it's something Joe could have created after reading a poorly written blog post about enumerators. This will cause problems when combined with Synchronized: CustomIteration(items.Synchronized(items)) will never release its lock.

Safe use-case:

About the only safe use-case, as far as I can tell, is directly taking a snapshot:

var snapshot = items.Synchronized(items).ToArray();

It seems to me that a method that does just that - and only that - will be safer:

T snapshot = items.GetSnapshotWithLock(items);

If you need a dictionary then you can safely add a ToDictionary call afterwards, or create a dictionary variant of this method.

Or just use a lock directly. More verbose, but less susceptible to the problems shown above.

You should set lockTaken = false in Dispose()

To elaborate a little on Pieter Witvoets CustomIteration(..):

Doing this:

  foreach (var item in CustomIteration(data.Synchronized(lockObj)))

  {

    Console.WriteLine(item);

  }

works, because MoveNext() calls Dispose() when called after the last item (sourceEnumerator.MoveNext() returns false).

But doing this:

  foreach (var item in CustomIteration(data.Synchronized(lockObj)).Take(5))

  {

    Console.WriteLine(item);

  }

is problematic because then Dispose() is never called from MoveNext() (because foreach only sees the outer IEnumerable<T> which hides the inner Enumerator. So Monitor.Exit(...) is never called.

So in the following scenario the ThreadPool thread is in trouble and will hang - because of the unreleased lock:

  int data = { 1, 2, 3, 4, 5, 6, 7, 8, 9 };

  object lockObj = new object();



  foreach (var item in CustomIteration(data.Synchronized(lockObj)).Take(5))

  {

    Console.WriteLine(item);

  }



  AutoResetEvent are1 = new AutoResetEvent(false);



  ThreadPool.QueueUserWorkItem((state) =>

  {

    // Here this thread will hang

    foreach (var item in data.Synchronized(lockObj))

    {

      Console.WriteLine(item);

    }



    are1.Set();

  });





  are1.WaitOne();