一個簡單的模擬例項說明Task及其排程問題

語言: CN / TW / HK

Task對於.NET的重要性毋庸置疑。通過最近的一些面試經歷,發現很多人對與Task及其排程機制,以及執行緒和執行緒池之間的關係並沒有清晰的認識。本文采用最簡單的方式模擬了Task的實現,旨在說明Task是什麼?它是如何被排程執行的?原始碼從這裡下載。

一、Task(Job)

二、TaskScheduler(JobScheduler)

三、基於執行緒池的排程

四、使用指定執行緒進行排程

五、非同步等待

六、await關鍵字的運用

七、狀態機

一、Task(Job)

Task代表一項具有某種狀態的操作,我們使用如下這個Job型別來模擬Task。Job封裝的操作體現為一個Action委託,狀態則通過JobStatus列舉來表示(對應TaskStatus列舉)。簡單起見,我們僅僅定義了四種狀態(建立、排程、執行和完成)。Invoke方法負責執行封裝的Action委託,並對狀態進行相應設定。

public class Job
{
    private readonly Action _work;
    public Job(Action work)=> _work = work;
    public JobStatus Status { get; internal set; }

    internal protected virtual void Invoke()
    {
        Status = JobStatus.Running;
        _work();
        Status = JobStatus.Completed;        }
}

public enum JobStatus
{
    Created,
    Scheduled,
    Running,
    Completed
}

二、TaskScheduler(JobScheduler)

Task承載的操作通過排程得以執行,具體的排程策略取決於排程器的選擇。Task排程器通過TaskScheduler表示,我們利用如下這個JobScheduler型別對它進行模擬。如下面的程式碼片段所示,我們只為抽象類JobScheduler定義了唯一的QueueJob方法來排程作為引數的Job物件。靜態Current屬性表示當前預設實現的排程器。

public abstract class JobScheduler
{
    public abstract void QueueJob(Job job);
    public static JobScheduler Current { get; set; } = new ThreadPoolJobScheduler ();
}

對於開發者來說,執行Task就是將它提交給排程器,這一操作體現在我們為Job型別定義的靜態Start方法中。該方法通過引數指定具體的排程器,如果沒有顯式指定,預設採用JobScheduler的Current靜態屬性設定的預設排程器。為了方便後面的演示,我們還定義了一個靜態的Run方法,該方法會將指定的Action物件封裝成Job,並呼叫Start方法利用預設的排程器進行排程。

public class Job
{
    private readonly Action _work;
    public Job(Action work)=> _work = work;
    public JobStatus Status { get; internal set; }

    internal protected virtual void Invoke()
    {
        Status = JobStatus.Running;
        _work();
        Status = JobStatus.Completed;

    }

    public void Start(JobScheduler? scheduler = null) => (scheduler ?? JobScheduler.Current).QueueJob(this);
    public static Job Run(Action work)
    {
        var job = new Job(work);
        job.Start();
        return job;
    }
}

三、基於執行緒池的排程

Task如何執行取決於選擇怎樣的排程器,.NET預設採用基於執行緒池的排程策略,這一策略體現在ThreadPoolTaskScheduler型別上,我們使用如下這個ThreadPoolJobScheduler 進行模擬。如下面的程式碼片段所示,重寫的QueueJob方法通過呼叫ThreadPool.QueueUserWorkItem方法執行指定Job物件封裝的Action委託。JobScheduler的Current屬性設定的預設排程器就是這麼一個ThreadPoolJobScheduler 物件。

public class ThreadPoolJobScheduler : JobScheduler
{
    public override void QueueJob(Job job)
    {
        job.Status = JobStatus.Scheduled;
        var executionContext = ExecutionContext.Capture();
        ThreadPool.QueueUserWorkItem(_ => ExecutionContext.Run(executionContext!, _ => job.Invoke(), null));
    }
}

我們按照如下的方式呼叫Job的靜態Run方法建立並執行了三個Job,每個Job封裝的Action委託在執行的時候會將當前執行緒ID打印出來。

_ = Job.Run(() => Console.WriteLine($"Job1 is excuted in thread {Thread.CurrentThread.ManagedThreadId}"));
_ = Job.Run(() => Console.WriteLine($"Job2 is excuted in thread {Thread.CurrentThread.ManagedThreadId}"));
_ = Job.Run(() => Console.WriteLine($"Job3 is excuted in thread {Thread.CurrentThread.ManagedThreadId}"));

Console.ReadLine();

由於採用預設的基於執行緒池的排程策略,所以三個Job會在三個不同的執行緒上執行。

四、使用指定執行緒進行排程

我們知道.NET程序只有一個全域性的執行緒池,對於一些需要長時間執行且具有較高優先順序的操作,採用基於執行緒池的呼叫未必是好的選擇。比如在一個Web應用中,執行緒池的工作執行緒會被用來處理請求,對於一個需要持續執行的Job可能會因為可用工作執行緒的不足而被阻塞。.NET對於這種情況具有不同的處理方式(啟動Task的時候選擇TaskCreationOptions.LongRunning選項),這裡我們使用自定義排程器的方式來解決這個問題。如下這個DedicatedThreadJobScheduler 利用建立的“專有執行緒”來保證被呼叫的Job能夠“立即”執行。執行緒的數量通過建構函式的引數指定,執行緒在無事可做的時候被“掛起”以及有新的Job被排程時被“復甦”通過一個ManualResetEvent物件來完成。

public class DedicatedThreadJobScheduler : JobScheduler
{
    private readonly Queue<Job>[] _queues;
    private readonly Thread[] _threads;
    private readonly ManualResetEvent[] _events;
    public DedicatedThreadJobScheduler (int threadCount)
    {
        _queues = new Queue<Job>[threadCount];
        _threads = new  Thread[threadCount];
        _events = new ManualResetEvent[threadCount];

        for (int index = 0; index < threadCount; index++)
        {
            var queue = _queues[index] = new Queue<Job>();
            var thread  = _threads[index] = new Thread(Invoke);
            _events[index] = new ManualResetEvent(true);
            thread.Start(index);
        }

        void Invoke(object? state)
        {
            var index = (int)state!;
            var @event = _events[index];
            while (true)
            {
                if (@event.WaitOne())
                {
                    while (true)
                    {
                        if (!_queues[index].TryDequeue(out var job))
                        {
                            Suspend(index);
                            break;
                        }
                        job.Invoke();
                    }
                }
            }
        }
    }
    public override void QueueJob(Job job)
    {
        job.Status = JobStatus.Scheduled;
        var (queue, index) =  _queues.Select((queue, index) => (queue, index)).OrderBy(it => it.queue.Count).First();
        queue.Enqueue(job);
        Resume(index);
    }

    public void Suspend(int index) => _events[index].Reset();
    public void Resume(int index) => _events[index].Set();
}

還是上面演示的程式,這次我們將當前排程器設定為上面這個DedicatedThreadJobScheduler ,並將使用的執行緒數設定為2。

JobScheduler.Current = new DedicatedThreadJobScheduler (2);
_ = Job.Run(() => Console.WriteLine($"Job1 is excuted in thread {Thread.CurrentThread.ManagedThreadId}"));
_ = Job.Run(() => Console.WriteLine($"Job2 is excuted in thread {Thread.CurrentThread.ManagedThreadId}"));
_ = Job.Run(() => Console.WriteLine($"Job3 is excuted in thread {Thread.CurrentThread.ManagedThreadId}"));
_ = Job.Run(() => Console.WriteLine($"Job4 is excuted in thread {Thread.CurrentThread.ManagedThreadId}"));
_ = Job.Run(() => Console.WriteLine($"Job5 is excuted in thread {Thread.CurrentThread.ManagedThreadId}"));
_ = Job.Run(() => Console.WriteLine($"Job6 is excuted in thread {Thread.CurrentThread.ManagedThreadId}"));

Console.ReadLine();

我們會發現所有的操作只會在兩個固定的執行緒中被執行。

五、非同步等待

如果需要在某個Task執行之後接著執行後續的操作,我們可以呼叫其ContinueWith方法指定待執行的操作,現在我們將這個方法定義Job型別上。Job與Task的ContinueWith有些差異,在這裡我們認為ContinueWith指定的也是一個Job,那麼多個Job則可以按照預先編排的順序構成一個連結串列。當前Job執行後,只需要將後續這個Job交付給排程器就可以了。如下面的程式碼片段所示,我們利用_continue欄位來表示非同步等待執行的Job,並利用它維持一個Job連結串列。ContinueWith方法會將指定的Action委託封裝成Job並新增到連結串列末端。

public class Job
{
    private readonly Action _work;
    private Job? _continue;
    public Job(Action work) => _work = work;
    public JobStatus Status { get; internal set; }
    public void Start(JobScheduler? scheduler = null) => (scheduler ?? JobScheduler.Current).QueueJob(this);
    internal protected virtual void Invoke()
    {
        Status = JobStatus.Running;
        _work();
        Status = JobStatus.Completed;
        _continue?.Start();
    }

    public static Job Run(Action work)
    {
        var job = new Job(work);
        job.Start();
        return job;
    }

    public Job ContinueWith(Action<Job> continuation)
    {
        if (_continue == null)
        {
            var job = new Job(() => continuation(this));
            _continue = job;
        }
        else
        {
            _continue.ContinueWith(continuation);
        }
        return this;
    }
}

利用ContinueWith方法實現非同步操作的按序執行體現在如下的程式中。

Job.Run(() =>
{
    Thread.Sleep(1000);
    Console.WriteLine("Foo1");
}).ContinueWith(_ =>
{
    Thread.Sleep(100);
    Console.WriteLine("Bar1");
}).ContinueWith(_ =>
{
    Thread.Sleep(100);
    Console.WriteLine("Baz1");
});

Job.Run(() =>
{
    Thread.Sleep(100);
    Console.WriteLine("Foo2");
}).ContinueWith(_ =>
{
    Thread.Sleep(10);
    Console.WriteLine("Bar2");
}).ContinueWith(_ =>
{
    Thread.Sleep(10);
    Console.WriteLine("Baz2");
});

Console.ReadLine();

輸出結果

六、await關鍵字的運用

雖然ContinueWith方法能夠解決“非同步等待”的問題,但是我們更喜歡使用await關鍵字,接下來我們就為Job賦予這個能力。為此我們定義瞭如下這個實現了ICriticalNotifyCompletion介面的JobAwaiter結構體。顧名思義,該介面用來發送操作完成的通知。一個JobAwaiter物件由一個Job物件構建而成,當它自身執行完成之後,OnCompleted方法會被呼叫,我們利用它執行後續的操作。

public struct JobAwaiter: ICriticalNotifyCompletion
{
    private readonly Job _job;
    public bool IsCompleted => _job.Status ==  JobStatus.Completed;
    public JobAwaiter(Job job)
    {
        _job = job;
        if (job.Status == JobStatus.Created)
        {
            job.Start();
        }
    }
    public void OnCompleted(Action continuation)
    {
        _job.ContinueWith(_ => continuation());
    }
    public void GetResult() { }
    public void UnsafeOnCompleted(Action continuation)=>OnCompleted(continuation);
}

我們在Job型別上新增這個GetAwaiter方法返回根據自身建立的JobAwaiter物件。

public class Job
{
    private readonly Action _work;
    private Job? _continue;
    public Job(Action work) => _work = work;
    public JobStatus Status { get; internal set; }
    public void Start(JobScheduler? scheduler = null) => (scheduler ?? JobScheduler.Current).QueueJob(this);
    internal protected virtual void Invoke()
    {
        Status = JobStatus.Running;
        _work();
        Status = JobStatus.Completed;
        _continue?.Start();
    }


    public static Job Run(Action work)
    {
        var job = new Job(work);
        job.Start();
        return job;
    }
    public Job ContinueWith(Action<Job> continuation)
    {
        if (_continue == null)
        {
            var job = new Job(() => continuation(this));
            _continue = job;
        }
        else
        {
            _continue.ContinueWith(continuation);
        }
        return this;
    }
    public JobAwaiter GetAwaiter() => new(this);
}

任何一個型別一旦擁有了這樣一個GetAwaiter方法,我們就能將await關鍵詞應用在對應的物件上面。

await Foo();
await Bar();
await Baz();
Console.ReadLine();

static Job Foo() =>  new Job(() =>
{
    Thread.Sleep(1000);
    Console.WriteLine("Foo");
});

static Job Bar() => new Job(() =>
{
    Thread.Sleep(100);
    Console.WriteLine("Bar");
});

static Job Baz() => new Job(() =>
{
    Thread.Sleep(10);
    Console.WriteLine("Baz");
});

輸出結果:

七、狀態機

我想你應該知道await關鍵字僅僅是編譯器提供的語法糖,編譯後的程式碼會利用一個“狀態機”實現“非同步等待”的功能,上面這段程式碼最終編譯成如下的形式。值得一提的是,Debug和Release模式編譯出來的程式碼是不同的,下面給出的是Release模式下的編譯結果,上述的狀態機體現為生成的<<Main>$>d__0這個結構體。它的實現其實很簡單:如果個方法出現了N個await關鍵字,它們相當於將整個方法的執行流程切割成N+1段,狀態機的狀態體現為當前應該執行那段,具體的執行體現在MoveNext方法上。GetAwaiter方法返回的ICriticalNotifyCompletion物件用來確定當前操作是否結束,如果結束則可以直接指定後續操作,否則需要呼叫AwaitUnsafeOnCompleted對後續操作進行處理。

// Program
using System;
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Runtime.InteropServices;
using System.Threading.Tasks;
using Jobs;

[CompilerGenerated]
internal class Program
{
	[StructLayout(LayoutKind.Auto)]
	[CompilerGenerated]
	private struct <<Main>$>d__0 : IAsyncStateMachine
	{
		public int <>1__state;

		public AsyncTaskMethodBuilder <>t__builder;

		private JobAwaiter <>u__1;

		private void MoveNext()
		{
			int num = <>1__state;
			try
			{
				JobAwaiter awaiter;
				switch (num)
				{
				default:
					awaiter = <<Main>$>g__Foo|0_0().GetAwaiter();
					if (!awaiter.IsCompleted)
					{
						num = (<>1__state = 0);
						<>u__1 = awaiter;
						<>t__builder.AwaitUnsafeOnCompleted(ref awaiter, ref this);
						return;
					}
					goto IL_006c;
				case 0:
					awaiter = <>u__1;
					<>u__1 = default(JobAwaiter);
					num = (<>1__state = -1);
					goto IL_006c;
				case 1:
					awaiter = <>u__1;
					<>u__1 = default(JobAwaiter);
					num = (<>1__state = -1);
					goto IL_00c6;
				case 2:
					{
						awaiter = <>u__1;
						<>u__1 = default(JobAwaiter);
						num = (<>1__state = -1);
						break;
					}
					IL_00c6:
					awaiter.GetResult();
					awaiter = <<Main>$>g__Baz|0_2().GetAwaiter();
					if (!awaiter.IsCompleted)
					{
						num = (<>1__state = 2);
						<>u__1 = awaiter;
						<>t__builder.AwaitUnsafeOnCompleted(ref awaiter, ref this);
						return;
					}
					break;
					IL_006c:
					awaiter.GetResult();
					awaiter = <<Main>$>g__Bar|0_1().GetAwaiter();
					if (!awaiter.IsCompleted)
					{
						num = (<>1__state = 1);
						<>u__1 = awaiter;
						<>t__builder.AwaitUnsafeOnCompleted(ref awaiter, ref this);
						return;
					}
					goto IL_00c6;
				}
				awaiter.GetResult();
				Console.ReadLine();
			}
			catch (Exception exception)
			{
				<>1__state = -2;
				<>t__builder.SetException(exception);
				return;
			}
			<>1__state = -2;
			<>t__builder.SetResult();
		}

		void IAsyncStateMachine.MoveNext()
		{
			//ILSpy generated this explicit interface implementation from .override directive in MoveNext
			this.MoveNext();
		}

		[DebuggerHidden]
		private void SetStateMachine([System.Runtime.CompilerServices.Nullable(1)] IAsyncStateMachine stateMachine)
		{
			<>t__builder.SetStateMachine(stateMachine);
		}

		void IAsyncStateMachine.SetStateMachine([System.Runtime.CompilerServices.Nullable(1)] IAsyncStateMachine stateMachine)
		{
			//ILSpy generated this explicit interface implementation from .override directive in SetStateMachine
			this.SetStateMachine(stateMachine);
		}
	}

	[AsyncStateMachine(typeof(<<Main>$>d__0))]
	private static Task <Main>$(string[] args)
	{
		<<Main>$>d__0 stateMachine = default(<<Main>$>d__0);
		stateMachine.<>t__builder = AsyncTaskMethodBuilder.Create();
		stateMachine.<>1__state = -1;
		stateMachine.<>t__builder.Start(ref stateMachine);
		return stateMachine.<>t__builder.Task;
	}

	[SpecialName]
	private static void <Main>(string[] args)
	{
		<Main>$(args).GetAwaiter().GetResult();
	}
}

上面提到過,編譯器生成的狀態機程式碼在Debug和Release模式是不一樣的。在Release模式下狀態機是一個結構體,雖然是以介面ICriticalNotifyCompletion的方式使用它,但是由於使用了ref關鍵字,所以不會涉及裝箱,所以不會對GC造成任何影響。但是Debug模式下生成的狀態機則是一個類(如下所示),將會涉及針對堆記憶體的分配和回收。對於遍佈await關鍵字的應用程式,兩者之間的效能差異肯定是不同的。實際上針對Task的很多優化策略,比如使用ValueTask,對某些Task<T>物件(比如狀態為Completed的Task<bool>物件)的複用,以及使用IValueTaskSource等,都是為了解決記憶體分配的問題。

// Program
using System;
using System.Diagnostics;
using System.Runtime.CompilerServices;
using System.Threading.Tasks;
using Jobs;

[CompilerGenerated]
internal class Program
{
	[CompilerGenerated]
	private sealed class <<Main>$>d__0 : IAsyncStateMachine
	{
		public int <>1__state;

		public AsyncTaskMethodBuilder <>t__builder;

		public string[] args;

		private JobAwaiter <>u__1;

		private void MoveNext()
		{
			int num = <>1__state;
			try
			{
				JobAwaiter awaiter3;
				JobAwaiter awaiter2;
				JobAwaiter awaiter;
				switch (num)
				{
				default:
					awaiter3 = <<Main>$>g__Foo|0_0().GetAwaiter();
					if (!awaiter3.IsCompleted)
					{
						num = (<>1__state = 0);
						<>u__1 = awaiter3;
						<<Main>$>d__0 stateMachine = this;
						<>t__builder.AwaitUnsafeOnCompleted(ref awaiter3, ref stateMachine);
						return;
					}
					goto IL_007e;
				case 0:
					awaiter3 = <>u__1;
					<>u__1 = default(JobAwaiter);
					num = (<>1__state = -1);
					goto IL_007e;
				case 1:
					awaiter2 = <>u__1;
					<>u__1 = default(JobAwaiter);
					num = (<>1__state = -1);
					goto IL_00dd;
				case 2:
					{
						awaiter = <>u__1;
						<>u__1 = default(JobAwaiter);
						num = (<>1__state = -1);
						break;
					}
					IL_00dd:
					awaiter2.GetResult();
					awaiter = <<Main>$>g__Baz|0_2().GetAwaiter();
					if (!awaiter.IsCompleted)
					{
						num = (<>1__state = 2);
						<>u__1 = awaiter;
						<<Main>$>d__0 stateMachine = this;
						<>t__builder.AwaitUnsafeOnCompleted(ref awaiter, ref stateMachine);
						return;
					}
					break;
					IL_007e:
					awaiter3.GetResult();
					awaiter2 = <<Main>$>g__Bar|0_1().GetAwaiter();
					if (!awaiter2.IsCompleted)
					{
						num = (<>1__state = 1);
						<>u__1 = awaiter2;
						<<Main>$>d__0 stateMachine = this;
						<>t__builder.AwaitUnsafeOnCompleted(ref awaiter2, ref stateMachine);
						return;
					}
					goto IL_00dd;
				}
				awaiter.GetResult();
				Console.ReadLine();
			}
			catch (Exception exception)
			{
				<>1__state = -2;
				<>t__builder.SetException(exception);
				return;
			}
			<>1__state = -2;
			<>t__builder.SetResult();
		}

		void IAsyncStateMachine.MoveNext()
		{
			//ILSpy generated this explicit interface implementation from .override directive in MoveNext
			this.MoveNext();
		}

		[DebuggerHidden]
		private void SetStateMachine([System.Runtime.CompilerServices.Nullable(1)] IAsyncStateMachine stateMachine)
		{
		}

		void IAsyncStateMachine.SetStateMachine([System.Runtime.CompilerServices.Nullable(1)] IAsyncStateMachine stateMachine)
		{
			//ILSpy generated this explicit interface implementation from .override directive in SetStateMachine
			this.SetStateMachine(stateMachine);
		}
	}

	[AsyncStateMachine(typeof(<<Main>$>d__0))]
	[DebuggerStepThrough]
	private static Task <Main>$(string[] args)
	{
		<<Main>$>d__0 stateMachine = new <<Main>$>d__0();
		stateMachine.<>t__builder = AsyncTaskMethodBuilder.Create();
		stateMachine.args = args;
		stateMachine.<>1__state = -1;
		stateMachine.<>t__builder.Start(ref stateMachine);
		return stateMachine.<>t__builder.Task;
	}

	[SpecialName]
	[DebuggerStepThrough]
	private static void <Main>(string[] args)
	{
		<Main>$(args).GetAwaiter().GetResult();
	}
}