Are Ienumerable Linq Methods Thread-Safe

Are IEnumerable Linq methods thread-safe?

The interface IEnumerable<T> is not thread safe. See the documentation on http://msdn.microsoft.com/en-us/library/s793z9y2.aspx, which states:

An enumerator remains valid as long as the collection remains unchanged. If changes are made to the collection, such as adding, modifying, or deleting elements, the enumerator is irrecoverably invalidated and its behavior is undefined.

The enumerator does not have exclusive access to the collection; therefore, enumerating through a collection is intrinsically not a thread-safe procedure. To guarantee thread safety during enumeration, you can lock the collection during the entire enumeration. To allow the collection to be accessed by multiple threads for reading and writing, you must implement your own synchronization.

Linq does not change any of this.

Locking can obviously be used to synchronize access to objects. You must lock the object everywhere you access it though, not just when iterating over it.

Declaring the collection as volatile will have no positive effect. It only results in a memory barrier before a read and after a write of the reference to the collection. It does not synchronize collection reading or writing.

Is this LINQ dynamic orderby method thread safe?

The answer is simple:

It is thread safe if your original collection is thread safe.

All extension methods in LINQ merely call .GetEnumerator() of the original collection. Sorting and Ordering doesn't manipulate the original collection, but rather lets you enumerate it in a sorted order. Thus, you only do read operations on the data. As a general rule of thumb, if you only do read data, you do not need to implement any thread safety.

I'm tempted to say that in 99% of cases you don't need any thread safety, because you collect data only once and then expose the LINQ functionality. You might only need a thread safe collection if you want to create a framework that does not instantiate new collections when refreshing data, but rather keeps re-using the same (observable) collection instance that synchronizes itself with the database data in a fancy way.

But I don't know about your exact scenario. If you really do need thread safety, then it depends if you have control over the code where the original collection is instantiated and the data is added.

If you are using merely LINQ-to-objects, then it's entirely in your control to create an instance of a thread safe collection class at that spot.

If you're using LINQ-to-SQL or anything, then it might be difficult because the original collection that collects the data from the database is probably instantiated deep within the provider and hidden from you. I haven't looked at it though if there are extension points where you can override stuff to use a thread safe collection instead.

C#: How can I make an IEnumerableT thread safe?

There's an inherent problem in doing so, because IEnumerator<T> has both MoveNext() and Current. You really want a single call such as:

bool TryMoveNext(out T value)

at that point you can atomically move to the next element and get a value. Implementing that and still being able to use yield could be tricky... I'll have a think about it though. I think you'd need to wrap the "non-threadsafe" iterator in a thread-safe one which atomically performed MoveNext() and Current to implement the interface shown above. I don't know how you'd then wrap this interface back into IEnumerator<T> so that you could use it in foreach though...

If you're using .NET 4.0, Parallel Extensions may be able to help you - you'd need to explain more about what you're trying to do though.

This is an interesting topic - I may have to blog about it...

EDIT: I've now blogged about it with two approaches.

How to map ImmutableArray without getting it cast to IEnumerable which is not thread safe?

It seems that there are a lot of misunderstandings behind this question. You need to go look at the source code for the Select method and learn about the yield keyword.

Second, LINQ methods are made to be short-lived. You have various threads doing various processing tasks. Are you using a pipeline situation, where you want to transform data in one thread and pass the result to another thread? You have to be careful with the yield keyword in that situation; essentially, you need to flush (er, realize, for lack of a better word) your collections before passing them to the next thread so that the actual work is done in the present thread. In that scenario, object ownership kicks in and you don't need thread-safe collections.

In short, the enumerable returned from calling Select on ImmutableArray is perfectly thread-safe. You can realize it at any point and it won't give you any errors. Of course it will only iterate through the data that was contained in your collection at the time you called Select. It won't know anything about newly assigned instances.

Thread safe with Linq and Tasks on a Collection

Your code is the right way to do this, assuming starting DoABunchOfWorkElsewhere() multiple times is itself safe.

You don't need to worry about your LINQ query, because it doesn't actually run in parallel. All it does is to invoke DoWorkOnItemInCollection() multiple times. Those invocations may work in parallel (or not, depending on your synchronization context and the implementation of DoABunchOfWorkElsewhere()), but the code you showed is safe.

Extension methods on System.Collections.Concurrent collections are thread-safe?

The extension methods don't add or remove any thread-safety. But you can't really ignore that iterating a collection is never thread-safe, whether you do it explicitly or let Linq do it. You must ensure that no other thread can modify the collection at the same time.

Beware that this is extra tricky in Linq because of its delayed execution behavior.

Thread-safe Cached Enumerator - lock with yield

Your class is not thread safe, because shared state is mutated in unprotected regions inside your class. The unprotected regions are:

1. The constructor
2. The Dispose method

The shared state is:

1. The _enumerator private field
2. The _cache private field
3. The CachingComplete public property

Some other issues regarding your class:

1. Implementing IDisposable creates the responsibility to the caller to dispose your class. There is no need for IEnumerables to be disposable. In the contrary IEnumerators are disposable, but there is language support for their automatic disposal (feature of foreach statement).
2. Your class offers extended functionality not expected from an IEnumerable (ElementAt, Count etc). Maybe you intended to implement a CachedList instead? Without implementing the IList<T> interface, LINQ methods like Count() and ToArray() cannot take advantage of your extended functionality, and will use the slow path like they do with plain vanilla IEnumerables.

---

Update: I just noticed another thread-safety issue. This one is related to the public IEnumerator<T> GetEnumerator() method. The enumerator is compiler-generated, since the method is an iterator (utilizes yield return). Compiler-generated enumerators are not thread safe. Consider this code for example:

var enumerable = Enumerable.Range(0, 1_000_000);
var cachedEnumerable = new CachedEnumerable<int>(enumerable);
var enumerator = cachedEnumerable.GetEnumerator();
var tasks = Enumerable.Range(1, 4).Select(id => Task.Run(() =>
{
    int count = 0;
    while (enumerator.MoveNext())
    {
        count++;
    }
    Console.WriteLine($"Task #{id} count: {count}");
})).ToArray();
Task.WaitAll(tasks);

Four threads are using concurrently the same IEnumerator. The enumerable has 1,000,000 items. You may expect that each thread would enumerate ~250,000 items, but that's not what happens.

Output:

Task #1 count: 0

Task #4 count: 0

Task #3 count: 0

Task #2 count: 1000000

The MoveNext in the line while (enumerator.MoveNext()) is not your safe MoveNext. It is the compiler-generated unsafe MoveNext. Although unsafe, it includes a mechanism intended probably for dealing with exceptions, that marks temporarily the enumerator as finished before calling the externally provided code. So when multiple threads are calling the MoveNext concurrently, all but the first will get a return value of false, and will terminate instantly the enumeration, having completed zero loops. To solve this you must probably code your own IEnumerator class.

---

Update: Actually my last point about thread-safe enumeration is a bit unfair, because enumerating with the IEnumerator interface is an inherently unsafe operation, which is impossible to fix without the cooperation of the calling code. This is because obtaining the next element is not an atomic operation, since it involves two steps (call MoveNext() + read Current). So your thread-safety concerns are limited to the protection of the internal state of your class (fields _enumerator, _cache and CachingComplete). These are left unprotected only in the constructor and in the Dispose method, but I suppose that the normal use of your class may not follow code paths that create the race conditions that would result to internal state corruption.

Personally I would prefer to take care of these code paths too, and I wouldn't let it to the whims of chance.

---

Update: I wrote a cache for IAsyncEnumerables, to demonstrate an alternative technique. The enumeration of the source IAsyncEnumerable is not driven by the callers, using locks or semaphores to obtain exclusive access, but by a separate worker-task. The first caller starts the worker-task. Each caller at first yields all items that are already cached, and then awaits for more items, or for a notification that there are no more items. As notification mechanism I used a TaskCompletionSource<bool>. A lock is still used to ensure that all access to shared resources is synchronized.

public class CachedAsyncEnumerable<T> : IAsyncEnumerable<T>
{
    private readonly object _locker = new object();
    private IAsyncEnumerable<T> _source;
    private Task _sourceEnumerationTask;
    private List<T> _buffer;
    private TaskCompletionSource<bool> _moveNextTCS;
    private Exception _sourceEnumerationException;
    private int _sourceEnumerationVersion; // Incremented on exception

    public CachedAsyncEnumerable(IAsyncEnumerable<T> source)
    {
        _source = source ?? throw new ArgumentNullException(nameof(source));
    }

    public async IAsyncEnumerator<T> GetAsyncEnumerator(
        CancellationToken cancellationToken = default)
    {
        lock (_locker)
        {
            if (_sourceEnumerationTask == null)
            {
                _buffer = new List<T>();
                _moveNextTCS = new TaskCompletionSource<bool>();
                _sourceEnumerationTask = Task.Run(
                    () => EnumerateSourceAsync(cancellationToken));
            }
        }
        int index = 0;
        int localVersion = -1;
        while (true)
        {
            T current = default;
            Task<bool> moveNextTask = null;
            lock (_locker)
            {
                if (localVersion == -1)
                {
                    localVersion = _sourceEnumerationVersion;
                }
                else if (_sourceEnumerationVersion != localVersion)
                {
                    ExceptionDispatchInfo
                        .Capture(_sourceEnumerationException).Throw();
                }
                if (index < _buffer.Count)
                {
                    current = _buffer[index];
                    index++;
                }
                else
                {
                    moveNextTask = _moveNextTCS.Task;
                }
            }
            if (moveNextTask == null)
            {
                yield return current;
                continue;
            }
            var moved = await moveNextTask;
            if (!moved) yield break;
            lock (_locker)
            {
                current = _buffer[index];
                index++;
            }
            yield return current;
        }
    }

    private async Task EnumerateSourceAsync(CancellationToken cancellationToken)
    {
        TaskCompletionSource<bool> localMoveNextTCS;
        try
        {
            await foreach (var item in _source.WithCancellation(cancellationToken))
            {
                lock (_locker)
                {
                    _buffer.Add(item);
                    localMoveNextTCS = _moveNextTCS;
                    _moveNextTCS = new TaskCompletionSource<bool>();
                }
                localMoveNextTCS.SetResult(true);
            }
            lock (_locker)
            {
                localMoveNextTCS = _moveNextTCS;
                _buffer.TrimExcess();
                _source = null;
            }
            localMoveNextTCS.SetResult(false);
        }
        catch (Exception ex)
        {
            lock (_locker)
            {
                localMoveNextTCS = _moveNextTCS;
                _sourceEnumerationException = ex;
                _sourceEnumerationVersion++;
                _sourceEnumerationTask = null;
            }
            localMoveNextTCS.SetException(ex);
        }
    }
}

This implementation follows a specific strategy for dealing with exceptions. If an exception occurs while enumerating the source IAsyncEnumerable, the exception will be propagated to all current callers, the currently used IAsyncEnumerator will be discarded, and the incomplete cached data will be discarded too. A new worker-task may start again later, when the next enumeration request is received.

.ToList() thread-safe

is it safe when using this approach when dirty reads are ok?

Easy answer - no, it's not.

First, the method does not guarantee that by definition. Even if you look at the current implementation (which you should not do), you'll see that it's using List<T> constructor accepting the IEnumerable<T>. Then it checks for ICollection<T> and if yes, uses CopyTo method or just iterates the enumerable. Both are unsafe, because the first one relies on the (unknown) implementation of CopyTo method while the second will receive exception when (the standard behavior) the collection enumerator is invalidated during the Add. And even if the source is a List<T> and is using the Array method you mentioned http://referencesource.microsoft.com/#mscorlib/system/collections/generic/list.cs,d2ac2c19c9cf1d44

Array.Copy(_items, 0, array, arrayIndex, _size);

still is unsafe because it may call copy with _items and _size beyond the _items.Length which of course will throw exception. Your assumption would have been correct only if this code is somehow getting both members atomically, but it doesn't.

So, simple don't do that.

EDIT: All above applies to your concrete question. But I think there is a simple solution for the situation you have explained. If single thread add/multiple thread read with acceptable stale reads is the requirement, then it could be achieved by using the ImmutableList<T> class from the System.Collections.Immutable package like this:

public class Container
{
    private ImmutableList<Item> _items = ImmutableList<Item>.Empty;
    public IReadOnlyList<Item> Items { get { return _items; } }
    public void Add(Item item) { _items = _items.Add(item); }
}

---

Related Topics