C# Lock Free Queue : ConcurrentQueue 쓰레드 안전 큐

멀티 쓰레딩은 다양한 이점을 제공합니다. 예를 들어, 멀티 쓰레드를 사용하면 다중 코어 프로세서를 효과적으로 활용하여 프로그램의 성능을 향상시킬 수 있습니다. 또한 멀티 쓰레드를 사용하면 동시성 작업을 수행하거나 시간이 오래 걸리는 작업을 별도의 쓰레드에서 처리할 수 있어 응용 프로그램이 빠르게 응답하도록 만들 수 있습니다.
그러나 멀티 쓰레드 프로그래밍은 여러 스레드 간의 공유 데이터에 대한 일관성(Thread-Safe)을 보장해야합니다.
그 일관성을 보장해주는 컬렉션 C# Lock Free Queue 라고 불리는 ConcurrentQueue 에 대해서 알아보도록 하겠습니다.

소개

이 글에서는 C#에서 병행성을 다루는 데 도움이 되는 데이터 구조 중 하나인 ConcurrentQueue 클래스에 대해 알아보겠습니다. 우리는 ConcurrentQueue가 무엇인지, 어떻게 사용하는지, 그리고 병행 프로그래밍에 어떤 이점을 제공하는지 알아볼 것입니다.

C#에서의 병행성이란?

병행성(Concurrency)은 시스템이 여러 작업을 동시에 실행할 수 있는 능력입니다. 예를 들어, 웹 서버는 다른 클라이언트로부터의 여러 요청을 동시에 처리할 수 있으며, 비디오 게임은 동시에 그래픽을 렌더링하고 사용자 입력을 처리할 수 있습니다. 병행성은 시스템의 성능과 응답성을 향상시키고 사용 가능한 리소스를 더 효율적으로 활용할 수 있게 합니다.

그러나 병행 프로그래밍은 많은 도전 과제를 동반합니다.

• 경합 상태(Race conditions): 두 개 이상의 스레드가 동시에 동일한 공유 데이터에 액세스하거나 수정할 때 서로 간섭하여 올바르지 않거나 일관성 없는 결과를 생성할 수 있습니다.
• 데드락(Deadlocks): 두 개 이상의 스레드가 서로가 필요로 하는 리소스를 해제할 때 서로를 기다릴 때, 그들은 서로 진행할 수 없는 상황에 빠질 수 있습니다.
• 라이블락(Livelocks): 두 개 이상의 스레드가 서로에게 반응하여 반복적으로 상태를 변경할 때, 그들은 서로 진행할 수 없는 상황에 빠질 수 있습니다.
• 기아(Starvation): 스레드가 다른 스레드가 더 높은 우선순위를 가지거나 리소스를 계속 점유하기 때문에 필요한 리소스에 대한 액세스를 오랫동안 할 수 없는 상황에 빠질 수 있습니다.

이러한 문제를 피하기 위해서는 동시 코드가 스레드 안전(thread-safe)하다는 것을 보장해야 합니다. 이것은 동시에 여러 스레드에 의해 실행될 때 올바르게 작동하고 일관성 있게 작동함을 의미합니다.

스레드 안전성을 달성하는 한 가지 방법은 락(lock), 뮤텍스(mutex), 세마포어(semaphore), 모니터(monitor) 등과 같은 동기화 메커니즘을 사용하는 것입니다. 이러한 메커니즘은 여러 스레드 간의 공유 데이터의 액세스와 수정을 제어하여 한 번에 한 스레드만 데이터에 액세스하거나 수정하도록 보장합니다. 그러나 동기화에도 일부 단점이 있습니다.

• 성능 오버헤드: 동기화는 락 획득 및 해제, 스레드 간의 컨텍스트 전환에 대한 추가 비용을 도입합니다.
• 복잡성: 동기화는 락 및 리소스의 잠금 및 해제의 로직 및 순서를 주의 깊게 설계하고 구현해야 하며, 이것은 오류 발생 가능성이 있고 유지 관리하기 어려울 수 있습니다.
• 확장성: 동기화는 시스템이 달성할 수 있는 병행성의 정도를 제한하며 스레드의 병렬성과 처리량을 감소시킵니다.

스레드 안전성을 달성하는 또 다른 방법은 ConcurrentQueue와 같은 병행 데이터 구조를 사용하는 것입니다. 이러한 데이터 구조는 명시적 동기화를 필요로 하지 않고 여러 스레드에 의한 액세스 및 수정을 처리하도록 설계되었습니다. 이러한 데이터 구조는 원자적인 동작, 락을 사용하지 않는 알고리즘, 낙관적인 병행 제어(Optimistic Concurrency Control) 등과 같은 다양한 기술을 사용하여 작업이 스레드 안전하고 효율적임을 보장합니다.

ConcurrentQueue 소개

ConcurrentQueue는 원소의 FIFO(First-In-First-Out) 큐를 나타내는 데이터 구조입니다. 큐는 한쪽 끝(꼬리)에서 원소를 추가하고 다른 쪽 끝(머리)에서 원소를 제거할 수 있는 컬렉션입니다. 큐는 FIFO의 원칙을 따릅니다. 큐에 추가된 첫 번째 원소가 큐에서 제거되는 첫 번째 원소입니다.

ConcurrentQueue는 일반 큐와 두 가지 방면에서 다릅니다.

1. ConcurrentQueue는 스레드 안전(thread-safe)합니다. 이는 명시적 동기화를 필요로 하지 않고 여러 스레드가 큐에 원소를 동시에 추가하고 제거할 수 있음을 의미합니다.
2. ConcurrentQueue는 락을 사용하지 않는 lock-free입니다. 이는 락이나 다른 블로킹 메커니즘을 사용하지 않고 큐를 조작하기 위해 원자적 동작 및 비교 및 교체 알고리즘을 사용합니다.

ConcurrentQueue 사용의 이점은 다음과 같습니다.

• ConcurrentQueue는 성능을 향상시킵니다. 동기화와 컨텍스트 전환의 오버헤드를 줄이며 스레드의 병렬성과 처리량을 증가시킵니다.
• ConcurrentQueue는 코드를 단순화합니다. 명시적 동기화와 락 로직이 필요 없어 복잡하고 오류 발생 가능성이 있는 부분을 제거합니다.
• ConcurrentQueue는 확장성을 향상시킵니다. 큐의 액세스 및 수정에 대한 순서나 제한을 부과하지 않으므로 시스템에 더 많은 병렬성과 유연성을 제공합니다.

ConcurrentQueue는 다른 데이터 구조와 비교할 수 있습니다.

ConcurrentQueue 사용

ConcurrentQueue를 사용하려면 다음 네임스페이스를 코드에 추가해야 합니다.

using System.Collections.Concurrent;

그런 다음 다음 구문을 사용하여 ConcurrentQueue를 만들 수 있습니다.

ConcurrentQueue<T> queue = new ConcurrentQueue<T>();

여기서 <T>는 큐의 요소 유형입니다. 예를 들어, 다음과 같이 정수의 ConcurrentQueue를 만들 수 있습니다.

ConcurrentQueue<int> queue = new ConcurrentQueue<int>();

또한 기존 요소 컬렉션으로 ConcurrentQueue를 초기화할 수 있습니다.

ConcurrentQueue<T> queue = new ConcurrentQueue<T>(collection);

여기서 collection은 큐에 추가할 요소를 포함하는 컬렉션입니다. 예를 들어, 문자열의 ConcurrentQueue를 문자열 배열로 초기화할 수 있습니다.

string[] words = { "Hello", "World", "Welcome" };
ConcurrentQueue<string> queue = new ConcurrentQueue<string>(words);

ConcurrentQueue에 요소 추가하기

ConcurrentQueue의 꼬리에 요소를 추가하려면 다음 메서드를 사용할 수 있습니다.

queue.Enqueue(element);

여기서 element는 추가할 요소입니다. 예를 들어, 다음과 같이 큐에 정수를 추가할 수 있습니다.

queue.Enqueue(42);

이 메서드는 스레드 안전하며 락을 사용하지 않으므로 여러 스레드에서 동시에 호출하여 동기화나 블로킹을 필요로하지 않습니다.

ConcurrentQueue에서 요소 제거하기

ConcurrentQueue의 헤드에서 요소를 제거하려면 다음 메서드를 사용할 수 있습니다.

bool success = queue.TryDequeue(out element);

여기서 element는 작업이 성공하면 제거된 요소를 받는 출력 매개변수이고 success는 작업이 성공했는지 여부를 나타내는 부울 값입니다. 큐가 비어있는 경우 작업은 실패합니다. 예를 들어, 다음과 같이 큐에서 문자열을 제거할 수 있습니다.

var success = queue.TryDequeue(out var word);
Console.WriteLine(success ? word : "큐가 비어 있습니다."); // "Hello"를 출력합니다.

이 메서드 역시 스레드 안전하며 락을 사용하지 않으므로 동기화나 블로킹을 필요로하지 않습니다.

요소의 존재 여부 확인

ConcurrentQueue에 요소가 있는지 확인하려면 다음 속성을 사용할 수 있습니다.

bool empty = queue.IsEmpty;

여기서 empty는 큐가 비어 있는지 여부를 나타내는 부울 값입니다. 예를 들어, 다음과 같이 큐가 비어 있는지 확인할 수 있습니다.

Console.WriteLine(queue.IsEmpty ? "큐가 비어 있습니다." : "큐가 비어 있지 않습니다.");

이 속성 또한 스레드 안전하며 락을 사용하지 않으므로 동기화나 블로킹을 필요로하지 않습니다. 그러나 이 속성은 큐에 다른 스레드가 속성을 확인한 후 요소를 추가하거나 제거할 수 있기 때문에 큐의 가장 최근 상태를 반영하지 않을 수 있음에 유의해야 합니다.

ConcurrentQueue에 특정 요소가 포함되어 있는지 확인하려면 다음 메서드를 사용할 수 있습니다.

bool found = queue.Contains(element);

여기서 element는 검색할 요소이고 found는 요소를 찾았는지 여부를 나타내는 부울 값입니다. 예를 들어, 다음과 같이 큐가 문자열을 포함하고 있는지 확인할 수 있습니다.

var found = queue.Contains("Welcome");
Console.WriteLine(found ? "큐에 Welcome이 포함되어 있습니다." : "큐에 Welcome이 포함되어 있지 않습니다.");

이 메서드는 스레드 안전하지만 락을 사용하지 않으며 검색 작업을 수행하는 동안 다른 스레드가 요소를 추가하거나 제거하는 것을 차단할 수 있습니다. 따라서 이 메서드를 자주 사용하거나 큰 큐에서 사용하는 것은 권장되지 않습니다.

ConcurrentQueue 반복

ConcurrentQueue의 요소를 반복하려면 다음과 같이 루프를 사용할 수 있습니다.

foreach (var element in queue)
{
    // 요소를 처리합니다.
}

여기서 element는 큐의 각 요소를 받는 변수입니다. 예를 들어, 다음과 같이 큐의 모든 요소를 출력할 수 있습니다.

foreach (var word in queue)
{
    Console.WriteLine(word); // "World", "Welcome"를 출력합니다.
}

이 루프 또한 스레드 안전하지만 락을 사용하지 않으며

반복하는 동안 다른 스레드가 요소를 추가하거나 제거하는 것을 차단할 수 있습니다. 따라서 이 루프를 자주 사용하거나 큰 큐에서 사용하는 것은 권장되지 않습니다. 또한 이 루프는 시작한 후 다른 스레드가 요소를 추가하거나 제거할 수 있기 때문에 큐의 모든 요소를 반복하거나 추가 또는 제거된 순서대로 반복하는 것이 보장되지 않을 수 있음에 유의해야 합니다.

ConcurrentQueue 비우기

ConcurrentQueue에서 모든 요소를 제거하려면 다음 메서드를 사용할 수 있습니다.

queue.Clear();

이 메서드는 스레드 안전하며 락을 사용하지 않으므로 여러 스레드에서 동시에 호출하여 동기화나 블로킹을 필요로하지 않습니다. 그러나 이 메서드를 호출한 후 다른 스레드가 새로운 요소를 큐에 추가할 수 있기 때문에 메서드를 호출한 후 큐가 비어 있을 것이라고 보장되지는 않습니다.

기타 유용한 메서드와 속성

ConcurrentQueue는 몇 가지 기타 유용한 메서드와 속성도 제공합니다.

• Count 속성: 큐의 요소 수를 반환합니다. 스레드 안전하고 락을 사용하지 않지만 큐의 가장 최근 상태를 반영하지 않을 수 있습니다.
• TryPeek 메서드: 요소를 제거하지 않고 큐의 헤드에서 요소를 반환하려고 시도합니다. 스레드 안전하고 락을 사용하지 않지만 큐가 비어 있거나 다른 스레드가 요소를 제거하기 전에 실패할 수 있습니다.
• CopyTo 메서드: 큐의 요소를 배열로 복사합니다. 스레드 안전하지만 락을 사용하지 않지만 복사 중에 다른 스레드가 요소를 추가하거나 제거하는 것을 차단할 수 있습니다. 또한 모든 요소를 복사하지 않거나 추가 또는 제거된 순서대로 복사하지 않을 수 있습니다.
• ToArray 메서드: 큐의 요소를 포함하는 스냅샷을 가진 새 배열을 반환합니다. 스레드 안전하지만 락을 사용하지 않지만 배열을 만들 때 다른 스레드가 요소를 추가하거나 제거하는 것을 차단할 수 있습니다. 또한 모든 요소를 포함하지 않을 수 있거나 추가 또는 제거된 순서대로 포함하지 않을 수 있습니다.

Lock Free Queue 예제

이 코드는 ConcurrentQueue를 사용하여 여러 스레드 간에 안전하게 큐를 조작하는 방법을 보여줍니다. 큐에 데이터를 추가하는 Enqueue, 첫 번째 요소를 확인하는 TryPeek, 그리고 데이터를 추출하고 큐에서 제거하는 TryDequeue 메서드를 사용합니다. 여러 스레드가 데이터를 안전하게 조작할 수 있도록 설계되었으며, 병렬로 작업을 수행하여 결과를 안전하게 누적합니다.

using System;
using System.Collections.Concurrent;
using System.Threading;
using System.Threading.Tasks;

class CQ_EnqueueDequeuePeek
{
   // Demonstrates:
   // ConcurrentQueue<T>.Enqueue()
   // ConcurrentQueue<T>.TryPeek()
   // ConcurrentQueue<T>.TryDequeue()
   static void Main ()
   {
      // ConcurrentQueue 생성.
      ConcurrentQueue<int> cq = new ConcurrentQueue<int>();

      // 큐에 데이터 채우기.
      for (int i = 0; i < 10000; i++)
      {
          cq.Enqueue(i);
      }

      // 첫 번째 요소 확인.
      int result;
      if (!cq.TryPeek(out result))
      {
         Console.WriteLine("CQ: TryPeek가 성공해야 하는데 실패했습니다");
      }
      else if (result != 0)
      {
         Console.WriteLine("CQ: 예상한 TryPeek 결과는 0이지만 {0}을 얻었습니다", result);
      }

      int outerSum = 0;
      // ConcurrentQueue를 소비하는 작업 정의.
      Action action = () =>
      {
         int localSum = 0;
         int localValue;
         while (cq.TryDequeue(out localValue)) localSum += localValue;
         Interlocked.Add(ref outerSum, localSum);
      };

      // 4개의 병렬 작업 시작.
      Parallel.Invoke(action, action, action, action);

      Console.WriteLine("outerSum = {0}, 49995000이어야 합니다", outerSum);
   }
}

출력:

outerSum = 49995000, 49995000이어야 합니다

이 코드를 통해 동시성 프로그래밍에서 ConcurrentQueue를 어떻게 사용하는지에 대한 기본적인 이해를 얻을 수 있습니다.

결론

이 글에서는 C#의 ConcurrentQueue 클래스를 살펴보았는데, 이 클래스는 스레드 안전하고 락을 사용하지 않는 FIFO 큐를 나타내는 데이터 구조입니다. ConcurrentQueue가 무엇인지, 어떻게 사용하는지, 동시 프로그래밍에 어떻게 유용한지, 그리고 일반적으로 사용되는 경우에 대해 알아보았습니다.