Skip to content

Commit 71cadb1

Browse files
CornerstoneCodeCornerstoneCode
committed
Add channel-based write queue for SQLite concurrency and stress tests
This commit fundamentally refactors the internal write serialization mechanism for SQLite from `SemaphoreSlim` to `System.Threading.Channels.Channel`. This change allows for: - Improved write throughput and better resource utilization by processing writes through a single background task per database file. - Configurable back-pressure via the new `WriteQueueCapacity` option, enabling bounded queues to manage write load. - Broader compatibility for the core queue logic by multi-targeting the library to `netstandard2.0`. Additionally, a new test project `EFCore.Sqlite.Concurrency.Test` has been introduced. This project contains comprehensive stress tests that verify the correctness and robustness of all concurrent write operations, including `SaveChangesSerializedAsync`, `BulkInsertOptimizedAsync`, and `ThreadSafeSqliteContext.ExecuteWriteAsync`, as well as mixed read/write scenarios. These tests run dozens of concurrent writers and readers to assert data integrity under heavy load.
1 parent f4d8a0a commit 71cadb1

19 files changed

Lines changed: 881 additions & 209 deletions

‎.gitignore‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,3 +12,7 @@ EntityFrameworkCore.Sqlite.Concurrency/bin/
1212
Tests/bin/
1313

1414
Tests/obj/
15+
16+
EFCore.Sqlite.Concurrency.Test/bin/
17+
18+
EFCore.Sqlite.Concurrency.Test/obj/

‎EFCore.Sqlite.Concurrency.Test/ConcurrencyStressTests.cs‎

Lines changed: 188 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,188 @@
1+
using EntityFrameworkCore.Sqlite.Concurrency;
2+
using Microsoft.EntityFrameworkCore;
3+
4+
namespace EFCore.Sqlite.Concurrency.Test;
5+
6+
/// <summary>
7+
/// Stress tests that verify correctness under concurrent write load.
8+
/// Each test owns an isolated temp SQLite file and runs 50–100 concurrent writers.
9+
/// Assertions verify that no rows are lost, duplicated, or corrupted.
10+
/// </summary>
11+
public class ConcurrencyStressTests
12+
{
13+
// ── Test 1: SaveChangesSerializedAsync ────────────────────────────────────
14+
15+
[Fact]
16+
public async Task SaveChangesSerializedAsync_WritesAllRows()
17+
{
18+
const int writers = 50;
19+
const int rowsPerWriter = 10;
20+
const int expected = writers * rowsPerWriter;
21+
22+
using var db = new TempDatabase();
23+
24+
var tasks = Enumerable.Range(0, writers).Select(async writerIdx =>
25+
{
26+
await using var ctx = db.CreateContext();
27+
28+
var entities = Enumerable.Range(0, rowsPerWriter)
29+
.Select(i => new StressEntity
30+
{
31+
Id = Guid.NewGuid(),
32+
WriterIndex = writerIdx,
33+
Payload = $"w{writerIdx}-r{i}"
34+
})
35+
.ToList();
36+
37+
ctx.Entities.AddRange(entities);
38+
await ctx.SaveChangesSerializedAsync();
39+
});
40+
41+
await Task.WhenAll(tasks);
42+
43+
await using var verify = db.CreateContext();
44+
var all = await verify.Entities.ToListAsync();
45+
var ids = all.Select(e => e.Id).ToHashSet();
46+
47+
Assert.Equal(expected, all.Count);
48+
Assert.Equal(expected, ids.Count); // no duplicate IDs
49+
}
50+
51+
// ── Test 2: BulkInsertOptimizedAsync ─────────────────────────────────────
52+
53+
[Fact]
54+
public async Task BulkInsertOptimizedAsync_WritesAllRows()
55+
{
56+
const int writers = 10;
57+
const int entitiesPerWriter = 50;
58+
const int expected = writers * entitiesPerWriter;
59+
60+
using var db = new TempDatabase();
61+
62+
var tasks = Enumerable.Range(0, writers).Select(async writerIdx =>
63+
{
64+
await using var ctx = db.CreateContext();
65+
66+
var entities = Enumerable.Range(0, entitiesPerWriter)
67+
.Select(i => new StressEntity
68+
{
69+
Id = Guid.NewGuid(),
70+
WriterIndex = writerIdx,
71+
Payload = $"bulk-w{writerIdx}-r{i}"
72+
})
73+
.ToList();
74+
75+
await ctx.BulkInsertOptimizedAsync(entities);
76+
});
77+
78+
await Task.WhenAll(tasks);
79+
80+
await using var verify = db.CreateContext();
81+
var all = await verify.Entities.ToListAsync();
82+
var ids = all.Select(e => e.Id).ToHashSet();
83+
84+
Assert.Equal(expected, all.Count);
85+
Assert.Equal(expected, ids.Count);
86+
}
87+
88+
// ── Test 3: ThreadSafeSqliteContext.ExecuteWriteAsync ────────────────────
89+
90+
[Fact]
91+
public async Task ExecuteWriteAsync_ThreadSafeSqliteContext_WritesAllRows()
92+
{
93+
const int writers = 50;
94+
const int rowsPerWriter = 10;
95+
const int expected = writers * rowsPerWriter;
96+
97+
using var db = new TempDatabase();
98+
99+
var tasks = Enumerable.Range(0, writers).Select(async writerIdx =>
100+
{
101+
// ThreadSafeSqliteContext must be constructed with the connection string
102+
// so it can find the shared write queue for this database file.
103+
var ctx = new ThreadSafeSqliteStressContext(db.ConnectionString);
104+
await using (ctx)
105+
{
106+
await ctx.ExecuteWriteAsync(async c =>
107+
{
108+
var entities = Enumerable.Range(0, rowsPerWriter)
109+
.Select(i => new StressEntity
110+
{
111+
Id = Guid.NewGuid(),
112+
WriterIndex = writerIdx,
113+
Payload = $"tssc-w{writerIdx}-r{i}"
114+
});
115+
116+
c.Entities.AddRange(entities);
117+
});
118+
}
119+
});
120+
121+
await Task.WhenAll(tasks);
122+
123+
await using var verify = db.CreateContext();
124+
var all = await verify.Entities.ToListAsync();
125+
var ids = all.Select(e => e.Id).ToHashSet();
126+
127+
Assert.Equal(expected, all.Count);
128+
Assert.Equal(expected, ids.Count);
129+
}
130+
131+
// ── Test 4: Mixed concurrent reads + writes ───────────────────────────────
132+
133+
[Fact]
134+
public async Task MixedReadsAndWrites_NoExceptionsOrDataLoss()
135+
{
136+
const int writers = 50;
137+
const int readers = 20;
138+
const int rowsPerWriter = 10;
139+
const int expected = writers * rowsPerWriter;
140+
141+
using var db = new TempDatabase();
142+
143+
// Pre-insert one seed row so readers have something to count from the start.
144+
await using (var seed = db.CreateContext())
145+
{
146+
seed.Entities.Add(new StressEntity { Payload = "seed" });
147+
await seed.SaveChangesSerializedAsync();
148+
}
149+
150+
var writerTasks = Enumerable.Range(0, writers).Select(async writerIdx =>
151+
{
152+
await using var ctx = db.CreateContext();
153+
var entities = Enumerable.Range(0, rowsPerWriter)
154+
.Select(i => new StressEntity
155+
{
156+
Id = Guid.NewGuid(),
157+
WriterIndex = writerIdx,
158+
Payload = $"mixed-w{writerIdx}-r{i}"
159+
})
160+
.ToList();
161+
162+
ctx.Entities.AddRange(entities);
163+
await ctx.SaveChangesSerializedAsync();
164+
});
165+
166+
var readResults = new System.Collections.Concurrent.ConcurrentBag<int>();
167+
var readerTasks = Enumerable.Range(0, readers).Select(async _ =>
168+
{
169+
await using var ctx = db.CreateContext();
170+
// WAL mode allows reads to run concurrently with writes.
171+
var count = await ctx.Entities.CountAsync();
172+
readResults.Add(count);
173+
});
174+
175+
await Task.WhenAll(writerTasks.Concat(readerTasks));
176+
177+
// Verify all writes landed
178+
await using var verify = db.CreateContext();
179+
var all = await verify.Entities.ToListAsync();
180+
181+
// +1 for the seed row
182+
Assert.Equal(expected + 1, all.Count);
183+
Assert.Equal(expected + 1, all.Select(e => e.Id).Distinct().Count());
184+
185+
// All reader results must be non-negative (reads always saw valid data)
186+
Assert.All(readResults, count => Assert.True(count >= 0));
187+
}
188+
}

‎EFCore.Sqlite.Concurrency.Test/EFCore.Sqlite.Concurrency.Test.csproj‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
<Project Sdk="Microsoft.NET.Sdk">
2+
3+
<PropertyGroup>
4+
<TargetFramework>net10.0</TargetFramework>
5+
<ImplicitUsings>enable</ImplicitUsings>
6+
<Nullable>enable</Nullable>
7+
<IsPackable>false</IsPackable>
8+
</PropertyGroup>
9+
10+
<ItemGroup>
11+
<PackageReference Include="coverlet.collector" Version="10.0.1">
12+
<PrivateAssets>all</PrivateAssets>
13+
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
14+
</PackageReference>
15+
<PackageReference Include="Microsoft.NET.Test.Sdk" Version="18.7.0" />
16+
<PackageReference Include="xunit" Version="2.9.3"/>
17+
<PackageReference Include="xunit.runner.visualstudio" Version="3.1.5">
18+
<PrivateAssets>all</PrivateAssets>
19+
<IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
20+
</PackageReference>
21+
</ItemGroup>
22+
23+
<ItemGroup>
24+
<ProjectReference Include="..\EntityFrameworkCore.Sqlite.Concurrency\EFCore.Sqlite.Concurrency.csproj" />
25+
</ItemGroup>
26+
27+
<ItemGroup>
28+
<Using Include="Xunit"/>
29+
</ItemGroup>
30+
31+
</Project>

‎EFCore.Sqlite.Concurrency.Test/StressDbContext.cs‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
using EntityFrameworkCore.Sqlite.Concurrency;
2+
using EntityFrameworkCore.Sqlite.Concurrency.Models;
3+
using Microsoft.EntityFrameworkCore;
4+
5+
namespace EFCore.Sqlite.Concurrency.Test;
6+
7+
public class StressDbContext : DbContext
8+
{
9+
private readonly string _connectionString;
10+
11+
public StressDbContext(string connectionString)
12+
{
13+
_connectionString = connectionString;
14+
}
15+
16+
public DbSet<StressEntity> Entities => Set<StressEntity>();
17+
18+
protected override void OnConfiguring(DbContextOptionsBuilder options) =>
19+
options.UseSqliteWithConcurrency(_connectionString);
20+
21+
protected override void OnModelCreating(ModelBuilder modelBuilder) =>
22+
modelBuilder.Entity<StressEntity>(e =>
23+
{
24+
e.HasKey(x => x.Id);
25+
e.Property(x => x.Payload).HasMaxLength(256);
26+
});
27+
}

‎EFCore.Sqlite.Concurrency.Test/StressEntity.cs‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
namespace EFCore.Sqlite.Concurrency.Test;
2+
3+
public class StressEntity
4+
{
5+
public Guid Id { get; set; } = Guid.NewGuid();
6+
public int WriterIndex { get; set; }
7+
public string Payload { get; set; } = string.Empty;
8+
}

‎EFCore.Sqlite.Concurrency.Test/TempDatabase.cs‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
namespace EFCore.Sqlite.Concurrency.Test;
2+
3+
/// <summary>
4+
/// Creates an isolated SQLite temp file for a single test. Deleted on Dispose.
5+
/// </summary>
6+
public sealed class TempDatabase : IDisposable
7+
{
8+
private readonly string _dbPath;
9+
private bool _disposed;
10+
11+
public string ConnectionString { get; }
12+
13+
public TempDatabase()
14+
{
15+
_dbPath = Path.Combine(Path.GetTempPath(), $"stress_{Guid.NewGuid():N}.db");
16+
ConnectionString = $"Data Source={_dbPath}";
17+
18+
// Create schema
19+
using var ctx = CreateContext();
20+
ctx.Database.EnsureCreated();
21+
}
22+
23+
public StressDbContext CreateContext() => new(ConnectionString);
24+
25+
public void Dispose()
26+
{
27+
if (_disposed) return;
28+
_disposed = true;
29+
30+
// Allow EF Core connection pool to drain before deleting the file.
31+
Thread.Sleep(50);
32+
33+
foreach (var ext in new[] { "", "-wal", "-shm" })
34+
{
35+
var path = _dbPath + ext;
36+
if (File.Exists(path))
37+
try { File.Delete(path); } catch { /* best-effort */ }
38+
}
39+
}
40+
}

‎EFCore.Sqlite.Concurrency.Test/ThreadSafeSqliteStressContext.cs‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
using EntityFrameworkCore.Sqlite.Concurrency;
2+
using Microsoft.EntityFrameworkCore;
3+
4+
namespace EFCore.Sqlite.Concurrency.Test;
5+
6+
/// <summary>
7+
/// Concrete subclass of <see cref="ThreadSafeSqliteContext{TContext}"/> for stress tests.
8+
/// </summary>
9+
public class ThreadSafeSqliteStressContext : ThreadSafeSqliteContext<ThreadSafeSqliteStressContext>
10+
{
11+
public ThreadSafeSqliteStressContext(string connectionString) : base(connectionString) { }
12+
13+
public DbSet<StressEntity> Entities => Set<StressEntity>();
14+
15+
protected override void OnModelCreating(ModelBuilder modelBuilder) =>
16+
modelBuilder.Entity<StressEntity>(e =>
17+
{
18+
e.HasKey(x => x.Id);
19+
e.Property(x => x.Payload).HasMaxLength(256);
20+
});
21+
}

‎EntityFrameworkCore.Sqlite.Concurrency.sln‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,8 @@ VisualStudioVersion = 17.0.31903.59
55
MinimumVisualStudioVersion = 10.0.40219.1
66
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "EFCore.Sqlite.Concurrency", "EntityFrameworkCore.Sqlite.Concurrency\EFCore.Sqlite.Concurrency.csproj", "{CDD465FA-9DE5-48C7-931E-5D3B0A9AEED5}"
77
EndProject
8+
Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "EFCore.Sqlite.Concurrency.Test", "EFCore.Sqlite.Concurrency.Test\EFCore.Sqlite.Concurrency.Test.csproj", "{5EFE0CEB-C626-44AF-A963-D437A9816B2B}"
9+
EndProject
810
Global
911
GlobalSection(SolutionConfigurationPlatforms) = preSolution
1012
Debug|Any CPU = Debug|Any CPU
@@ -18,5 +20,9 @@ Global
1820
{CDD465FA-9DE5-48C7-931E-5D3B0A9AEED5}.Debug|Any CPU.Build.0 = Debug|Any CPU
1921
{CDD465FA-9DE5-48C7-931E-5D3B0A9AEED5}.Release|Any CPU.ActiveCfg = Release|Any CPU
2022
{CDD465FA-9DE5-48C7-931E-5D3B0A9AEED5}.Release|Any CPU.Build.0 = Release|Any CPU
23+
{5EFE0CEB-C626-44AF-A963-D437A9816B2B}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
24+
{5EFE0CEB-C626-44AF-A963-D437A9816B2B}.Debug|Any CPU.Build.0 = Debug|Any CPU
25+
{5EFE0CEB-C626-44AF-A963-D437A9816B2B}.Release|Any CPU.ActiveCfg = Release|Any CPU
26+
{5EFE0CEB-C626-44AF-A963-D437A9816B2B}.Release|Any CPU.Build.0 = Release|Any CPU
2127
EndGlobalSection
2228
EndGlobal

0 commit comments

Comments
 (0)