A simple method to execute Entity Framework Core queries in parallel

Question

As you might know, EF Core does not support executing multiple queries in parallel on the same context using something like Task.WhenAll.

So you have to write this code instead:

using var context1 = _factory.CreateDbContext();
using var context2 = _factory.CreateDbContext();
using var context3 = _factory.CreateDbContext();

var result1Task = context1.Set<T1>().ToListAsync();
var result2Task = context2.Set<T2>().ToListAsync();
var result3Task = context3.Set<T3>().ToListAsync();

await Task.WhenAll(result1Task, result2Task, result3Task);

var result1 = result1Task.Result;
var result2 = result2Task.Result;
var result3 = result3Task.Result;

To get rid of all these using statements, I started writing a library which exposes a Set<TEntity() extension method on IDbContextFactory<TContext>. You can use it like this.

var result1Task = _factory.Set<T1>().ToListAsync();
var result2Task =  _factory.Set<T2>().ToListAsync();
var result3Task =  _factory.Set<T3>().ToListAsync();

await Task.WhenAll(result1Task, result2Task, result3Task);

var result1 = result1Task.Result;
var result2 = result2Task.Result;
var result3 = result3Task.Result;

To achieve this, it heavily relies on reflection and EF Core internals. And because of this, I would really like a code review from someone experienced with EF Core, Expressions and IQueryable.

What it basically does is, it creates an IQueryable<TEntity> which is then exchanged for a real DbSet<TEntity> once it is executed and immediately disposes the context after that. When using the async version, it is a bit more complex. It passes down the DbContext to the AsyncEnumerator and disposes the context together with the AsyncEnumerator.

You can find the whole project on Github too https://github.com/wertzui/EntityFrameworkCore.Parallel

Tank you in advance.

DbContextFactoryExtensions.cs

/// <summary>
/// Contains the <see cref="Set{TContext, TEntity}(IDbContextFactory{TContext})"/> extension method wich is the starting point for any query.
/// </summary>
public static class DbContextFactoryExtensions
{
    /// <summary>
    /// The starting point for your query. Make sure <typeparamref name="TEntity"/> is available as a <see cref="DbSet{TEntity}"/> on your <typeparamref name="TContext"/>.
    /// </summary>
    /// <typeparam name="TContext">The type of the <see cref="DbContext"/>.</typeparam>
    /// <typeparam name="TEntity">The type fo the entities.</typeparam>
    /// <param name="contextFactory">The factory which can create the <see cref="DbContext"/>.</param>
    /// <returns></returns>
    public static IQueryable<TEntity> Set<TEntity>(this IDbContextFactory<DbContext> contextFactory)
        //where TContext : DbContext
        where TEntity : class
    {
        var query = new EntityQueryable<TEntity>(
            new QueryProvider(new DbContextFactoryQueryContext<TEntity>(contextFactory)),
            new EntityType<TEntity>());
        return query;
    }
}

EntityType.cs

/// <summary>
/// This is a minimal implementation of <see cref="IEntityType"/> to support the creation of queries without the need of a DbContext.
/// Most of its methods will just throw exceptions.
/// When the query is actually executed, the expression which is started with this class is replaced to contain an instance of the implementation from the real Entity Framework library.
/// </summary>
/// <typeparam name="TEntity">The type of the entities in the DbSet.</typeparam>
public class EntityType<TEntity> : IEntityType
{
    public object this[string name] => throw new NotImplementedException();

    public IEntityType BaseType => throw new NotImplementedException();

    public string DefiningNavigationName => throw new NotImplementedException();

    public IEntityType DefiningEntityType => throw new NotImplementedException();

    public IModel Model => throw new NotImplementedException();

    public string Name => throw new NotImplementedException();

    public Type ClrType => typeof(TEntity);

    public bool HasSharedClrType => throw new NotImplementedException();

    public bool IsPropertyBag => throw new NotImplementedException();

    public IAnnotation FindAnnotation(string name) => throw new NotImplementedException();
    public IForeignKey FindForeignKey(IReadOnlyList<IProperty> properties, IKey principalKey, IEntityType principalEntityType) => throw new NotImplementedException();
    public IIndex FindIndex(IReadOnlyList<IProperty> properties) => throw new NotImplementedException();
    public IIndex FindIndex(string name) => throw new NotImplementedException();
    public IKey FindKey(IReadOnlyList<IProperty> properties) => throw new NotImplementedException();
    public IKey FindPrimaryKey() => throw new NotImplementedException();
    public IProperty FindProperty(string name) => throw new NotImplementedException();
    public IServiceProperty FindServiceProperty(string name) => throw new NotImplementedException();
    public ISkipNavigation FindSkipNavigation(string name) => throw new NotImplementedException();
    public IEnumerable<IAnnotation> GetAnnotations() => throw new NotImplementedException();
    public IEnumerable<IForeignKey> GetForeignKeys() => throw new NotImplementedException();
    public IEnumerable<IIndex> GetIndexes() => throw new NotImplementedException();
    public IEnumerable<IKey> GetKeys() => throw new NotImplementedException();
    public IEnumerable<IProperty> GetProperties() => throw new NotImplementedException();
    public IEnumerable<IServiceProperty> GetServiceProperties() => throw new NotImplementedException();
    public IEnumerable<ISkipNavigation> GetSkipNavigations() => throw new NotImplementedException();
}

QueryProvider.cs

/// <summary>
/// A basic query provider which will pass the execution logic down the the given <see cref="IQueryContext"/>.
/// </summary>
public class QueryProvider : IAsyncQueryProvider
{
    private readonly IQueryContext _queryContext;
    private static readonly MethodInfo _genericCreateQueryMethod
        = typeof(QueryProvider)
            .GetRuntimeMethods()
            .Single(m => m.Name == nameof(CreateQuery) && m.IsGenericMethod);
    private readonly MethodInfo _genericExecuteMethod;

    public QueryProvider(IQueryContext queryContext)
    {
        _queryContext = queryContext;

        _genericExecuteMethod = _queryContext.GetType()
             .GetRuntimeMethods()
             .Single(m => m.Name == nameof(IQueryContext.Execute) && m.IsGenericMethod);
    }

    public virtual IQueryable CreateQuery(Expression expression)
        => (IQueryable)_genericCreateQueryMethod
            .MakeGenericMethod(expression.Type.GetSequenceType())
            .Invoke(this, new object[] { expression });

    public virtual IQueryable<T> CreateQuery<T>(Expression expression)
        => new EntityQueryable<T>(this, expression);

    public virtual object Execute(Expression expression)
        => _genericExecuteMethod
        .MakeGenericMethod(expression.Type)
            .Invoke(_queryContext, new object[] { expression });

    TResult IQueryProvider.Execute<TResult>(Expression expression)
        => _queryContext.Execute<TResult>(expression);

    public TResult ExecuteAsync<TResult>(Expression expression, CancellationToken cancellationToken = default)
        => _queryContext.ExecuteAsync<TResult>(expression, cancellationToken);
}

IQueryContext.cs

/// <summary>
/// Interface for a query context. A query context is the thing that will actually execute any query.
/// </summary>
public interface IQueryContext
{
    TResult Execute<TResult>(Expression query);
    TResult ExecuteAsync<TResult>(Expression query, CancellationToken cancellationToken);
}

DbContextFactoryQueryContext.cs

/// <summary>
/// This class contains the logic which will actually create the <see cref="DbContext"/> and execute the query on it.
/// </summary>
/// <typeparam name="TContext"></typeparam>
/// <typeparam name="TEntity"></typeparam>
public class DbContextFactoryQueryContext<TEntity> : DbContextFactoryQueryContext, IQueryContext
    //where TContext : DbContext
    where TEntity : class
{
    private readonly IDbContextFactory<DbContext> factory;

    public DbContextFactoryQueryContext(IDbContextFactory<DbContext> factory)
    {
        this.factory = factory ?? throw new ArgumentNullException(nameof(factory));
    }

    public TResult Execute<TResult>(Expression query)
    {
        using (var context = factory.CreateDbContext())
        {
            context.ChangeTracker.QueryTrackingBehavior = QueryTrackingBehavior.NoTracking;
            var set = context.Set<TEntity>();
            var queryaple = set.AsQueryable();
            var provider = queryaple.Provider;

            var replaced = ReplaceProvider(query, set, provider);
            var result = provider.Execute<TResult>(replaced);
            var buffered = Buffer(result);
            return buffered;
        }
    }

    private static Expression ReplaceProvider(Expression query, DbSet<TEntity> set, IQueryProvider provider)
    {
        var setQuery = provider is IAsyncQueryProvider asyncProvider ?
                            new QueryRootExpression(asyncProvider, set.EntityType) :
                            new QueryRootExpression(set.EntityType);
        Expression replaced;
        if (query is MethodCallExpression call)
            replaced = call.Update(call.Object, new Expression[] { setQuery }.Concat(call.Arguments.Skip(1)));
        else if (query is QueryRootExpression root)
            replaced = setQuery;
        else
            throw new ArgumentException($"An Expression of type {query.Type} is not supported.", nameof(query));
        return replaced;
    }

    private static TResult Buffer<TResult>(TResult result, CancellationToken cancellationToken = default)
    {
        if (result is IEnumerable enumerable)
        {
            var type = enumerable.GetType();
            if (type.GenericTypeArguments.Length == 1)
            {
                var genericArgument = type.GenericTypeArguments[0];

                var genericToList = _toListMethodInfo.MakeGenericMethod(genericArgument);
                var list = genericToList.Invoke(null, new object[] { enumerable });

                var casted = (TResult)list;
                return casted;
            }
        }

        return result;
    }

    public TResult ExecuteAsync<TResult>(Expression query, CancellationToken cancellationToken)
    {
        // We cannot use a using block here, because the result will be enumerated after this method has already returned.
        // Instead we pass the DbContext down to the enumerator which will then dispose the context once itself gets disposed.
        // This will happen when the result is enumerated.
        var context = factory.CreateDbContext();
        try
        {
            context.ChangeTracker.QueryTrackingBehavior = QueryTrackingBehavior.NoTracking;
            var set = context.Set<TEntity>();
            var queryaple = set.AsQueryable();
            var provider = queryaple.Provider;

            if (provider is IAsyncQueryProvider asyncProvider)
            {
                var replaced = ReplaceProvider(query, set, provider);
                var result = asyncProvider.ExecuteAsync<TResult>(replaced);
                var buffered = BufferAsync(result, context, cancellationToken);
                return buffered;
            }

            throw new InvalidOperationException("Cannot execute an async query on a non async query provider.");
        }
        catch
        {
            context.Dispose();
            throw;
        }
    }

    private static TResult BufferAsync<TResult>(TResult result, IAsyncDisposable dbSet, CancellationToken cancellationToken)
    {
        if (result is IEnumerable enumerable)
        {
            var buffered = EntitiyFrameworkCore.Parallel.AsyncEnumerableExtensions.Buffer((dynamic)result, dbSet);
            return (TResult)buffered;
        }

        return result;
    }
}


public abstract class DbContextFactoryQueryContext
{
    protected static readonly MethodInfo _toListMethodInfo = typeof(Enumerable)
                    .GetMethod(nameof(Enumerable.ToList));

    protected static readonly MethodInfo _toListAsyncMethodInfo = typeof(AsyncEnumerable)
                    .GetMethod(nameof(AsyncEnumerable.ToListAsync));

    protected static readonly MethodInfo _toAsyncEnumerableMethodInfo = typeof(AsyncEnumerable)
                    .GetMethods()
                    .Single(m =>
                    {
                        if (m.Name != nameof(AsyncEnumerable.ToAsyncEnumerable))
                            return false;

                        var parameter = m.GetParameters().FirstOrDefault();
                        if (parameter == null)
                            return false;

                        var parameterType = parameter.ParameterType;
                        if (!typeof(Task).IsAssignableFrom(parameterType))
                            return false;

                        return true;
                    });
}

DbContextFactoryQueryContext.cs

/// <summary>
/// This class contains the logic which will actually create the <see cref="DbContext"/> and execute the query on it.
/// </summary>
/// <typeparam name="TContext"></typeparam>
/// <typeparam name="TEntity"></typeparam>
public class DbContextFactoryQueryContext<TEntity> : DbContextFactoryQueryContext, IQueryContext
    //where TContext : DbContext
    where TEntity : class
{
    private readonly IDbContextFactory<DbContext> factory;

    public DbContextFactoryQueryContext(IDbContextFactory<DbContext> factory)
    {
        this.factory = factory ?? throw new ArgumentNullException(nameof(factory));
    }

    public TResult Execute<TResult>(Expression query)
    {
        using (var context = factory.CreateDbContext())
        {
            context.ChangeTracker.QueryTrackingBehavior = QueryTrackingBehavior.NoTracking;
            var set = context.Set<TEntity>();
            var queryaple = set.AsQueryable();
            var provider = queryaple.Provider;

            var replaced = ReplaceProvider(query, set, provider);
            var result = provider.Execute<TResult>(replaced);
            var buffered = Buffer(result);
            return buffered;
        }
    }

    private static Expression ReplaceProvider(Expression query, DbSet<TEntity> set, IQueryProvider provider)
    {
        var setQuery = provider is IAsyncQueryProvider asyncProvider ?
                            new QueryRootExpression(asyncProvider, set.EntityType) :
                            new QueryRootExpression(set.EntityType);
        Expression replaced;
        if (query is MethodCallExpression call)
            replaced = call.Update(call.Object, new Expression[] { setQuery }.Concat(call.Arguments.Skip(1)));
        else if (query is QueryRootExpression root)
            replaced = setQuery;
        else
            throw new ArgumentException($"An Expression of type {query.Type} is not supported.", nameof(query));
        return replaced;
    }

    private static TResult Buffer<TResult>(TResult result, CancellationToken cancellationToken = default)
    {
        if (result is IEnumerable enumerable)
        {
            var type = enumerable.GetType();
            if (type.GenericTypeArguments.Length == 1)
            {
                var genericArgument = type.GenericTypeArguments[0];

                var genericToList = _toListMethodInfo.MakeGenericMethod(genericArgument);
                var list = genericToList.Invoke(null, new object[] { enumerable });

                var casted = (TResult)list;
                return casted;
            }
        }

        return result;
    }

    public TResult ExecuteAsync<TResult>(Expression query, CancellationToken cancellationToken)
    {
        // We cannot use a using block here, because the result will be enumerated after this method has already returned.
        // Instead we pass the DbContext down to the enumerator which will then dispose the context once itself gets disposed.
        // This will happen when the result is enumerated.
        var context = factory.CreateDbContext();
        try
        {
            context.ChangeTracker.QueryTrackingBehavior = QueryTrackingBehavior.NoTracking;
            var set = context.Set<TEntity>();
            var queryaple = set.AsQueryable();
            var provider = queryaple.Provider;

            if (provider is IAsyncQueryProvider asyncProvider)
            {
                var replaced = ReplaceProvider(query, set, provider);
                var result = asyncProvider.ExecuteAsync<TResult>(replaced);
                var buffered = BufferAsync(result, context, cancellationToken);
                return buffered;
            }

            throw new InvalidOperationException("Cannot execute an async query on a non async query provider.");
        }
        catch
        {
            context.Dispose();
            throw;
        }
    }

    private static TResult BufferAsync<TResult>(TResult result, IAsyncDisposable dbSet, CancellationToken cancellationToken)
    {
        if (result is IEnumerable enumerable)
        {
            var buffered = EntitiyFrameworkCore.Parallel.AsyncEnumerableExtensions.Buffer((dynamic)result, dbSet);
            return (TResult)buffered;
        }

        return result;
    }
}


public abstract class DbContextFactoryQueryContext
{
    protected static readonly MethodInfo _toListMethodInfo = typeof(Enumerable)
                    .GetMethod(nameof(Enumerable.ToList));

    protected static readonly MethodInfo _toListAsyncMethodInfo = typeof(AsyncEnumerable)
                    .GetMethod(nameof(AsyncEnumerable.ToListAsync));

    protected static readonly MethodInfo _toAsyncEnumerableMethodInfo = typeof(AsyncEnumerable)
                    .GetMethods()
                    .Single(m =>
                    {
                        if (m.Name != nameof(AsyncEnumerable.ToAsyncEnumerable))
                            return false;

                        var parameter = m.GetParameters().FirstOrDefault();
                        if (parameter == null)
                            return false;

                        var parameterType = parameter.ParameterType;
                        if (!typeof(Task).IsAssignableFrom(parameterType))
                            return false;

                        return true;
                    });
}

SharedTypeExtensions.cs

/// <summary>
/// Contains extension methods to find the generic type parameter of a collection type.
/// </summary>
internal static class SharedTypeExtensions
{
    public static Type GetSequenceType(this Type type)
    {
        var sequenceType = TryGetSequenceType(type);
        if (sequenceType == null)
        {
            throw new ArgumentException("The given type is not a collection type", nameof(type));
        }

        return sequenceType;
    }

    public static Type TryGetSequenceType(this Type type)
        => type.TryGetElementType(typeof(IEnumerable<>))
            ?? type.TryGetElementType(typeof(IAsyncEnumerable<>));
    public static Type TryGetElementType(this Type type, Type interfaceOrBaseType)
    {
        if (type.IsGenericTypeDefinition)
        {
            return null;
        }

        var types = GetGenericTypeImplementations(type, interfaceOrBaseType);

        Type singleImplementation = null;
        foreach (var implementation in types)
        {
            if (singleImplementation == null)
            {
                singleImplementation = implementation;
            }
            else
            {
                singleImplementation = null;
                break;
            }
        }

        return singleImplementation?.GenericTypeArguments.FirstOrDefault();
    }

    public static IEnumerable<Type> GetGenericTypeImplementations(this Type type, Type interfaceOrBaseType)
    {
        var typeInfo = type.GetTypeInfo();
        if (!typeInfo.IsGenericTypeDefinition)
        {
            var baseTypes = interfaceOrBaseType.GetTypeInfo().IsInterface
                ? typeInfo.ImplementedInterfaces
                : type.GetBaseTypes();
            foreach (var baseType in baseTypes)
            {
                if (baseType.IsGenericType
                    && baseType.GetGenericTypeDefinition() == interfaceOrBaseType)
                {
                    yield return baseType;
                }
            }

            if (type.IsGenericType
                && type.GetGenericTypeDefinition() == interfaceOrBaseType)
            {
                yield return type;
            }
        }
    }

    public static IEnumerable<Type> GetBaseTypes(this Type type)
    {
        type = type.BaseType;

        while (type != null)
        {
            yield return type;

            type = type.BaseType;
        }
    }
}

To those encountering this question in the close queue, the issue has been corrected. — pacmaninbw, Commented Nov 18, 2020 at 12:47

Johnbot · Accepted Answer · 2020-11-19 11:23:36Z

All public API surfaces should validate input parameters.

Restricting the queries to a single entity type seems limiting. You can't join entities with no navigation properties.

It fails when using navigation properties (C# 9/EF Core 5.0):

using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using System.Collections.Generic;
using System.Diagnostics.CodeAnalysis;
using System.Linq;

var services = new ServiceCollection();
services.AddPooledDbContextFactory<TestDbContext>(options => options.UseInMemoryDatabase("hest"));

var provider = services.BuildServiceProvider();
var factory = provider.GetRequiredService<IDbContextFactory<TestDbContext>>();

using (var db = provider.GetRequiredService<IDbContextFactory<TestDbContext>>().CreateDbContext())
{
    await db.Ones.AddRangeAsync(
        Enumerable.Range(1, 10)
                  .Select(i => new Entity1 { Value = (i * 2).ToString(), Entity2 = new Entity2 { Value = (i * 7).ToString(), } }));
    await db.SaveChangesAsync();
}

// Fails when rewriting the query
var fails = await factory.Set<Entity1>()
              .Select(t => t.Entity2)
              .Select(t => t.Entity1)
              .ToListAsync();

// Can't join two DbSets with different contexts.
var test = await factory.Set<Entity1>()
                        .Join(factory.Set<Entity2>(), e1 => e1.ID, e2 => e2.ID, (e1, e2) => e2.Value)
                        .ToListAsync();

public class Entity1
{
    public long ID { get; set; }
    public string Value { get; set; }

    public Entity2 Entity2 { get; set; }
}

public class Entity2
{
    public long ID { get; set; }
    public string Value { get; set; }

    public Entity1 Entity1 { get; set; }
}

public class TestDbContext : DbContext
{
    public TestDbContext([NotNull] DbContextOptions<TestDbContext> options) : base(options) { }

    public DbSet<Entity1> Ones { get; protected set; }
    public DbSet<Entity2> Twos { get; protected set; }

    protected override void OnModelCreating(ModelBuilder modelBuilder)
    {
        base.OnModelCreating(modelBuilder);

        modelBuilder.Entity<Entity2>()
                    .HasOne(e => e.Entity1)
                    .WithOne(e => e.Entity2)
                    .HasPrincipalKey<Entity1>(e => e.ID)
                    .HasForeignKey<Entity2>(e => e.ID);
    }
}

If a slightly uglier API is acceptable the whole thing can be implemented with a single extension method:

public static class EFConcurrentQueryExtension
{
    public static async Task<TResult> ConcurrentAsync<TDbContext, TResult>(
        this IDbContextFactory<TDbContext> dbContextFactory,
        Func<TDbContext, Task<TResult>> asyncQuery)
        where TDbContext : DbContext
    {
        if (dbContextFactory is null)
        {
            throw new ArgumentNullException(nameof(dbContextFactory));
        }

        if (asyncQuery is null)
        {
            throw new ArgumentNullException(nameof(asyncQuery));
        }

        using var context = dbContextFactory.CreateDbContext();
        context.ChangeTracker.QueryTrackingBehavior = QueryTrackingBehavior.NoTracking;
        var result = await asyncQuery(context);
        return result;
    }
}

And the example would look like:

var result1Task = _factory.ConcurrentAsync(ctx => ctx.Set<T1>().ToListAsync());
var result2Task = _factory.ConcurrentAsync(ctx => ctx.T2.ToListAsync());
var result3Task = _factory.ConcurrentAsync(ctx => ctx.Set<T3>().ToListAsync());

await Task.WhenAll(result1Task, result2Task, result3Task);

var result1 = await result1Task;
var result2 = await result2Task;
var result3 = await result3Task;

Thanks for the review! I'll try to fix the problem with the navigation properties. I know that Join() is not working and that is acceptable. Also the "uglier API" is not acceptable as I want to keep the syntax equal to what everybody knows from a normal EF query. — wertzui, Commented Nov 19, 2020 at 15:21
For anyone interested: The first (and fixed) version is now available on NuGet nuget.org/packages/EntityFrameworkCore.Parallel — wertzui, Commented Nov 25, 2020 at 16:32

Stack Exchange Network

A simple method to execute Entity Framework Core queries in parallel

1 Answer 1

Your Answer

Not the answer you're looking for? Browse other questions tagged
c#
linq
expression-trees
entity-framework-core
linq-expressions
or ask your own question.

Hot Network Questions

A simple method to execute Entity Framework Core queries in parallel

1 Answer 1

Your Answer

Sign up or log in

Post as a guest

Not the answer you're looking for? Browse other questions tagged c#linqexpression-treesentity-framework-corelinq-expressions or ask your own question.

Related

Hot Network Questions

Not the answer you're looking for? Browse other questions tagged
c#
linq
expression-trees
entity-framework-core
linq-expressions
or ask your own question.