24.05.26 C# 동시성 프로그래밍 (동기화)

1. 블로킹 잠금

닷넷에서는 여러 기능을 통해 잠금 기법을 제공한다. 다만 대부분의 상황에서의 잠금은 lock으로 처리가 가능하다.
단 lock을 사용함에 있어서 네 가지 지침을 확인할 필요가 있다.

• 잠금의 가시성을 제한해야 한다.
• 잠금으로 보호하는 대상을 문서화한다.
• 잠그는 코드를 최소화한다.
• 잠금을 유지하는 동안 절대로 임의의 코드를 실행하지 말아야 한다.

class MyClass
{
    private readonly object _mutex = new object();
    private int _value;
    
    public void Increment()
    {
        lock(_mutex)
        {
            _value = _value + 1;
        }
    }
}

 

2. 비동기 잠금

SemaphoreSlim과 Nito.AsyncEx의 AsyncLock을 통해서 비동기 잠금을 구현할 수 있다.
잠금을 사용한 코드는 이벤트 발생, 가상 메서드 호출, 대리자 호출 등을 포함한 임의의 코드에서 호출되면 안된다.

// SemaphoreSlim
class MyClass
{
    private readonly SemaphoreSlim _mutex = new SemaphoreSlim(1);
    private int _value;
    
    public async Task DelayAndIncrementAsync()
    {
        await _mutex.WaitAsync();
        try
        {
            int oldValue = _value;
            await Task.Delay(TimeSpan.FromSeconds(oldValue));
            _value = oldValue + 1;
        }
        finally
        {
            _mutex.Release();
        }
    }
}

//Nito.AsyncEx 
class MyClass
{
    private readonly AsyncLock _mutex = new AsyncLock();
    private int _value;
    
    public async Task DelayAndIncrementAsync()
    {
        using (await _mutex.LockAsync())
        {
            int oldValue = _value;
            await Task.Delay(TimeSpan.FromSeconds(oldValue));
            _value = oldValue + 1;
        }
    }
}

 

3. 블로킹 신호

스레드간 알림을 사용할 땐 ManualResetEventSlim을 사용한다. 수동으로 재설정하거나 모든 스레드들을 일괄처리도 가능하다.

class MyClass
{
    private readonly ManualresetEventSlim _initialized = 
        new ManualResetEventSlim();
    
    private int _value;
    
    public int WaitForInitialization()
    {
        _initialized.Wait()
        return _value;
    }
    
    public void InitialzeFromAnotherThread()
    {
        _value = 13;
        _initialized.Set();
    }
}

 

4. 비동기 신호

알림을 한 번만 보낸다면 TaskCompletionSource<T>를 사용해서 비동기적으로 알람을 보낼 수 있다.

class MyClass
{
    private readonly TaskCompletionSource<object> _initialized = 
        new TaskCompletionSource<object>();
    
    private int _value1;
    private int _value2;
    
    public async Task<int> WaitForInitializationAsync()
    {
        await _initialized.Task;
        return _value1 + _value2;
    }
    
    public void Initialize()
    {
        _value1 = 13;
        _value2 = 17;
        _initialized.TrySetResult(null);
    }
}

Nito.AsyncEx에서 비슷한 기능인 AsyncManualResetEvent가 있다.

class MyClass
{
    private readonly AsyncManualResetEvent _connected = 
        new AsyncManualResetEvent();
        
    public async Task WaitForConnectedAsync()
    {
        await _connected.WaitAsync();
    }
    
    public void ConnectedChanged(bool connected)
    {
        if (connected)
            _connected.Set();
        else    
            _connected.Reset();
    }
}

 

5. 조절

처리속도를 따라가지 못해서 데이터 항목이 많아지고 불필요한 메모리 소비가 늘어난다면 코드를 조절하여 메모리 문제를 방지할 수 있다.

IPropagatorBlock<int, int> DataflowMultiplyBy2()
{
    var options = new ExecutionDataflowBlockOptions
    {
        MaxDegreeOfParallelism = 10
    };
    
    return new TransformBlock<int, int>(data => data * 2, options);
}

// PLINQ 사용
IEnumerable<int> ParallelMultiplyBy2(IEnumerable<int> values)
{
    return values.AsParallel()
        .WithDegreeOfParallelism(10)
        .Select(item => item * 2)
}

// Parallel 클래스 사용
void ParallelRotateMatrices(IEnumerable<Matrix> matrices, float degrees)
{
    var options = new ParallelOptinos
    {
        MaxDegreeOfParallelism = 10,
    };
    Parallel.ForEach(matrices, options, matrix => matrix.Rotate(degrees));
}

// 동시 비동기 코드는 SemaphoreSlim으로 조절
async Task<string[]> DownloadUrlsAsync(HttpClient client,
    IEnumerable<string> urls)
{
    using var semaphore = new SemaphoreSlim(10);
    Task<string>[] tasks = urls.Select(async url =>
    {
        await semaphore.WaitAsync();
        try
        {
            return await client.GetStringAsync(url);
        }
        finally
        {
            semaphore.Release();
        }
    }).ToArray();
    return await Task.WhenAll(tasks);
}