logo MSJO.kr

C#, IAsyncEnumerable yield return

2025-01-04
MsJ
 

데이터베이스는 기본적으로 ‘select’와 같이 조회했을 때 그 결과가 운반 단위(array)에 도달하면 호출한 쪽으로 그 결과를 바로 리턴해준다. 예를 들어 어떤 테이블에 row 개수가 1000만 개가 있든 10만 개가 있든지 상관없이 select * from tablename을 실행하면 바로 화면에 나와야 정상이다.

이 부분을 간과하고 조회하는 프로그램 작성한다고 하면 조회 결과를 어떠한 DataSet에 넣고 그 결과가 완료될 때 비로서 Loop을 사용하여 화면에 표현하기 때문에 row 개수에 영향을 받는다. 물론, 페이징 처리를 한다고 하지만 쿼리문 자체에 ‘order by’, ‘group by’ 등을 사용하면 Sort가 발생하고 이 모든 정렬이 끝나야 비로소 그 결과를 출력하기 시작한다. 즉, ‘부분범위처리’와 ‘전체범위처리’의 차이다.

부분범위처리’라 함은 데이터베이스가 해당 내용을 다 읽지 않고도 바로 하나의 row를 바로 출력할 수 있는 상태를 말한다. 옵티마이저가 봤을 때 끝까지 읽어서 분석할 필요가 없다고 판단하기 때문이다. 그래서 정렬하되 전체범위처리(끝까지 다 읽고나서야 결과를 뽑아낼수 있는 상태)가 되지 않도록 쿼리문을 잘 작성해야 한다.

여기에는 인덱스 전략도 포함되며 특히, ‘where’절에 column을 가공하면(예, where left(xxxx) = '1234') 작성한 ‘left(xxxx)’라는 column은 존재하지 않기 때문에 옵티마이저는 전체를 다 읽고 해당 column을 ‘left(xxxx)’로 모두 2차 가공한 다음에 조회하고 그 결과를 출력한다. 즉, 전체범위처리를 하는 것이다. 요지는 ‘어떻게 부분범위처리가 되도록 유도하느냐’이다.

부분범위처리가 되었다고 가정하고 조회 결과를 클라이언트의 프로그램에서 조회한 결과를 DataSet에 모두 넣고 그 다음에 화면에 Loop를 사용하여 표현하는 방법으로 처리하는 것이 아닌 비동기적으로 스트리밍하여 그 결과가 들어오는 즉시 화면에 출력하도록 IAsyncEnumerable, yield return을 사용하는 예제를 작성해 보았다. SQL Server의 특정 테이블에 50만 개 정도의 row가 저장된 테스트 데이터를 사용한다.

nuget 패키지 추가
<Project Sdk="Microsoft.NET.Sdk">  
  
    <PropertyGroup>  
        <OutputType>Exe</OutputType>  
        <TargetFramework>net8.0</TargetFramework>  
        <Nullable>enable</Nullable>  
    </PropertyGroup>  
  
    <ItemGroup>  
        <PackageReference Include="Dapper" Version="2.1.35"/>  
        <PackageReference Include="Microsoft.Data.SqlClient" Version="6.0.0-preview3.24332.3"/>  
        <PackageReference Include="System.Linq.Async" Version="6.0.1" />  
    </ItemGroup>  
  
</Project>
데이터 모델 정의
// 데이터베이스 table 중에서 특정 컬럼만 조회
public class MyData  
{  
    public string ZipCode { get; init; } = string.Empty;  
    public string SiDo { get; init; } = string.Empty;  
}
데이터베이스 처리 로직
using Dapper;
using Microsoft.Data.SqlClient;
using System.Collections.Generic;
using System.Data;
using System.Linq;


namespace Hello;

public class DatabaseContext
{
    public const string M_CON = "Server = 192.168.1.2; Uid = sa; Pwd = xxxxx; database = TestDatabase; TrustServerCertificate = True";

    public static async IAsyncEnumerable<MyData> GetDataAsync()
    {
        const string sql = "select s.ZIP_CODE as ZipCode, s.SIDO as SiDo  from COMMON.ZIP_CODE_SEOUL s";
        await using SqlConnection connection = new(M_CON);

        // IAsyncEnumerable<IEnumerable<MyData>> result = connection.QueryAsync<MyData>(sql, commandType: CommandType.Text).ToAsyncEnumerable();
        //
        // await foreach (IEnumerable<MyData> items in result.AsAsyncEnumerable())
        // {
        //     foreach (MyData item in items)
        //     {
        //         yield return item;
        //     }
        // }

        // IEnumerable<MyData> results = await connection.QueryAsync<MyData>(sql, commandType: CommandType.Text);
        //
        // await foreach (MyData data in results.ToAsyncEnumerable())
        // {
        //     yield return data;
        // }

        IEnumerable<MyData> results = await connection.QueryAsync<MyData>(sql, commandType: CommandType.Text);

        foreach (MyData result in results)
        {
            yield return result;
        }
    }
}
테스트
using System;  
using System.Diagnostics;  
using System.Linq;  
using System.Threading.Tasks;  
  
namespace Hello;  
  
internal class Program  
{  
    private static async Task Main()  
    {  
        Stopwatch sw = Stopwatch.StartNew();  
        sw.Start();  
  
        await foreach ((MyData data, int i) in DatabaseContext.GetDataAsync().Select((v, i) => (v, i)))  
        {  
            Console.WriteLine($"{i + 1} : {data.ZipCode}, {data.SiDo}");  
        }  
  
        sw.Stop();  
        Console.WriteLine($"{sw.Elapsed.Seconds} 초");  
    }  
}

위와 같이 작성하고 실행하면 실행하자마자 화면에 출력하는 내용을 볼 수 있다.

CancellationToken 사용 예제

CancellationTokenSource를 사용하여 비동기 작업을 취소하여 데이터 읽기 작업을 중간에 중단한다.

public static async IAsyncEnumerable<MyData> GetDataAsync([EnumeratorCancellation] CancellationToken cancellationToken)
{
    const string sql = "select s.ZIP_CODE as ZipCode, s.SIDO as SiDo  from COMMON.ZIP_CODE_SEOUL s";
    await using SqlConnection connection = new(M_CON);

    IEnumerable<MyData> results = await connection.QueryAsync<MyData>(sql, commandType: CommandType.Text);

    foreach (MyData result in results)
    {
        cancellationToken.ThrowIfCancellationRequested();
        yield return result;
    }
}

private static async Task Main()
{
    Stopwatch sw = Stopwatch.StartNew();
    sw.Start();

    using CancellationTokenSource cts = new();

	// 해당 작업은 5초 후에 종료한다
    cts.CancelAfter(TimeSpan.FromSeconds(5));

    try
    {
        await foreach ((MyData data, int i) in DatabaseContext.GetDataAsync(cts.Token).Select((v, i) => (v, i)).WithCancellation(cts.Token))
        {
            Console.WriteLine($"{i + 1} : {data.ZipCode}, {data.SiDo}");
        }
    }
    catch (Exception ex)
    {
        Console.WriteLine(ex.Message);
    }

    sw.Stop();
    Console.WriteLine($"{sw.Elapsed.Seconds} 초");
}
실제 스트리밍 부분범위처리
// connection timout = 3초, 예외처리 생략
using CancellationTokenSource cts = new(TimeSpan.FromSeconds(3));
await using SqlConnection connection = new(M_CON);
await connection.OpenAsync(cts.Token).ConfigureAwait(false);

CommandDefinition cmd = new(sql, commandType: CommandType.Text, cancellationToken: cancellationToken);
await using DbDataReader reader = await connection.ExecuteReaderAsync(cmd);
Func<DbDataReader, TestModel> parser = reader.GetRowParser<TestModel>();

while (await reader.ReadAsync(cancellationToken).ConfigureAwait(false))
{
    yield return parser(reader);
}
Row Versioning

글을 작성하고 보니 부분범위처리1와 더불어 데이터베이스별로 Row Versioning하는 방법이 달라서 대표적으로 Oracle과 비교해서 SQL Server에서 lock에 의한 경합을 다루는 트랜잭션 격리 수준2의 하나인 SNAPSHOT에 대해 소개해 볼까 한다.

오라클(MariaDB, PostgreSQL)은 기본적으로 ‘select’는 서로 경합이 없으며 for update row versioning을 할 수 있다. dirty read가 발생할 수는 있어도 일반적인 lock은 발생하지 않는다. 그러나 SQL Server는 반대 개념이라고 볼 수 있는데 기본적으로 SQL Server는 읽기 작업 시 공유 잠금을 얻어야 한다. 이 때문에 경합을 피하려면 tablename with(nolock) 힌트를 사용해야 dirty read가 가능해진다.

그래서 이 반대 상황을 이해하지 못하고 사용하면 SQL Server는 lock이 많다고 오해할 수 있는데 snpshot 격리 수준을 사용하여 트랜잭션 수준에서 경합을 처리하는 예제를 소개한다. 참고로 아래의 예제는 ‘Gemini 2.0’, ‘ChatGPT’ 그리고 ‘Github Copilot’에서 조회하고 테스트 하였다.

-- SNAPSHOT ISOLATION` 활성화
ALTER DATABASE YourDatabaseName
SET ALLOW_SNAPSHOT_ISOLATION ON;

-- 트랜잭션 내에서 SNAPSHOT ISOLATION 사용
SET TRANSACTION ISOLATION LEVEL SNAPSHOT;
BEGIN TRANSACTION;

-- 데이터 읽기
SELECT * FROM employees WHERE employee_id = 1;

-- 데이터 수정
UPDATE employees SET salary = salary + 1000 WHERE employee_id = 1;

COMMIT;

이렇게 하면 다른 세션의 사용자는 경합 없이 commit전 데이터를 읽을 수 있다. 이때 SQL Server tempdb를 사용한다. snapshot을 사용하려면 충분한 메모리 확보와 더불어 tempdb의 성능도 중요하다. 일반적으로 cpu 코어의 수만큼 tempdb를 만들고 생성할 때 동일한 크기로 설정하고 될 수 있으면 빠른 스토리지를 사용한다. SQL Server에서 자동으로 작업을 분산하여 tempdb를 활용한다. 오라클은 이런 경우 Undo 세그먼트를 이용한다.

nolock 활용
BEGIN TRANSACTION;

-- 임시 테이블이 존재하면 삭제
IF OBJECT_ID('tempdb..#TempEmployees') IS NOT NULL
BEGIN
    DROP TABLE #TempEmployees;
END;

-- NOLOCK 힌트를 사용하여 데이터를 읽음 (잠금 없이 읽기)
SELECT e.employee_id, e.salary
INTO #TempEmployees
FROM employees e WITH (NOLOCK)
WHERE e.department_id = 10;

-- 임시 테이블의 데이터를 기반으로 업데이트 수행
UPDATE e
SET e.salary = e.salary + t.salary * 0.1
FROM employees e
INNER JOIN #TempEmployees t ON e.employee_id = t.employee_id;

-- 트랜잭션 커밋
COMMIT;

-- 임시 테이블 삭제 (필요시)
DROP TABLE #TempEmployees;

BEGIN TRANSACTION;

-- 테이블 변수 선언
DECLARE @TempEmployees TABLE (
    employee_id INT,
    salary DECIMAL(10, 2)
);

-- NOLOCK 힌트를 사용하여 데이터를 읽음 (잠금 없이 읽기)
INSERT INTO @TempEmployees (employee_id, salary)
SELECT e.employee_id, e.salary
FROM employees e WITH (NOLOCK)
WHERE e.department_id = 10;

-- 테이블 변수의 데이터를 기반으로 업데이트 수행
UPDATE e
SET e.salary = e.salary + t.salary * 0.1
FROM employees e
INNER JOIN @TempEmployees t ON e.employee_id = t.employee_id;

-- 트랜잭션 커밋
COMMIT;

DECLARE @TempEmployees TABLE () 형태로 테이블 변수를 사용하면 작은 데이터 집합의 경우 테이블 변수가 더 빠를 수 있지만, 큰 데이터 집합에서는 임시 테이블이 더 나은 성능을 제공할 수 있다. 또한 테이블 변수를 사용하면 통계 처리가 되지 않는다.

‘SNAPSHOT ISOLATION’을 활성화한 후에 각 쿼리에서 트랜잭션을 처리 할 때마다 SET TRANSACTION ISOLATION LEVEL SNAPSHOT;을 사용해야 하는 약간의 불편함이 발생할 수 있다. 이 외에도 read committed with snapshot을 사용할 수 있다.3

Reference
  1. msjo.kr, “업무용 프로그래밍에서 개발자의 자질과 덕목”에서 소개한 ‘오라클 성능 고화화 원리와 해법’, ‘SQL Server 튜닝 원리와 해법’은 필수적으로 읽어봐야 할 서적이다.
  2. 정재우, SQL Server 튜닝 원리와 해법, 비투엔컨설팅, 2010, p431
  3. learn.microsoft.com, “SET TRANSACTION ISOLATION LEVEL (Transact-SQL)”
Prεv(Θld)   Nεxt(Nεw)
Content
Search     RSS Feed     BY-NC-ND