← Барлық мақалаларға оралу

Race condition түзету - SemaphoreSlim in Action

Жарияланды

Серияның 2-посты: Advanced C# for Your Next Interview

Қай жерде тоқтадық

Алдыңғы post ішінде ең simple file storage құрдық. Whole file оқимыз, deserialize жасаймыз, record қосамыз, қайта serialize етеміз, whole file жазамыз. Clean, readable, fully async.

Содан кейін 20 parallel writes іске қосып, мынаны алдық:

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

Кінәлі read-modify-write race condition болды. Барлық tasks file-ды ешқайсысы write back жасап үлгермей тұрып оқыды, сондықтан бәрі empty list-тен бастады. Әрқайсысы тек өз record-ын жазды. Last writer won, қалғандары silently discarded болды немесе OS басқа task ұстап тұрған file-ды ашудан бас тартқанда IOException алды.

Бүгін осыны түзетеміз.

Бізге не керек

Write operation үшін mutual exclusion керек. Термин күрделі, идея simple: бір уақытта тек бір task толық read-modify-write cycle орындауы керек. Қалғандары line ішінде wait етуі тиіс.

Synchronous code ішінде lock қолданар едіңіз. Бірақ lock await-пен жұмыс істемейді, өйткені async operation бойы lock ұстап тұруға болмайды. Осы жерде SemaphoreSlim керек.

SemaphoreSlim - basics

SemaphoreSlim - async-friendly synchronization primitive. Ол maximum value бар counter сияқты жұмыс істейді. Task WaitAsync() шақырғанда counter біреуге азаяды. Release() шақырғанда counter қайта өседі. Егер counter already zero болса, WaitAsync() біреу release жасағанша waits етеді.

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

(1, 1) арқылы binary semaphore аламыз, сондықтан inside бір уақытта тек бір task болады. Бұл mutex-ке equivalent, бірақ await-пен жұмыс істейді.

lock-тан key difference:

// 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();
}

Fix

Міне updated 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() емес - sync version Wait() waiting кезінде thread block етеді. WaitAsync() thread-ті pool-ға қайтарады және semaphore available болғанда resumes етеді. Async code ішінде always async version қолданыңыз.

  2. finally block optional емес - try ішінде бірдеңе throw етсе де, lock release жасауымыз керек. finally болмаса, бір failed write semaphore-ды forever zero күйінде қалдырады. Келесі writes мәңгі wait етеді. Application hang болады.

  3. ct мәнін WaitAsync ішіне береміз - operation line ішінде waiting кезінде cancelled болса, WaitAsync OperationCanceledException throw етеді және semaphore acquired болмайды. Бұл correct behavior. Release қажет емес, өйткені біз inside кірген жоқпыз.

Сол 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 list өсіп жатқанын көреді. Collisions жоқ, exceptions жоқ, data loss жоқ.

Reads ше?

Біз тек writes lock етеміз. Reads semaphore-сыз еркін run бола алады:

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);
}

Бұл intentional. Multiple readers file-ға simultaneously access ете алады, risk жоқ. Reading ештеңені modify етпейді. Тек writes serialized болуы керек.

Strict read-write isolation керек болса, мысалы read partially written file көрмейтініне guarantee қажет болса, ReaderWriterLockSlim қолдануға болады. Ол multiple concurrent readers рұқсат етеді, бірақ writers үшін exclusive access береді. Interview үшін білуге тұрарлық, бірақ біздің мақсатқа writes үстіндегі SemaphoreSlim жеткілікті.

SemaphoreSlim vs басқа primitives

Бұл interviews кезінде жиі шығады:

lock - тек synchronous, await-пен қолдануға болмайды. Short CPU-bound critical sections үшін жақсы.

SemaphoreSlim(1, 1) - async-friendly mutex. Async operation қорғау керек болғанда қолданыңыз. Біз қолданған осы.

SemaphoreSlim(n, n) - n concurrent tasks дейін рұқсат етеді. Throttling үшін пайдалы, мысалы max 5 concurrent HTTP requests.

ReaderWriterLockSlim - read және write access ажыратады. Multiple readers бір уақытта allowed, writers exclusive access алады. Reads frequent және writes rare болғанда жақсы. Native async support жоқ, сондықтан wrapper керек.

Mutex - system-level, processes арасында жұмыс істейді. Heavy. Тек 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 whole file memory ішіне оқиды, оны list-ке deserialize етеді, бір item қосады, бәрін қайта serialize жасап, file-ды scratch-тен қайта жазады. 1 000 records болса, бұл көп allocations. 100 000 records болса, serious problem болады.

Келесі post ішінде allocation side-қа шабуыл жасай бастаймыз. Append-only format-қа ауысып, unnecessary memory pressure жою үшін Span<T> және ArrayPool<T> енгіземіз.