← Назад ко всем статьям

Исправляем race condition - SemaphoreSlim в действии

Опубликовано

Пост 2 из серии: Advanced C# for Your Next Interview

Где мы остановились

В предыдущем посте мы построили максимально простое file storage. Прочитать весь файл, deserialize, добавить record, serialize обратно, записать весь файл. Чисто, читаемо, полностью async.

Потом мы запустили 20 parallel writes и получили это:

Rows in file : 1
Lost         : 16
PROBLEM - data lost

Причиной была read-modify-write race condition. Все tasks прочитали файл до того, как какая-либо из них записала результат обратно, поэтому все начали с пустого списка. Каждая записала только свой record. Последний writer победил, все остальные были тихо отброшены или получили IOException, когда OS отказалась открыть файл, уже занятый другой task.

Сегодня мы это исправим.

Что нам нужно

Нам нужна mutual exclusion на write operation. Красивый термин, простая идея: только одной task должно быть разрешено выполнять полный read-modify-write cycle за раз. Остальные должны ждать в очереди.

В synchronous code вы бы использовали lock. Но lock не работает с await, потому что нельзя удерживать lock через async operation. Здесь и нужен SemaphoreSlim.

SemaphoreSlim - основы

SemaphoreSlim - async-friendly synchronization primitive. Он работает как счетчик с максимальным значением. Когда task вызывает WaitAsync(), счетчик уменьшается на один. Когда вызывает Release(), счетчик увеличивается обратно. Если счетчик уже равен нулю, WaitAsync() ждет, пока кто-то не освободит semaphore.

var semaphore = new SemaphoreSlim(1, 1);
//                               ^  ^
//                               |  maximum count
//                               initial count

С (1, 1) мы получаем binary semaphore, поэтому внутри одновременно может быть только одна task. Это эквивалент mutex, но работающий с await.

Ключевое отличие от lock:

// This is WRONG - you cannot await inside lock
lock (_syncObj)
{
    await File.WriteAllTextAsync(...); // compiler error
}

// This works
await _semaphore.WaitAsync();
try
{
    await File.WriteAllTextAsync(...); // perfectly fine
}
finally
{
    _semaphore.Release();
}

Исправление

Вот обновленный WriteAsync:

public class ConcurrentFileStorage : IFileStorage<FileRecord>
{
    private readonly string _filePath;
    private readonly SemaphoreSlim _writeLock = new(1, 1);

    public async Task WriteAsync(FileRecord record, CancellationToken ct = default)
    {
        await _writeLock.WaitAsync(ct);
        try
        {
            var json = await File.ReadAllTextAsync(_filePath, ct);
            var records = JsonSerializer.Deserialize<List<FileRecord>>(json) ?? [];

            records.Add(record);

            await File.WriteAllTextAsync(
                _filePath,
                JsonSerializer.Serialize(records),
                ct);
        }
        finally
        {
            _writeLock.Release();
        }
    }
}

Три вещи, на которые стоит обратить внимание:

  1. WaitAsync(ct), а не Wait() - synchronous version Wait() блокировала бы thread во время ожидания. WaitAsync() возвращает thread обратно в pool и продолжает выполнение, когда semaphore доступен. В async code всегда используйте async version.

  2. Блок finally обязателен - если внутри try что-то выбросит exception, мы все равно должны освободить lock. Без finally один failed write оставит semaphore в нуле навсегда. Все будущие writes будут ждать бесконечно. Приложение зависнет.

  3. Мы передаем ct в WaitAsync - если operation cancelled во время ожидания в очереди, WaitAsync выбросит OperationCanceledException, и semaphore не будет захвачен. Это правильное поведение. Release не нужен, потому что мы не вошли внутрь.

Запускаем тот же stress test

var tasks = Enumerable.Range(0, 20).Select(i =>
    storage.WriteAsync(
        new FileRecord(Guid.NewGuid(), $"Record-{i}", $"Data-{i}", DateTime.UtcNow))
).ToList();

await Task.WhenAll(tasks);

Output:

[Thread 04] WRITE - Record-3  (list: 2 records)
[Thread 06] WRITE - Record-11 (list: 3 records)
[Thread 04] WRITE - Record-7  (list: 4 records)
[Thread 09] WRITE - Record-0  (list: 5 records)
...

Rows in file : 20
Lost         : 0
OK

Каждая task видит, что список растет. Никаких collisions, exceptions и data loss.

А что с reads?

Мы блокируем только writes. Reads могут выполняться без semaphore:

public async Task<FileRecord?> FindAsync(Guid id, CancellationToken ct = default)
{
    // no lock here
    var json = await File.ReadAllTextAsync(_filePath, ct);
    var records = JsonSerializer.Deserialize<List<FileRecord>>(json) ?? [];
    return records.FirstOrDefault(r => r.Id == id);
}

Это сделано намеренно. Multiple readers могут обращаться к файлу одновременно без риска. Reading ничего не изменяет. Сериализовать нужно только writes.

Если нужна строгая read-write isolation, например чтобы гарантировать, что read никогда не увидит частично записанный файл, можно использовать ReaderWriterLockSlim. Он разрешает multiple concurrent readers, но дает exclusive access для writers. Это стоит знать для интервью, но для наших целей SemaphoreSlim на writes достаточно.

SemaphoreSlim vs другие primitives

Это часто всплывает на interviews:

lock - только synchronous, нельзя использовать с await. Хорош для коротких CPU-bound critical sections.

SemaphoreSlim(1, 1) - async-friendly mutex. Используйте, когда нужно защитить async operation. Именно это мы использовали.

SemaphoreSlim(n, n) - разрешает до n concurrent tasks. Полезно для throttling, например максимум 5 concurrent HTTP requests.

ReaderWriterLockSlim - различает read и write access. Multiple readers разрешены одновременно, writers получают exclusive access. Хорош, когда reads частые, а writes редкие. Native async support нет, поэтому нужен wrapper.

Mutex - system-level, работает между processes. Тяжелый. Используйте только когда нужна cross-process synchronization.

Что все еще не так

Lock решает data corruption. Но посмотрите, что все еще делает каждая operation:

var json = await File.ReadAllTextAsync(_filePath, ct);              // reads ENTIRE file
var records = JsonSerializer.Deserialize<List<FileRecord>>(json);   // allocates a list
records.Add(record);
await File.WriteAllTextAsync(_filePath, JsonSerializer.Serialize(records), ct); // writes ENTIRE file

Каждый write читает весь файл в memory, deserializes его в list, добавляет один item, serializes все обратно и полностью перезаписывает файл. С 1 000 records это уже много allocations. С 100 000 records это становится серьезной проблемой.

В следующем посте мы начнем атаковать сторону allocations. Перейдем на append-only format и введем Span<T> и ArrayPool<T>, чтобы убрать лишнее memory pressure.