Microsoft MVP성태의 닷넷 이야기
글쓴 사람
정성태 (techsharer at outlook.com)
홈페이지
첨부 파일

C# - async/await 그리고 스레드 (4) 비동기 I/O 재현

해본 김에,

C# - async/await 그리고 스레드 (1) MyTask로 재현
; https://www.sysnet.pe.kr/2/0/13055

C# - async/await 그리고 스레드 (2) MyTask의 호출 흐름
; https://www.sysnet.pe.kr/2/0/13056

C# - async/await 그리고 스레드 (3) Task.Delay 재현
; https://www.sysnet.pe.kr/2/0/13060

이제 비동기 I/O에서의 async/await도 구현해 보겠습니다. 그런데, 이것을 설명하기 위해 부가적으로 Windows의 "비동기 I/O"에 대한 기본적인 이해가 필요합니다. 그러니까... 결국 이번 주제를 위해 다음의 글이 쓰인 것입니다. ^^;

C# - CLR ThreadPool의 I/O 스레드에 작업을 맡기는 방법
; https://www.sysnet.pe.kr/2/0/13059

그리고, 사실 위의 글만 이해했다면 async/await과의 연동은 이미 설명한 것이나 다름없습니다. 왜냐하면 결국 비동기 I/O에 대한 async/await은 위의 글에서 만든 MyAsyncFileStream에 "C# - async/await 그리고 스레드 (1)" 글에서 만든 MyTask를 연결만 하면 되기 때문입니다.




예를 한번 들어볼까요? ^^ 우리는 그동안 C#에서 다음과 같은 식으로 간단하게 파일 I/O를 비동기로 접근할 수 있었습니다.

using System;

internal class Program
{
    static async Task Main(string[] args)
    {
        Console.WriteLine($"[{DateTime.Now}] 1단계: threadid == {Thread.CurrentThread.ManagedThreadId}");
        
        string filePath = @"test.txt";

        FileStream fs = new FileStream(filePath, FileMode.Open, FileAccess.Read, FileShare.None, 0, true);

        byte[] buffer = new byte[2];
        int readBytes = await fs.ReadAsync(buffer, 0, buffer.Length);
        Console.WriteLine($"[{DateTime.Now}] 2단계: {BitConverter.ToString(buffer)} threadid == {Thread.CurrentThread.ManagedThreadId}");

        readBytes = await fs.ReadAsync(buffer, 0, buffer.Length);
        Console.WriteLine($"[{DateTime.Now}] 3단계: {BitConverter.ToString(buffer)} threadid == {Thread.CurrentThread.ManagedThreadId}");
    }
}

/* 출력 결과 */
/* test.txt의 파일 내용: 1234567890

[2022-05-15 오전 9:01:43] 1단계: threadid == 1
[2022-05-15 오전 9:01:43] 2단계: 31-32 threadid == 4
[2022-05-15 오전 9:01:43] 3단계: 33-34 threadid == 4
*/

자, 이 코드와 유사한 것을 지난 글에서 만든 MyAsyncFileStream에 구현해 보겠습니다. 추가해야 할 코드는 간단합니다. MyTask 코드는 더 추가할 것이 없고, 단지 MyAsyncFileStream에 MyTask를 반환하는 ReadAsync/WriteAsync만 아래와 같이 추가해 주면 됩니다.

using Microsoft.Win32.SafeHandles;
using System.Diagnostics.CodeAnalysis;
using System.Runtime.InteropServices;

namespace Console1
{
    public class MyAsyncFileStream : IDisposable
    {
        long _pos;
        SafeFileHandle? _pHandle;

        public MyAsyncFileStream(string filePath, FileAccess fileAccess, FileShare fileShare, FileMode fileMode)
        {
            _pHandle = NativeMethods.CreateFile(filePath, fileAccess, fileShare, IntPtr.Zero, fileMode,
                            NativeMethods.FILE_FLAG_OVERLAPPED, IntPtr.Zero);

            if (_pHandle.IsInvalid == true)
            {
                int lastError = Marshal.GetHRForLastWin32Error();
                Marshal.ThrowExceptionForHR(lastError);
            }
        }

        public long Position
        {
            get { return _pos; }
            set
            {
                _pos = value;
                SeekCore(_pos, SeekOrigin.Begin);
            }
        }

        void SeekCore(long offset, SeekOrigin origin)
        {
            if (NativeMethods.SetFilePointerEx(_pHandle!, offset, out long moved, (uint)origin) == true)
            {
                _pos = moved;
            }
        }

        public unsafe MyTask WriteAsync(byte[] buffer)
        {
            uint written;
            NativeOverlapped o = new NativeOverlapped();
            o.OffsetLow = unchecked((int)_pos);
            o.OffsetHigh = (int)(_pos >> 32);

            SeekCore(buffer.Length, SeekOrigin.Current);

            EventWaitHandle writeEvent = new EventWaitHandle(false, EventResetMode.AutoReset);
            o.EventHandle = writeEvent.SafeWaitHandle.DangerousGetHandle();

            int ret = NativeMethods.WriteFile(_pHandle!, buffer, buffer.Length, out written, ref o);
            MyTask task = new MyTask();

            if (ret == 1)
            {
                task.SetComplete();
            }
            else
            {
                int lastError = Marshal.GetLastWin32Error();
                if (lastError == 0)
                {
                    task.SetComplete();
                }
                else if (lastError == 997) //  ERROR_IO_PENDING == 997
                {
                    OverlappedParameter op = new OverlappedParameter()
                    {
                        Event = writeEvent,
                        Task = task,
                    };

                    op.WaitHandle = ThreadPool.RegisterWaitForSingleObject(writeEvent, ioCompletedCallback, op, -1, true);
                    op.TryUnregister();
                }
                else
                {
                    throw Marshal.GetExceptionForHR(NativeMethods.HRESULT_FROM_WIN32(lastError)) ?? new IOException();
                }
            }

            return task;
        }

        public unsafe MyTask ReadAsync(byte[] buffer)
        {
            NativeOverlapped no = new NativeOverlapped();
            no.OffsetLow = unchecked((int)_pos);
            no.OffsetHigh = (int)(_pos >> 32);

            SeekCore(buffer.Length, SeekOrigin.Current);

            EventWaitHandle readEvent = new EventWaitHandle(false, EventResetMode.AutoReset);
            no.EventHandle = readEvent.SafeWaitHandle.DangerousGetHandle();

            int ret = NativeMethods.ReadFile(_pHandle!, buffer, buffer.Length, out uint readLen, ref no);
            MyTask task = new MyTask();

            if (ret == 1)
            {
                task.SetComplete();
            }
            else
            {
                int lastError = Marshal.GetLastWin32Error();
                if (lastError == 0)
                {
                    task.SetComplete();
                }
                else if (lastError == NativeMethods.ERROR_IO_PENDING) // 997 == Overlapped I/O operation is in progress. 
                {
                    OverlappedParameter op = new OverlappedParameter
                    {
                        Event = readEvent,
                        Task = task,
                    };

                    op.WaitHandle = ThreadPool.RegisterWaitForSingleObject(readEvent, ioCompletedCallback, op, -1, true);
                    op.TryUnregister();
                }
                else
                {
                    throw Marshal.GetExceptionForHR(NativeMethods.HRESULT_FROM_WIN32(lastError)) ?? new IOException();
                }
            }

            return task;
        }


        private static void ioCompletedCallback(object? state, bool timedOut)
        {
            OverlappedParameter? op = state as OverlappedParameter;
            if (op == null)
            {
                return;
            }

            op.Done();

            MyTask? task = op.Task;
            if (task == null)
            {
                return;
            }

            if (timedOut == true)
            {
                // task.SetException();
                return;
            }

            task.SetComplete(); // 이 단계에서 await으로 나뉜 "분할 2" 코드를 수행
        }

        public void Dispose()
        {
            if (_pHandle != null)
            {
                _pHandle.Dispose();
                _pHandle = null;
            }
        }
    }

    public class OverlappedParameter
    {
        // ...[생략]...
    }
}

끝입니다. ^^ 이제 우리도 FileStream이 했던 것처럼, MyAsyncFileStream과 MyTask의 협업으로 다음과 같이 (거의 유사하게) 코딩을 할 수 있습니다.

static async Task Main(string[] args)
{
    Console.WriteLine($"[{DateTime.Now}] 1단계: threadid == {Thread.CurrentThread.ManagedThreadId}");

    string filePath = @"test.txt";

    MyAsyncFileStream fs = new MyAsyncFileStream(filePath, FileAccess.ReadWrite, FileShare.ReadWrite, FileMode.Open);

    {
        byte[] buffer = Encoding.UTF8.GetBytes("12");
        await fs.WriteAsync(buffer);
        buffer = Encoding.UTF8.GetBytes("34");
        await fs.WriteAsync(buffer);
    }

    fs.Position = 0;

    {
        byte[] buffer = new byte[2];
        await fs.ReadAsync(buffer);
        Console.WriteLine($"[{DateTime.Now}] 2단계: {BitConverter.ToString(buffer)} threadid == {Thread.CurrentThread.ManagedThreadId}");

        await fs.ReadAsync(buffer);
        Console.WriteLine($"[{DateTime.Now}] 3단계: {BitConverter.ToString(buffer)} threadid == {Thread.CurrentThread.ManagedThreadId}");
    }
}

/* 출력 결과 */
/* test.txt의 파일 내용: 1234567890

[2022-05-15 오전 9:58:09] 1단계: threadid == 1
[2022-05-15 오전 9:58:09] 2단계: 31-32 threadid == 8
[2022-05-15 오전 9:58:09] 3단계: 33-34 threadid == 8
*/

(첨부 파일은 이 글의 예제 코드를 포함합니다.)




잠깐 출력 결과를 해석해 볼까요?

[2022-05-15 오전 9:58:09] 1단계: threadid == 1
[2022-05-15 오전 9:58:09] 2단계: 31-32 threadid == 8
[2022-05-15 오전 9:58:09] 3단계: 33-34 threadid == 8

Main 메서드를 시작한 thread는 1번인데요, 이어서 ReadAsync를 호출해 비동기 I/O를 개시하고 RegisterWaitForSingleObject로 ioCompletedCallback을 등록합니다. 이후 I/O가 완료되면 스레드 풀의 여유 스레드(위의 출력에서는 8번 스레드)에 의해 ioCompletedCallback이 호출됩니다. 그리고 그 8번 스레드는 다시 두 번째 ReadAsync를 호출하고 스레드 풀에 돌아갑니다. 마찬가지로 다시 한번 ioCompletedCallback이 발생하고 그것을 (스레드 풀에 돌아갔던) 8번 스레드가 실행하는 결과를 보여줍니다.

혹은 아래의 문서에 따르면,

Asynchronous disk I/O appears as synchronous on Windows
; https://docs.microsoft.com/en-US/troubleshoot/windows/win32/asynchronous-disk-io-synchronous

Frequently, asynchronous I/O operations behave just as synchronous I/O.
...
if an operation is completed immediately, then &NumberOfBytesRead passed into ReadFile is valid for the number of bytes read. In this case, ignore the OVERLAPPED structure passed into ReadFile; do not use it with GetOverlappedResult or WaitForSingleObject.


ReadAsync는 ReadFile 호출에서 동기 방식으로 그순간 완료가 돼 반환하는 경우도 있으므로 이전 단계의 async callback을 실행하던 8번 스레드가 두 번째 async read를 끝까지 실행하는 경우도 가능하므로 해석은 상황에 따라 다를 수 있습니다.

그런데 실제로 이 상황이 있는지 확인하려고 ReadAsync (또는 WriteAsync)의 반환에서 확인 코드를 넣어봤지만,

int ret = NativeMethods.ReadFile(_pHandle!, buffer, buffer.Length, out uint readLen, ref no);
MyTask task = new MyTask();

if (ret == 1)
{
    Console.WriteLine("Sync");
    task.SetComplete();
}

아무리 실행해도 "Sync" 메시지가 화면에 출력되는 경우를 볼 수 없었습니다. (혹시, 이와 관련해서 저 메시지를 볼 수 있는 "요령"을 알고 계신 분은 덧글 부탁드립니다. ^^)

마지막으로, 위의 예제 코드를 실행하다 보면 간혹 다음과 같은 식으로 2단계와 3단계의 스레드가 다르게 나오는 경우를 볼 수 있는데요,

[2022-05-15 오전 10:06:57] 1단계: threadid == 1
[2022-05-15 오전 10:06:57] 2단계: 31-32 threadid == 8
[2022-05-15 오전 10:06:57] 3단계: 33-34 threadid == 11

8번 스레드가 스레드 풀에 여유 스레드 자원으로 안착하기도 전에 두 번째 ioCompletedCallback이 호출돼 11번 여유 스레드가 호출된 것으로 해석할 수 있습니다.

자, 그럼 이것으로 async/await에 관한 스레드 이야기는 모두 마쳤습니다. async/await의 마법 같은 동작 방식이, 아마도 지금까지의 글을 잘 읽고 이해하셨다면 더 이상 마법이 아닌, FileStream과 Task와 C# 컴파일러의 노력으로 빚은 노가다 예술임을 느끼실 수 있을 것입니다. ^^




이 글에서 사용한 MyTask, MyAsyncFileStream은 async/await 학습을 위한 딱 그 정도 역할만 하도록 만들어진 것이므로 현업에서 쓰는 것에는 권장하지 않습니다. (미구현 코드가 너무 많기도 하고, 안정성을 보장할 수 없습니다.)




[이 글에 대해서 여러분들과 의견을 공유하고 싶습니다. 틀리거나 미흡한 부분 또는 의문 사항이 있으시면 언제든 댓글 남겨주십시오.]

[연관 글]






[최초 등록일: ]
[최종 수정일: 5/17/2022]

Creative Commons License
이 저작물은 크리에이티브 커먼즈 코리아 저작자표시-비영리-변경금지 2.0 대한민국 라이센스에 따라 이용하실 수 있습니다.
by SeongTae Jeong, mailto:techsharer at outlook.com

비밀번호

댓글 작성자
 




[1]  2  3  4  5  6  7  8  9  10  11  12  13  14  15  ...
NoWriterDateCnt.TitleFile(s)
13094정성태7/6/202218오류 유형: 816. Golang - "short write" 오류 원인
13093정성태7/5/202257.NET Framework: 2029. C# - HttpWebRequest로 localhost 접속 시 2초 이상 지연
13092정성태7/3/2022121.NET Framework: 2028. C# - HttpWebRequest의 POST 동작 방식파일 다운로드1
13091정성태7/3/202279.NET Framework: 2027. C# - IPv4, IPv6를 모두 지원하는 서버 소켓 생성 방법
13090정성태6/29/202260오류 유형: 815. PyPI에 업로드한 패키지가 반영이 안 되는 경우
13089정성태6/28/202280개발 환경 구성: 646. HOSTS 파일 변경 시 Edge 브라우저에 반영하는 방법
13088정성태6/27/202258개발 환경 구성: 645. "Developer Command Prompt for VS 2022" 명령행 환경의 폰트를 바꾸는 방법
13087정성태6/23/2022129스크립트: 41. 파이썬 - FastAPI / uvicorn 호스팅 환경에서 asyncio 사용하는 방법
13086정성태6/22/2022238.NET Framework: 2026. C# 11 - 문자열 보간 개선 2가지파일 다운로드1
13085정성태6/22/2022192.NET Framework: 2025. C# 11 - 원시 문자열 리터럴(raw string literals)파일 다운로드1
13084정성태6/21/202270개발 환경 구성: 644. Windows - 파이썬 2.7을 msi 설치 없이 구성하는 방법
13083정성태6/20/2022195.NET Framework: 2024. .NET 7에 도입된 GC의 메모리 해제에 대한 segment와 region의 차이점
13082정성태6/19/2022117.NET Framework: 2023. C# - Process의 I/O 사용량을 보여주는 GetProcessIoCounters Win32 API파일 다운로드1
13081정성태6/17/2022192.NET Framework: 2022. C# - .NET 7 Preview 5 신규 기능 - System.IO.Stream ReadExactly / ReadAtLeast파일 다운로드1
13080정성태6/17/2022129개발 환경 구성: 643. Visual Studio 2022 17.2 버전에서 C# 11 또는 .NET 7.0 preview 적용
13079정성태6/17/202296오류 유형: 814. 파이썬 - Error: The file/path provided (...) does not appear to exist
13078정성태6/16/2022188.NET Framework: 2021. WPF - UI Thread와 Render Thread파일 다운로드1
13077정성태6/15/2022116스크립트: 40. 파이썬 - postgresql 환경 구성
13075정성태6/15/2022201Linux: 50. Linux - apt와 apt-get의 차이 [2]
13074정성태6/13/2022205.NET Framework: 2020. C# - NTFS 파일에 사용자 정의 속성값 추가하는 방법파일 다운로드1
13073정성태6/12/2022244Windows: 207. Windows Server 2022에 도입된 WSL 2
13072정성태6/10/2022157Linux: 49. Linux - ls 명령어로 출력되는 디렉터리 색상 변경 방법
13071정성태6/9/2022200스크립트: 39. Python에서 cx_Oracle 환경 구성
13070정성태6/8/2022243오류 유형: 813. Windows 11에서 입력 포커스가 바뀌는 문제 [1]
13069정성태5/26/2022496.NET Framework: 2019. C# - .NET에서 제공하는 3가지 Timer 비교
[1]  2  3  4  5  6  7  8  9  10  11  12  13  14  15  ...