A Threadsafe implementation of DbContext

How to enable Entity Framework DbContext to multithreading

In concurrent environments, making your Entity Framework `DbContext` thread-safe is essential to ensure data integrity and avoid race conditions. The `DbContext` class in Entity Framework is not inherently thread-safe. This article aims to present an implementation that extends `DbContext` to make it thread-safe, allowing for smooth operations in multithreaded applications.

The source are available at https://github.com/remihenache/ThreadSafeDbContext

And the package can be installed using nuget:

dotnet add package ThreadSafeDbContext

Why is Thread-Safety Necessary?

In multi-threaded applications, multiple threads might attempt to access the database concurrently. If `DbContext` is not thread-safe, this leads to:

- Data Corruption
- Deadlocks
- Inconsistent Reads/Writes

A thread-safe `DbContext` ensures that operations are atomic and isolated from each other, which is crucial for data integrity and application stability.

As I like Domain Events, and publish them as concurrency task, I need to share my DbContext (and the attached transaction) to these task. This is typically the case that make me create this.

How does it work

The thread-safety is achieved by using `SemaphoreSlim` to lock the resources. The custom implementation provides a layer over `DbContext`, ensuring that only one thread can access it at any given time for sensitive operations.

1. ThreadSafeEnumerator: Wraps around `IEnumerator` and locks the resource while the enumerator move. Enable thread safe access at each iteration.

internal class ThreadSafeEnumerator : IEnumerator
{
    protected readonly IEnumerator Enumerator;
    protected readonly SemaphoreSlim SemaphoreSlim;

    public ThreadSafeEnumerator(IEnumerator enumerator, SemaphoreSlim semaphoreSlim)
    {
        this.Enumerator = enumerator;
        this.SemaphoreSlim = semaphoreSlim;
    }

    public Boolean MoveNext()
    {
        this.SemaphoreSlim.Wait();
        try
        {
            return this.Enumerator.MoveNext();
        }
        finally
        {
            this.SemaphoreSlim.Release();
        }
    }

    public void Reset()
    {
        this.SemaphoreSlim.Wait();
        try
        {
            this.Enumerator.Reset();
        }
        finally
        {
            this.SemaphoreSlim.Release();
        }
    }

    public Object Current => this.Enumerator.Current!;
}

2. ThreadSafeQueryable: Wraps around `IQueryable` and use the ThreadSafeEnumerator to enable thread safe access. It also wrap the QueryProvider into a thread safe query provider.

internal class ThreadSafeQueryable : IOrderedQueryable
{
    protected readonly SemaphoreSlim SemaphoreSlim;
    protected readonly IQueryable Set;

    public ThreadSafeQueryable(IQueryable set, SemaphoreSlim semaphoreSlim)
    {
        this.Set = set;
        this.SemaphoreSlim = semaphoreSlim;
    }


    public IEnumerator GetEnumerator()
    {
        return new ThreadSafeEnumerator(this.Set.GetEnumerator(), this.SemaphoreSlim);
    }

    public Type ElementType => this.Set.ElementType;

    public Expression Expression => this.Set.Expression;

    public IQueryProvider Provider => new ThreadSafeQueryProvider(this.Set.Provider, this.SemaphoreSlim);
}

3. ThreadSafeQueryProvider: Implements a thread-safe `IQueryProvider`. It wrap a query provider, to lock sensitive operation around it. And wrap the create query, to encapsulate it into a ThreadSafeQueryable, as above.

internal sealed class ThreadSafeQueryProvider : IAsyncQueryProvider
{
    private readonly IQueryProvider queryProvider;
    private readonly SemaphoreSlim semaphoreSlim;

    public ThreadSafeQueryProvider(
        IQueryProvider queryProvider,
        SemaphoreSlim semaphoreSlim)
    {
        this.queryProvider = queryProvider;
        this.semaphoreSlim = semaphoreSlim;
    }

    public IQueryable CreateQuery(Expression expression)
    {
        return new ThreadSafeQueryable(this.queryProvider.CreateQuery(expression), this.semaphoreSlim);
    }

    public IQueryable<TElement> CreateQuery<TElement>(Expression expression)
    {
        return new ThreadSafeQueryable<TElement>(this.queryProvider.CreateQuery<TElement>(expression),
            this.semaphoreSlim);
    }

    public Object? Execute(Expression expression)
    {
        this.semaphoreSlim.Wait();
        try
        {
            return this.queryProvider.Execute(expression);
        }
        finally
        {
            this.semaphoreSlim.Release();
        }
    }

    public TResult Execute<TResult>(Expression expression)
    {
        this.semaphoreSlim.Wait();
        try
        {
            return this.queryProvider.Execute<TResult>(expression);
        }
        finally
        {
            this.semaphoreSlim.Release();
        }
    }

    public TResult ExecuteAsync<TResult>(Expression expression, CancellationToken cancellationToken = new())
    {
        this.semaphoreSlim.Wait(cancellationToken);
        try
        {
            return (this.queryProvider as IAsyncQueryProvider)!.ExecuteAsync<TResult>(expression, cancellationToken);
        }
        finally
        {
            this.semaphoreSlim.Release();
        }
    }
}

4. ThreadSafeDbSet: A DbSet class extended to handle thread safety. This is probably the most sensitive wrapper. It wrap the DbSet, to make it use the ThreadSafeQueryable, and ensure that some other operations are wrap by the locker.

internal sealed class ThreadSafeDbSet<TEntity> :
    DbSet<TEntity>,
    IQueryable<TEntity>,
    IAsyncEnumerable<TEntity>
    where TEntity : class
{
    private readonly SemaphoreSlim semaphoreSlim;
    private readonly DbSet<TEntity> set;

    public ThreadSafeDbSet(DbSet<TEntity> set, SemaphoreSlim semaphoreSlim)
    {
        this.set = set;
        this.semaphoreSlim = semaphoreSlim;
    }

    public override IEntityType EntityType => this.set.EntityType;

    public override LocalView<TEntity> Local => SafeExecute(() => this.set.Local);

    public override EntityEntry<TEntity> Add(TEntity entity)
    {
        return SafeExecute(() => this.set.Add(entity));
    }

    public override EntityEntry<TEntity> Attach(TEntity entity)
    {
        return SafeExecute(() => this.set.Attach(entity));
    }

    public override EntityEntry<TEntity> Update(TEntity entity)
    {
        return SafeExecute(() => this.set.Update(entity));
    }

    // ... All other override using SafeExecute

    private void SafeExecute(Action func)
    {
        this.semaphoreSlim.Wait();
        try
        {
            func();
        }
        finally
        {
            this.semaphoreSlim.Release();
        }
    }

    private T SafeExecute<T>(Func<T> func)
    {
        this.semaphoreSlim.Wait();
        try
        {
            return func();
        }
        finally
        {
            this.semaphoreSlim.Release();
        }
    }

    #region Queryable

    public Type ElementType => (this.set as IQueryable).ElementType;
    public Expression Expression => (this.set as IQueryable).Expression;
    
    public override IQueryable<TEntity> AsQueryable()
    {
        return new ThreadSafeQueryable<TEntity>(this.set.AsQueryable(), this.semaphoreSlim);
    }

    public override IAsyncEnumerable<TEntity> AsAsyncEnumerable()
    {
        return new ThreadSafeQueryable<TEntity>(this.set.AsQueryable(), this.semaphoreSlim);
    }

    public override IAsyncEnumerator<TEntity> GetAsyncEnumerator(CancellationToken cancellationToken = new())
    {
        return new ThreadSafeQueryable<TEntity>(this.set.AsQueryable(), this.semaphoreSlim).GetAsyncEnumerator(
            cancellationToken);
    }

    IEnumerator IEnumerable.GetEnumerator()
    {
        return GetEnumerator();
    }

    public IQueryProvider Provider =>
        new ThreadSafeQueryable<TEntity>(this.set.AsQueryable(), this.semaphoreSlim).Provider;

    public IEnumerator<TEntity> GetEnumerator()
    {
        return new ThreadSafeQueryable<TEntity>(this.set.AsQueryable(), this.semaphoreSlim).GetEnumerator();
    }

    #endregion
}

5. ThreadSafeDbContext: Extends `DbContext` and overrides methods to make them thread-safe. It use the ThreadSafeDbSet as DbSet, to ensure that associated operation are safe.

public class ThreadSafeDbContext : DbContext
{
    private readonly SemaphoreSlim semaphoreSlim = new(1, 1);

    public ThreadSafeDbContext()
    {
    }

    public ThreadSafeDbContext(DbContextOptionsBuilder optionsBuilder)
        : base(optionsBuilder
            .EnableThreadSafetyChecks(false)
            .Options)
    {
    }

    protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
    {
        optionsBuilder.EnableThreadSafetyChecks(false);
        base.OnConfiguring(optionsBuilder);
    }

    public override DbSet<TEntity> Set<TEntity>()
    {
        this.semaphoreSlim.Wait();
        try
        {
            return new ThreadSafeDbSet<TEntity>(base.Set<TEntity>(), this.semaphoreSlim);
        }
        finally
        {
            this.semaphoreSlim.Release();
        }
    }


    public override async Task<Int32> SaveChangesAsync(CancellationToken cancellationToken = new())
    {
        await this.semaphoreSlim.WaitAsync(cancellationToken);
        try
        {
            return await base.SaveChangesAsync(cancellationToken);
        }
        finally
        {
            this.semaphoreSlim.Release();
        }
    }

    public override Int32 SaveChanges()
    {
        this.semaphoreSlim.Wait();
        try
        {
            return base.SaveChanges();
        }
        finally
        {
            this.semaphoreSlim.Release();
        }
    }
}

Using SemaphoreSlim

`SemaphoreSlim` is used as a locking mechanism. A count of one allows a single thread to enter the critical section, effectively making it thread-safe.

The combinaison of Enumerator, Queryable and QueryProvider

These three class work together, each thread safe implementation using the threadsafe implementation of the other, to wrap as well as possible the complexity of Entity Framework Queryable. As they share the same lock that the DbContext, each operation is thread safe.

How to Use It

The only available class is the ThreadSafeDbContext, so instead of making your application DbContext inherits from DbContext, make it inherits from the ThreadSafeDbContext:

public class MyDbContext : ThreadSafeDbContext
{
}

Then, you can use it as the standard DbContext:

public class MyDbContext : ThreadSafeDbContext
{

    protected override void OnConfiguring(DbContextOptionsBuilder optionsBuilder)
    {
        optionsBuilder.UseSqlServer("MyConnectionString");
        base.OnConfiguring(optionsBuilder);
    }

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        modelBuilder.Entity<MyEntity>().ToTable("MyEntity");
        modelBuilder.Entity<MyEntity>().HasKey(t => t.ID);
        base.OnModelCreating(modelBuilder);
    }
}
// or use DbContextOptionBuilder constructor
public class MyDbContext : ThreadSafeDbContext
{

    public MyDbContext(DbContextOptionsBuilder<ThreadSafeDbContext> optionsBuilder) : base(optionsBuilder)
    {
    }

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        modelBuilder.Entity<MyEntity>().ToTable("MyEntity");
        modelBuilder.Entity<MyEntity>().HasKey(t => t.ID);
        base.OnModelCreating(modelBuilder);
    }
}

Now, you can freely run your LINQ queries and CRUD operations. All of these will be thread-safe.

var users = myDbContext.Set<Users>.ToList();
// or async
var users = await myDbContext.Set<Users>.ToListAsync();

As you can modify your data, and persist them safely

myDbContext.Set<Users>.Add(new Users(){ Id = 1 });
dbContext.SaveChanges();
// or async
myDbContext.Set<Users>.Add(new Users(){ Id = 1 });
await dbContext.SaveChangesAsync();

Pros and Cons

Pros

- Guarantees data integrity in a multithreaded environment.
- Simple integration into existing projects.

Cons

- Potential for decreased performance due to locking.
- Complexity could increase if nested locks are involved.

Conclusion

Thread safety is critical for applications that require high reliability and data integrity. This implementation provides a foundational layer to ensure that your Entity Framework operations are thread-safe.

You could find the source at https://github.com/remihenache/ThreadSafeDbContext

And the package can be installed using nuget:

dotnet add package ThreadSafeDbContext

— -

Note: There are various ways to achieve thread safety, and the above example is just one approach. Always measure performance impacts when implementing synchronization primitives like semaphores or locks.