具有多个同时批量 MERGE 语句的 EF Core 5 事务死锁答案

【问题标题】：EF Core 5 Transaction Deadlock with multiple simultaneous bulk MERGE statements具有多个同时批量 MERGE 语句的 EF Core 5 事务死锁
【发布时间】：2023-04-02 13:09:01
【问题描述】：

我有一个应用程序，它运行多个线程以使用 EF Core 5 将数据插入到 SQL Server 2017 数据库表中。

使用 EF Core 5 插入域模型实体的 C# 代码如下：

using (var ctx = this.dbContextFactory.CreateDbContext())
{
    //ctx.Database.AutoTransactionsEnabled = false;
    foreach (var rootEntity in request.RootEntities)
    {
        ctx.ChangeTracker.TrackGraph(rootEntity, node =>
        {
            if ((request.EntityTypes != null && request.EntityTypes.Contains(node.Entry.Entity.GetType()))
                || rootEntity == node.Entry.Entity)
            {
                if (node.Entry.IsKeySet)
                    node.Entry.State = Microsoft.EntityFrameworkCore.EntityState.Modified;
                else
                    node.Entry.State = Microsoft.EntityFrameworkCore.EntityState.Added;
            }
        });
    }
    await ctx.SaveChangesAsync(cancellationToken);
}

每个线程负责实例化自己的 DbContext 实例，因此使用 dbContextFactory。

为 INSERT (MERGE) 生成的一些示例 SQL 如下：

SET NOCOUNT ON;
DECLARE @inserted0 TABLE ([OrderId] bigint, [_Position] [int]);
MERGE [dbo].[Orders] USING (
VALUES (@p0, 0),
(@p1, 1),
(@p2, 2),
...
(@43, 41)) AS i ([SomeColumn],  _Position) ON 1=0
WHEN NOT MATCHED THEN
INSERT ([SomeColumn])
VALUES (i.[SomeColumn])
OUTPUT INSERTED.[OrderId], i._Position
INTO @inserted0;

SELECT [t].[OrderId] FROM [dbo].[Orders] t
INNER JOIN @inserted0 i ON ([t].[OrderId] = [i].[OrderId])
ORDER BY [i].[_Position];

由于这些线程经常同时运行，我得到以下 SQL 异常：

Transaction (Process ID 99) was deadlocked on lock resources with another process and has been chosen as the deadlock victim. Rerun the transaction.

EF Core 将隔离级别隐式设置为 READ COMMITTED。

使用 SQL Profiler，事务死锁是由以下原因引起的：

我的担忧：

令人沮丧的是，EF Core 生成的 SQL 包含两条语句：一条 MERGE，然后一条 SELECT。我不明白 SELECT 的目的，因为主键的身份可从 @inserted0 表变量中获得。鉴于 this answer，单独的 MERGE 语句足以使这个原子化。

我相信是这个 SELECT 导致了事务死锁。

我尝试通过使用 READ COMMITTED SNAPSHOT 来解决此问题，以避免与主键查找发生冲突，但是即使此隔离级别应避免锁定并改用行版本控制，我仍然遇到相同的错误。

我解决问题的尝试：

我能找到解决此问题的唯一方法是明确阻止 EF Core 启动事务，因此代码如下：

ctx.Database.AutoTransactionsEnabled = false;

我已经对此进行了多次测试，但没有收到事务死锁。鉴于逻辑只是插入新记录，我相信这是可以做到的。

有人对解决这个问题有什么建议吗？

感谢您的宝贵时间。

【问题讨论】：

为什么不通过不同的线程插入同一张表？
因为 EF 核心已自动将其转换为 MERGE 而不是 INSERT。 40 条记录的批次合并比 40 条单独的 INSERT 语句快。除非 EF Core 中有一个选项可以使用单个 sql 语句进行批量插入？？
我知道为什么它会转化为 MERGE。试图弄清楚如何避免这种情况。您插入哪种数据？也许您只需要通过可用的扩展程序BulkCopy？

标签： c# sql-server transactions entity-framework-core deadlock

【解决方案1】：

我们在多个线程上使用 INSERT (MERGE) 语句时遇到了同样的问题。我们不想为所有事务启用 EnableRetryOnFailure() 选项，因此我们编写了以下 DbContent 扩展方法。

  public static async Task<TResult> SaveWithRetryAsync<TResult>(this DbContext context,
                                                                  Func<Task<TResult>> bulkInsertOperation,
                                                                  Func<TResult, Task<bool>> verifyBulkOperationSucceeded,
                                                                  IsolationLevel isolationLevel = IsolationLevel.Unspecified,
                                                                  int retryLimit = 6,
                                                                  int maxRetryDelayInSeconds = 30)
    {
        var existingTransaction = context.Database.CurrentTransaction?.GetDbTransaction();
        if (existingTransaction != null)
            throw new InvalidOperationException($"Cannot run {nameof(SaveWithRetryAsync)} inside a transaction");

        if (context.ChangeTracker.HasChanges())
        {
            throw new InvalidOperationException(
                "DbContext should be saved before running this action to revert only the changes of this action in case of a concurrency conflict.");
        }

        const int sqlErrorNrOnDuplicatePrimaryKey = 2627;
        const int sqlErrorNrOnSnapshotIsolation = 3960;
        const int sqlErrorDeadlock = 1205;
        int[] sqlErrorsToRetry = { sqlErrorNrOnDuplicatePrimaryKey, sqlErrorNrOnSnapshotIsolation, sqlErrorDeadlock };

        var retryState = new SaveWithRetryState<TResult>(bulkInsertOperation);

        // Use EF Cores connection resiliency feature for retrying (see https://docs.microsoft.com/en-us/ef/core/miscellaneous/connection-resiliency)
        // Usually the IExecutionStrategy is configured DbContextOptionsBuilder.UseSqlServer(..., options.EnableRetryOnFailure()).
        // In ASP.NET, the DbContext is configured in Startup.cs and we don't want this retry behaviour everywhere for each db operation.
        var executionStrategyDependencies = context.Database.GetService<ExecutionStrategyDependencies>();
        var retryStrategy = new CustomSqlServerRetryingExecutionStrategy(executionStrategyDependencies, retryLimit, TimeSpan.FromSeconds(maxRetryDelayInSeconds), sqlErrorsToRetry);

        try
        {
            var result = await retryStrategy.ExecuteInTransactionAsync(
                retryState,
                async (state, cancelToken) =>
                {
                    try
                    {
                        var r = await state.Action();

                        await context.SaveChangesAsync(false, cancelToken);

                        if (state.FirstException != null)
                        {
                            Log.Logger.Warning(
                                $"Action passed to {nameof(SaveWithRetryAsync)} failed {state.NumberOfRetries} times " +
                                $"(retry limit={retryLimit}, ThreadId={Thread.CurrentThread.ManagedThreadId}).\nFirst exception was: {state.FirstException}");
                        }

                        state.Result = r;
                        return r;
                    }
                    catch (Exception ex)
                    {
                        context.RevertChanges();
                        state.NumberOfRetries++;
                        state.FirstException ??= ex;
                        state.LastException = ex;

                        throw;
                    }
                },
                (state, cancelToken) => verifyBulkOperationSucceeded(retryState.Result),
                context.GetSupportedIsolationLevel(isolationLevel));

            context.ChangeTracker.AcceptAllChanges();
            return result;
        }
        catch (Exception ex)
        {
            throw new InvalidOperationException(
                $"DB Transaction in {nameof(SaveWithRetryAsync)} failed. " +
                $"Tried {retryState.NumberOfRetries} times (retry limit={retryLimit}, ThreadId={Thread.CurrentThread.ManagedThreadId}).\n" +
                $"First exception was: {retryState.FirstException}.\nLast exception was: {retryState.LastException}",
                ex);
        }
    }

使用以下 CustomSqlServerRetryingExecutionStrategy

 public class CustomSqlServerRetryingExecutionStrategy : SqlServerRetryingExecutionStrategy
{
    public CustomSqlServerRetryingExecutionStrategy(ExecutionStrategyDependencies executionStrategyDependencies, int retryLimit, TimeSpan fromSeconds, int[] sqlErrorsToRetry)
        : base(executionStrategyDependencies, retryLimit, fromSeconds, sqlErrorsToRetry)
    {
    }

    protected override bool ShouldRetryOn(Exception exception)
    {
        //SqlServerRetryingExecutionStrategy does not check the base exception, maybe a bug in EF core ?!
        return base.ShouldRetryOn(exception) || base.ShouldRetryOn(exception.GetBaseException());
    }
}

帮助类保存当前（重试）状态：

private class SaveWithRetryState<T>
    {
        public SaveWithRetryState(Func<Task<T>> action)
        {
            Action = action;
        }

        public Exception FirstException { get; set; }
        public Exception LastException { get; set; }
        public int NumberOfRetries { get; set; }
        public Func<Task<T>> Action { get; }
        public T Result { get; set; }
    }

现在，扩展方法可以如下使用。代码将尝试多次添加批量 (5)。

 await _context.SaveWithRetryAsync(
            // method to insert the bulk
            async () =>
            {
                var listOfAddedItems = new List<string>();
                foreach (var item in bulkImport)
                {
                    listOfAddedItems.Add(item.Guid);
                    await context.Import.AddAsync(item);
                }

                return listOfAddedItems;
            },
            // method to check if the bulk insert was successful
            listOfAddedItems =>
            {

                if (listOfAddedItems == null)
                    return Task.FromResult(true);
                return _context.Import.AsNoTracking().AnyAsync(x => x.Guid == listOfAddedItems.First());
            },
            IsolationLevel.Snapshot,
            5, // max retry attempts
            100); // max retry time

有关为什么会发生这种情况的背景信息，请查看此讨论：https://github.com/dotnet/efcore/issues/21899

【讨论】：