Microsoft MVP성태의 닷넷 이야기
글쓴 사람
정성태 (techsharer at outlook.com)
홈페이지
첨부 파일
(연관된 글이 2개 있습니다.)

C# - 직접 만들어 보는 TaskScheduler 실습 (SingleThreadTaskScheduler)

일반적으로 Task에 할당된 코드는 다음과 같은 구문으로 실행합니다.

// Windows Forms + .NET 7

public partial class Form1 : Form
{
    public Form1()
    {
        InitializeComponent();
    }

    private void button1_Click(object sender, EventArgs e)
    {
        System.Diagnostics.Trace.WriteLine($"{Thread.CurrentThread.ManagedThreadId}: button1_Click");

        Task.Run(() =>
        {
            System.Diagnostics.Trace.WriteLine($"{Thread.CurrentThread.ManagedThreadId}: Task.Run");
            this.button1.Text = "TEST";
        });
    }
}

/* 출력 결과
1: button1_Click
10: Task.Run
*/

보는 바와 같이 button1_Click 이벤트 핸들러를 실행하는 UI 스레드가 있고, Task.Run에 전달된 코드를 실행하는 스레드는 ThreadPool로부터 빌려온 것입니다. (위의 코드는 2차 스레드에서 button1을 접근하므로 예외가 발생합니다.)

여기서, Task.Run이 사용하는 스레드 운영은 TaskScheduler 타입을 상속받아 사용자 정의할 수 있습니다. 위의 경우처럼 사용자가 명시하지 않은 경우, 기본적으로는 System.Threading.Tasks.ThreadPoolTaskScheduler 타입이 사용되는데, 이 값은 TaskScheduler.Default 정적 속성으로 반환됩니다.

System.Diagnostics.Trace.WriteLine($"{TaskScheduler.Default}: TaskScheduler.Default");

// 출력 결과: System.Threading.Tasks.ThreadPoolTaskScheduler: TaskScheduler.Default

전에도 설명했지만, Task.Run은 다음의 구문과 동일하게 취급할 수 있습니다.

// 간단하게는 이렇게,
Task.Run(action);

// 복잡하게는 이렇게 호출
Task.Factory.StartNew(func, CancellationToken.None, 
                TaskCreationOptions.DenyChildAttach, TaskScheduler.Default)

TaskScheduler.Default에 설정된 ThreadPoolTaskScheduler 타입의 소스 코드는 그리 복잡하지 않은데,

runtime/src/libraries/System.Private.CoreLib/src/System/Threading/Tasks/ThreadPoolTaskScheduler.cs 
; https://github.com/dotnet/runtime/blob/main/src/libraries/System.Private.CoreLib/src/System/Threading/Tasks/ThreadPoolTaskScheduler.cs

핵심 코드는 QueueTask 메서드로, 이것은 사용자 작업을 나타내는 "Task"를 ThreadPool의 스레드에 실어 실행하는 코드를 담고 있습니다.

protected internal override void QueueTask(Task task)
{
    TaskCreationOptions options = task.Options;
    if (Thread.IsThreadStartSupported && (options & TaskCreationOptions.LongRunning) != 0)
    {
        // Run LongRunning tasks on their own dedicated thread.
        new Thread(s_longRunningThreadWork)
        {
            IsBackground = true,
            Name = ".NET Long Running Task"
        }.UnsafeStart(task);
    }
    else
    {
        // Normal handling for non-LongRunning tasks.
        ThreadPool.UnsafeQueueUserWorkItemInternal(task), (options & TaskCreationOptions.PreferFairness) == 0);
    }
}

게다가 LongRunning인 경우는 별도의 Thread를 생성해 (Task 인스턴스로 대표되는) 사용자 작업을 처리합니다.

그러니까, 사실 Task 자체는 스레드에 대한 어떤 선택권도 없고, 순수하게 사용자가 맡긴 작업과 그 작업에 대한 속성을 나타내는 인스턴스에 불과한 것입니다.




자, 그럼 다른 유형의 TaskScheduler를 알아볼까요? ^^ Task와 연관돼 직접 지정하는 유형으로 FromCurrentSynchronizationContext 메서드도 별도의 TaskScheduler를 반환합니다.

private void button2_Click(object sender, EventArgs e)
{
    System.Diagnostics.Trace.WriteLine($"{Thread.CurrentThread.ManagedThreadId}: button2_Click");

    Task.Factory.StartNew(() =>
    {
        System.Diagnostics.Trace.WriteLine($"{Thread.CurrentThread.ManagedThreadId}: StartNew");
        this.button2.Text = "TEST";
    }, CancellationToken.None, TaskCreationOptions.None, TaskScheduler.FromCurrentSynchronizationContext());
}

/* 출력 결과:
1: button2_Click
1: StartNew
*/

보는 바와 같이 TaskScheduler를 지정하려면 Task.Factory, 즉 TaskFactory 타입을 이용해야 합니다. 그리고 이때 TaskScheduler.FromCurrentSynchronizationContext 메서드는,

// https://github.com/dotnet/runtime/blob/main/src/libraries/System.Private.CoreLib/src/System/Threading/Tasks/TaskScheduler.cs#L344

public static TaskScheduler FromCurrentSynchronizationContext()
{
    return new SynchronizationContextTaskScheduler();
}

ThreadPoolTaskScheduler보다 더 간단한 구현 코드를 갖는 SynchronizationContextTaskScheduler 인스턴스를 반환합니다.

// https://github.com/dotnet/runtime/blob/main/src/libraries/System.Private.CoreLib/src/System/Threading/Tasks/TaskScheduler.cs#L567

// 아래의 타입을 여러분의 프로젝트에 그대로 추가하는 것도 가능합니다.
// 따라서 new를 한 인스턴스를 Task.Factory.StartNew의 TaskScheduler 인자에 전달해도 됩니다.
// 당연히 동작은 닷넷의 것과 동일하므로 전체적인 수행 과정을 이해하기 위한 용도로 사용할 수 있습니다. 
internal sealed class SynchronizationContextTaskScheduler : TaskScheduler
{
    private readonly SynchronizationContext m_synchronizationContext;

    internal SynchronizationContextTaskScheduler()
    {
        m_synchronizationContext = SynchronizationContext.Current ??
            throw new InvalidOperationException("TaskScheduler_FromCurrentSynchronizationContext_NoCurrent");
    }

    protected override void QueueTask(Task task)
    {
        m_synchronizationContext.Post(s_postCallback, (object)task);
    }

    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
    {
        if (SynchronizationContext.Current == m_synchronizationContext)
        {
            return TryExecuteTask(task);
        }
        else
        {
            return false;
        }
    }

    protected override IEnumerable<Task>? GetScheduledTasks() => null;

    public override int MaximumConcurrencyLevel => 1;

    private static readonly SendOrPostCallback s_postCallback = new SendOrPostCallback(PostCallback);

    private static void PostCallback(object? s)
    {
        // 원본 코드: ExecuteEntry가 internal이므로 호출이 불가능하기 때문에,
        //Debug.Assert(s is Task);
        //((Task)s).ExecuteEntry(); // with double-execute check because SC could be buggy

        // 수정 코드: Reflection을 이용해 호출하는 코드로 변경
        Task? task = s as Task;

        var mi = typeof(Task).GetMethod("ExecuteEntry", System.Reflection.BindingFlags.NonPublic | System.Reflection.BindingFlags.Instance);
        mi?.Invoke(s, null);

        // 원본을 맞추기 위해 위의 코드를 사용했지만, 원래대로라면 이런 경우 TaskScheduler의 TryExecuteTask를 사용
        // 부모 클래스(TaskScheduler)에서 protected로 제공하는 TryExecuteTask는 task.ExecuteEntry 호출
        // 실제로 이 글의 마지막에 작성하는 SingleThreadTaskScheduler에서는 TryExecuteTask를 이용해 Task의 작업을 수행
    }
}

이 코드의 핵심도 마찬가지로 QueueTask 메서드인데, 하는 일이라고는 SynchronizationContext.Current, 위의 경우 Windows Forms이므로 System.Windows.Forms.WindowsFormsSynchronizationContextPost를 하는 것뿐입니다.

protected override void QueueTask(Task task)
{
    m_synchronizationContext.Post(s_postCallback, (object)task);
}

보는 바와 같이 (ThreadPoolTaskScheduler와는 달리) 별도의 스레드가 만들어진 것도 아니고, ThreadPool을 사용하지도 않았는데요, 따라서 위의 QueueTask와 WindowsFormsSynchronizationContext.Post 메서드까지 모두 Task.Factory.StartNew를 호출하는 스레드가 담당하게 됩니다. 물론, Post로 전달된 Delegate의 실행은 SynchronizationContext 종류에 따라 달라집니다.

이렇게 보니까 별로 신기할 것이 없죠? ^^




TaskScheduler는 사용자가 원한다면 얼마든지 새롭게 정의하는 것이 가능합니다. 예를 하나 들어볼까요? ^^ 너무 복잡하지 않은 걸로... 가령 SynchronizationContextTaskScheduler와는 달리 UI 스레드에 얽매이지는 않게 만들면서 오로지 "단일 스레드"만을 사용해 Task를 처리하는 스케줄러는 어떨까요?

일종의 STAThread 특성처럼, 호출을 (동기화가 필요 없도록) 직렬화하는 역할만 하는 것으로 대충 다음과 같이 만들 수 있습니다.

internal sealed class SingleThreadTaskScheduler : TaskScheduler
{
    static Thread _thread;
    static BlockingCollection<WorkItem> _workItems;

    class WorkItem
    {
        Task _task;
        public Task Task => _task;
        Func<Task, bool> _func;
        public Func<Task, bool> Func => _func;

        internal WorkItem(Func<Task, bool> func, Task task)
        {
            _task = task;
            _func = func;
        }
    }

    static SingleThreadTaskScheduler()
    {
        _workItems = new BlockingCollection<WorkItem>();

        _thread = new Thread(threadFunc);
        _thread.IsBackground = true;
        _thread.Start();
    }

    static void threadFunc()
    {
        while (true)
        {
            var item = _workItems.Take();
            item.Func(item.Task);
        }
    }

    protected override void QueueTask(Task task)
    {
        _workItems.Add(new WorkItem(TryExecuteTask, task));
    }

    protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
    {
        return false;
    }

    protected override IEnumerable<Task>? GetScheduledTasks() => null;
    public override int MaximumConcurrencyLevel => 1;
}

QueueTask로 들어온 작업을, 미리 static 멤버로 생성해 두었던 Thread에서 실행하도록 만들었으므로 이를 이용하면,

private void button3_Click(object sender, EventArgs e)
{
    System.Diagnostics.Trace.WriteLine($"{Thread.CurrentThread.ManagedThreadId}: button3_Click");

    TaskScheduler staScheduler = new SingleThreadTaskScheduler();

    Task.Factory.StartNew(() =>
    {
        System.Diagnostics.Trace.WriteLine($"{Thread.CurrentThread.ManagedThreadId}: StartNew");
        this.Invoke(() => // SingleThreadTaskScheduler의 스레드에서 실행하므로 Control.Invoke 처리 필요
        {
            this.button3.Text = "TEST";
        });

    }, CancellationToken.None, TaskCreationOptions.None, staScheduler);
}

/* 출력 결과:
1: button3_Click
12: StartNew
*/

button3_Click을 몇 번을 실행해도 "StartNew" 출력은 매번 12번 스레드에서 실행되는 것을 확인할 수 있습니다.

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




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

[연관 글]






[최초 등록일: ]
[최종 수정일: 12/10/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)
13539정성태1/26/20245710개발 환경 구성: 703. Visual Studio - launchSettings.json을 이용한 HTTP/HTTPS 포트 바인딩
13538정성태1/25/20246137닷넷: 2211. C# - NonGC(FOH) 영역에 .NET 개체를 생성파일 다운로드1
13537정성태1/24/20246375닷넷: 2210. C# - Native 메모리에 .NET 개체를 생성파일 다운로드1
13536정성태1/23/20246159닷넷: 2209. .NET 8 - NonGC Heap / FOH (Frozen Object Heap) [1]
13535정성태1/22/20246243닷넷: 2208. C# - GCHandle 구조체의 메모리 분석
13534정성태1/21/20246201닷넷: 2207. C# - SQL Server DB를 bacpac으로 Export/Import파일 다운로드1
13533정성태1/18/20246419닷넷: 2206. C# - TCP KeepAlive의 서버 측 구현파일 다운로드1
13532정성태1/17/20246218닷넷: 2205. C# - SuperSimpleTcp 사용 시 주의할 점파일 다운로드1
13531정성태1/16/20246481닷넷: 2204. C# - TCP KeepAlive에 새로 추가된 Retry 옵션파일 다운로드1
13530정성태1/15/20245647닷넷: 2203. C# - Python과의 AES 암호화 연동파일 다운로드1
13529정성태1/15/20245676닷넷: 2202. C# - PublishAot의 glibc에 대한 정적 링킹하는 방법
13528정성태1/14/20245946Linux: 68. busybox 컨테이너에서 실행 가능한 C++, Go 프로그램 빌드
13527정성태1/14/20245886오류 유형: 892. Visual Studio - Failed to launch debug adapter. Additional information may be available in the output window.
13526정성태1/14/20246043닷넷: 2201. C# - Facebook 연동 / 사용자 탈퇴 처리 방법
13525정성태1/13/20245555오류 유형: 891. Visual Studio - Web Application을 실행하지 못하는 IISExpress
13524정성태1/12/20245782오류 유형: 890. 한국투자증권 KIS Developers OpenAPI - GW라우팅 중 오류가 발생했습니다.
13523정성태1/12/20245343오류 유형: 889. Visual Studio - error : A project with that name is already opened in the solution.
13522정성태1/11/20246008닷넷: 2200. C# - HttpClient.PostAsJsonAsync 호출 시 "Transfer-Encoding: chunked" 대신 "Content-Length" 헤더 처리
13521정성태1/11/20246253닷넷: 2199. C# - 한국투자증권 KIS Developers OpenAPI의 WebSocket Ping, Pong 처리
13520정성태1/10/20246003오류 유형: 888. C# - Unable to resolve service for type 'Microsoft.Extensions.ObjectPool.ObjectPool`....' [1]
13519정성태1/10/20245480닷넷: 2198. C# - Reflection을 이용한 ClientWebSocket의 Ping 호출파일 다운로드1
13518정성태1/9/20245859닷넷: 2197. C# - ClientWebSocket의 Ping, Pong 처리
13517정성태1/8/20245652스크립트: 63. Python - 공개 패키지를 이용한 위성 이미지 생성 (pystac_client, odc.stac)
13516정성태1/7/20245547닷넷: 2196. IIS - AppPool의 "Disable Overlapped Recycle" 옵션의 부작용
13515정성태1/6/20246212닷넷: 2195. async 메서드 내에서 C# 7의 discard 구문 활용 사례 [1]
13514정성태1/5/20245838개발 환경 구성: 702. IIS - AppPool의 "Disable Overlapped Recycle" 옵션
1  2  3  4  5  6  7  8  9  10  11  [12]  13  14  15  ...