darkseal/lockprovider

A lightweight C# class that can be used to selectively lock objects, resources or statement blocks according to given unique IDs.

Books

• ASP.NET Core and Angular 2 (Packt Publishing)
• ASP.NET Core 2 and Angular 5 (Packt Publishing)
• ASP.NET Core 3 and Angular 9 (Packt Publishing)
• ASP.NET Core 5 and Angular (Packt Publishing)
• ASP.NET Core 6 and Angular (Packt Publishing)
• ASP.NET Core 8 and Angular (Packt Publishing)
• • Building Web APIs with ASP.NET Core (Manning Publications)

Contacts

• LinkedIn
• Twitter

lockprovider's Issues

Atomicity in the Release method

I think you have an atomicity issue in the release method.

        public void Release(T idToUnlock)
        {
            InnerSemaphore semaphore;
            if (lockDictionary.TryGetValue(idToUnlock, out semaphore))
            {
                semaphore.Release();
                if (!semaphore.HasWaiters && lockDictionary.TryRemove(idToUnlock, out semaphore))
                    semaphore.Dispose();
            }
        }

There is a chance that between checking !semaphore.HasWaiters and removing the dictionary entry, that another thread could call Wait(). You would know this after the removal if the semaphore suddenly has waiters. At that point you could try to add the semaphore back to the dictionary, but there's a chance that yet another thread has called Wait() and created a new dictionary entry.

Simplifying

Hi, I would like to propose a simplified version of the LockProvider

    /// <summary>
    /// A LockProvider based upon the SemaphoreSlim class 
    /// to selectively lock objects, resources or statement blocks 
    /// according to given unique IDs in a sync or async way.
    /// </summary>
    public sealed class LockProvider<T>
    {
        private readonly ConcurrentDictionary<T, SemaphoreSlim> lockDictionary =
            new ConcurrentDictionary<T, SemaphoreSlim>();

        public LockProvider() { }

        /// <summary>
        /// Blocks the current thread (according to the given ID)
        /// until it can enter the LockProvider
        /// </summary>
        /// <param name="idToLock">the unique ID to perform the lock</param>
        public void Wait(T idToLock)
        {
            lockDictionary.GetOrAdd(idToLock, new SemaphoreSlim(1, 1)).Wait();
        }

        /// <summary>
        /// Asynchronously puts thread to wait (according to the given ID)
        /// until it can enter the LockProvider
        /// </summary>
        /// <param name="idToLock">the unique ID to perform the lock</param>
        public async Task WaitAsync(T idToLock)
        {
            await lockDictionary.GetOrAdd(idToLock, new SemaphoreSlim(1, 1)).WaitAsync();
        }

        /// <summary>
        /// Releases the lock (according to the given ID)
        /// </summary>
        /// <param name="idToUnlock">the unique ID to unlock</param>
        public void Release(T idToUnlock)
        {
            if (lockDictionary.TryGetValue(idToUnlock, out SemaphoreSlim semaphore)) 
            {
                semaphore.Release();
                if (semaphore.CurrentCount == 1) lockDictionary.TryRemove(idToUnlock, out semaphore);
            }
        }
    }

Multiple accesses into critical section

Am I doing something wrong?

Should the Interlocked class be used?

I was learning more about threads last week, and found out that the Interlocked class is the thread safe way for increment and decrement.

So instead of _waiters++; it should be

Interlocked.Increment(ref _waiters);

and instead of _waiters—; it should be

Interlocked.Decrement(ref _waiters);

Have you had any problems with the _waiters variable in your tests?

Race conditions

I may be mistaken here, but I wrote a similar library myself and know a thing or two how things need to be done within the context of locking based on keys. I believe I see multiple race conditions in this code. The reference counters are not incremented and decremented within locks (#2 mentions it already, but the suggestion is still not enough).

So what could happen?

First of all, two concurrent calls for the same key could both try to do a _waiters++ at the same time, which could result in the value being incremented only by 1 instead of 2. Should this happen, it will be removed from the dictionary whilst it's still in use, and a new thread for the same key could now enter and concurrently do stuff for the same key, which is what this library is trying to avoid.

Secondly, if thread A decrements _waiters from 1 to 0 and starts doing a tryremove, a thread B can get it from the dictionary and increment that 0 to a 1, and is blissfully unaware that the key is being removed by thread A. So thread A removes the item from the dictionary and thread B is still processing for a key that is not in the dictionary anymore. Once again, a thread C can enter, not find anything in the dictionary for the same key as B, and parallel processes for the same key, again against the expectation from this library.

My attempt at resolving issue (4)

Darkseal,

This is my attempt at fixing issue (4) but keeping the spirit of the code you wrote by using 'lock (locker)'. As a more experienced developer can you check that what I have done makes sense and you are free to use this solution.

Also can you close issue (4), which seems to concluded.

Note: A senior dev found a problem with my code which I have now corrected, it should now be correct.

Thanks.

Here is the code:

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

namespace TestLockProvider
{
    public class LockProvider<T> where T : notnull
    {
        private readonly object locker = new();
        readonly LazyConcurrentDictionary<T, InnerSemaphore> LockDictionary = new();        
        readonly InnerSemaphore _innerSemaphore = new(1, 1);

        public LockProvider() { }

        /// <summary>
        /// Use lock to make it exclusive
        /// </summary>
        /// <param name="idToLock">the unique ID to perform the lock</param>
        /// <returns></returns>
        private InnerSemaphore GetOrAddExclusive(T idToLock)
        {
            InnerSemaphore semaphore;
            if (!LockDictionary.TryGetValue(idToLock, out semaphore!))
            {
                semaphore = _innerSemaphore;

                lock (locker)
                {
                    LockDictionary.GetOrAdd(idToLock, semaphore);
                }
            }

            return semaphore;
        }

        /// <summary>
        /// Blocks the current thread (according to the given ID) until it can enter the LockProvider
        /// </summary>
        /// <param name="idToLock">the unique ID to perform the lock</param>
        public void Wait(T idToLock)
        {
            InnerSemaphore semaphore;
            semaphore = GetOrAddExclusive(idToLock);
            semaphore.Wait();
        }

        /// <summary>
        /// Asynchronously puts thread to wait (according to the given ID) until it can enter the LockProvider
        /// </summary>
        /// <param name="idToLock">the unique ID to perform the lock</param>
        public async Task WaitAsync(T idToLock, CancellationToken cancellationToken = default)
        {
            InnerSemaphore semaphore;
            semaphore = GetOrAddExclusive(idToLock);
            await semaphore.WaitAsync(cancellationToken).ConfigureAwait(false);

        }

        public void Release(T idToUnlock)
        {
            if (LockDictionary.TryGetValue(idToUnlock, out var semaphore))
            {
                lock (locker)
                {
                    semaphore!.Release();
                    if (!semaphore.HasWaiters && LockDictionary.TryRemove(idToUnlock, out semaphore!))
                        semaphore!.Dispose();
                }
            }
        }
    }

    public class InnerSemaphore : IDisposable
    {
        private readonly SemaphoreSlim _semaphore;
        private int _waiters;

        public InnerSemaphore(int initialCount, int maxCount)
        {
            _semaphore = new SemaphoreSlim(initialCount, maxCount);
            _waiters = 0;
        }

        public void Wait()
        {
            Interlocked.Increment(ref _waiters);
            _semaphore.Wait();
        }

        public async Task WaitAsync(CancellationToken cancellationToken = default)
        {
            Interlocked.Increment(ref _waiters);
            await _semaphore.WaitAsync(cancellationToken);
        }

        public void Release()
        {
            Interlocked.Decrement(ref _waiters);
            _semaphore.Release();
        }

        public void Dispose()
        {
            _semaphore?.Dispose();
            GC.SuppressFinalize(this);
        }
        public bool HasWaiters => _waiters > 0;
    }

    public class LazyConcurrentDictionary<TKey, TValue> where TKey : notnull
    {
        private readonly ConcurrentDictionary<TKey, Lazy<TValue>> _concurrentDictionary;

        public LazyConcurrentDictionary()
        {
            _concurrentDictionary = new ConcurrentDictionary<TKey, Lazy<TValue>>();
        }

        public TValue GetOrAdd(TKey key, TValue value)
        {
            var lazyResult = _concurrentDictionary.GetOrAdd(key, k => new Lazy<TValue>(() => value, LazyThreadSafetyMode.ExecutionAndPublication));
            return lazyResult.Value;
        }

        public TValue GetOrAdd(TKey key, Func<TKey, TValue> valueFactory)
        {
            var lazyResult = _concurrentDictionary.GetOrAdd(key, k => new Lazy<TValue>(() => valueFactory(k), LazyThreadSafetyMode.ExecutionAndPublication));
            return lazyResult.Value;
        }

        public bool TryGetValue(TKey key, out TValue? value)
        {
            bool success = _concurrentDictionary.TryGetValue(key, out Lazy<TValue>? lazyResult);
            value = success ? lazyResult!.Value : default;
            return success;
        }

        public bool TryRemove(TKey key, out TValue? value)
        {
            var success = _concurrentDictionary.TryRemove(key, out Lazy<TValue>? lazyResult);
            value = success ? lazyResult!.Value : default;
            return success;
        }
    }
}

Found a different way

Found this in a Microsoft repo

https://github.com/microsoft/FASTER/blob/master/cs/src/core/Utilities/SafeConcurrentDictionary.cs

This is a simpler solution to the atomicity problem but not sure if it needs the lazy wrapped on it, just wanted to share.