在上一期專欄《並發事件：透過 C# 簡化 APM》中，我探討了如何運用 C# 新語言特性（匿名方法、Lambda 表達式與迭代器）來簡化非同步程式設計。文末展示了如何利用 C# 迭代器，以同步程式設計模型撰寫非同步程式碼。圖 1 即為該範例迭代器。然而，僅靠一般的 C# foreach 陳述式無法執行迭代器程式碼，因為程式碼始終由呼叫 foreach 的執行緒連續執行，不會暫停以完成非同步 I/O 作業。
本期專欄將介紹我開發的 AsyncEnumerator 類別，它能智慧型地驅動迭代器，讓不同的執行緒池執行緒在不同時間執行程式碼，確保迭代器只有在非同步 I/O 作業完成後，才會執行下一次迭代。我將詳細說明 AsyncEnumerator 的架構與運作原理。
AsyncEnumerator 類別的定義如下：
public class AsyncEnumerator {
// 由迭代器外部程式碼呼叫的方法
public AsyncEnumerator();
public void Execute(IEnumerator<Int32> enumerator);
// 由迭代器內部程式碼呼叫的方法
public AsyncCallback End();
public IAsyncResult DequeueAsyncResult();
}
使用 AsyncEnumerator 相當直觀。首先說明如何實作迭代器成員，再說明如何呼叫。
將迭代器成員定義為可接受所需任意參數，並額外加入一個 AsyncEnumerator 物件參考參數。迭代器成員回傳 Int32 集合時，其原型必須設為 IEnumerator<Int32>。
在迭代器內部，透過呼叫對應的 BeginXxx 方法啟動每個非同步作業。呼叫 BeginXxx 時，必須傳遞作業完成時應呼叫的方法名稱。您不需自行定義方法，只需呼叫 AsyncEnumerator 物件的 End 方法，它會回傳一個 AsyncCallback 委派，指向 AsyncEnumerator 內部的私有方法。因此，每個非同步作業完成時，AsyncEnumerator 內部程式碼都會收到通知，並將已完成作業的 IAsyncResult 物件放入一個 List<IAsyncResult> 集合（我稱之為「收件箱」inbox）。
在呼叫 BeginXxx 方法後，放置一個 yield return 陳述式，回傳排隊等候的非同步作業數。圖 1 中因只呼叫一個 BeginRead，故 yield return 回傳值 1。此陳述式會暫停迭代器方法，停止其中程式碼執行。
private static IEnumerator<Int32> ApmPatternWithIterator(
AsyncEnumerator ae, String pathname) {
using (FileStream fs = new FileStream(pathname, FileMode.Open,
FileAccess.Read, FileShare.Read, 8192, FileOptions.Asynchronous)) {
Byte[] data = new Byte[fs.Length];
fs.BeginRead(data, 0, data.Length, ae.End(), null);
yield return 1;
Int32 bytesRead = fs.EndRead(ae.DequeueAsyncResult());
ProcessData(data);
}
}
yield return 指定的數字，用於告知 AsyncEnumerator 需完成多少非同步作業才能恢復迭代器執行。迭代器暫停期間，所有啟動的非同步作業持續完成。每完成一個，收件箱就增加一個項目。當收件箱項目數等於 yield return 指定數字時，AsyncEnumerator 會恢復迭代器執行。
yield return 陳述式後，應呼叫對應的 EndXxx 方法。呼叫時需傳遞 IAsyncResult 物件，而 AsyncEnumerator 已將這些物件排隊在收件箱中，只需呼叫 DequeueAsyncResult 方法取出即可（先進先出 FIFO）。
EndXxx 回傳值後，即可直接在迭代器內部處理非同步作業結果！這非常棒，因為雖然執行非同步作業，卻擁有同步程式設計模型。
許多情況下，最好接著呼叫其他 BeginXxx 排隊下一個非同步作業。在迭代器中，可再呼叫 BeginXxx，然後在對應 EndXxx 後加上 yield return 1。如此即可輕鬆逐個執行一系列非同步作業，不阻塞任何執行緒，且所有程式碼皆以同步模型撰寫！
private static IEnumerator<Int32> HtmlToFile(AsyncEnumerator ae,
String url, String file) {
// 發行非同步 Web 請求作業
WebRequest webRequest = WebRequest.Create(url);
webRequest.BeginGetResponse(ae.End(), null);
yield return 1;
WebResponse webResponse;
try {
webResponse = webRequest.EndGetResponse(ae.DequeueAsyncResult());
}
catch (WebException e) {
Console.WriteLine("Failed to contact server: {0}", e.Message);
yield break;
}
using (webResponse) {
Stream webResponseStream = webResponse.GetResponseStream();
// 以 1KB 區塊讀取資料流並寫入檔案
Byte[] data = new Byte[1024];
using (FileStream fs = new FileStream(file, FileMode.Create,
FileAccess.Write, FileShare.None, data.Length,
FileOptions.Asynchronous)) {
fs.SetLength(webResponse.ContentLength);
while (true) {
// 發行非同步 Web 回應資料流讀取作業
webResponseStream.BeginRead(data, 0, data.Length,
ae.End(), null);
yield return 1;
// 取得 Web 回應資料流讀取作業結果
Int32 bytesRead = webResponseStream.EndRead(
ae.DequeueAsyncResult());
if (bytesRead == 0) break; // 資料流結束：關閉檔案並離開
// 發行非同步檔案寫入作業
fs.BeginWrite(data, 0, bytesRead, ae.End(), null);
yield return 1;
// 取得檔案寫入作業結果
fs.EndWrite(ae.DequeueAsyncResult());
}
}
}
}
呼叫迭代器非常簡單：建構 AsyncEnumerator 物件，呼叫其 Execute 方法並傳遞迭代器成員名稱。可向迭代器傳遞參數，也會傳遞 AsyncEnumerator 參考，使迭代器能呼叫 End 與 DequeueAsyncResult。呼叫 HtmlToFile 的程式碼如下：
AsyncEnumerator ae = new AsyncEnumerator();
ae.Execute(HtmlToFile(
ae,
"https://www.Wintellect.com/",
"LocalFile.html"));
在迭代器成員結束或執行 yield break 前，AsyncEnumerator 的 Execute 方法不會傳回。當主執行緒受阻時，執行緒池執行緒會在非同步作業完成時繼續執行迭代器。這意味著迭代器的每部分程式碼理論上都可由不同執行緒執行，因此迭代器程式碼不應依賴執行緒特定狀態（如執行緒區域儲存、UI 區域、主體或執行緒優先權）。
順帶一提，我不建議讓執行緒阻塞直到迭代器完成。但這樣較易說明 AsyncEnumerator 運作原理，也簡化測試與偵錯。實際上有方法讓 AsyncEnumerator 在執行迭代器時不阻塞。在 Windows Forms 或 WPF 應用程式中，若要從 GUI 執行緒呼叫迭代器以獲得延展性與回應性，此方法極為重要。下一部分將介紹如何非同步執行迭代器成員。
AsyncEnumerator 類別定義圖 3 所示的私有欄位。m_enumerator 指向您的迭代器成員，建構函式將其初始化為 null，Execute 再將其設為傳遞的值。m_inbox 指向 IAsyncResult 物件集合。m_waitAndInboxCounts 是包含兩個 16 位整數值的結構，稍後將說明其用法。
public partial class AsyncEnumerator {
// 參考迭代器成員的程式碼
private IEnumerator<Int32> m_enumerator;
// 已完成非同步作業的集合（收件箱）
private List<IAsyncResult> m_inbox =
new List<IAsyncResult>();
// 包含 Wait 與 Inbox 計數器的結構
private WaitAndInboxCounts m_waitAndInboxCounts;
}
如前一期專欄所述，使用迭代器可輕鬆讓編譯器建立實作 IEnumerator<T> 介面的類別。通常使用 foreach 陳述式遍歷 IEnumerator<T> 物件，但您隨時可在程式碼中顯式呼叫 MoveNext 方法與 Current 屬性，這正是 AsyncEnumerator 的作用。
AsyncEnumerator 的 Execute 方法內部會呼叫名為 ResumeIterator 的私有方法（圖 4），用於啟動迭代器，也可使迭代器從暫停狀態恢復執行。
private void ResumeIterator() {
Boolean continueIterating;
// 當還有作業要執行時...
while (continueIterating = m_enumerator.MoveNext()) {
// 取得迭代器回傳的值
UInt16 numberOpsToWaitFor = checked((UInt16) m_enumerator.Current);
// 若收件箱項目少於請求數，保持迭代器暫停
if (!m_waitAndInboxCounts.AtomicSetWait(numberOpsToWaitFor)) break;
// 收件箱項目足夠，迴圈繼續恢復迭代器
}
// 迭代器暫停，直接返回
if (continueIterating) return;
// 迭代器已結束，執行迭代器的 finally 程式碼
m_enumerator.Dispose();
}
當執行緒呼叫 ResumeIterator 時，會呼叫 IEnumerator<T>.MoveNext 喚醒並執行迭代器。若迭代器結束或執行 yield break，MoveNext 傳回 false，表示無作業可執行。若 continueIterating 設為 false，迴圈結束並呼叫 Dispose 執行清理，然後 ResumeIterator 返回。
若 MoveNext 傳回 true，表示迭代器已啟動部分非同步作業，並透過 yield return 暫停。此時 ResumeIterator 需知迭代器在 yield return 指定的數字，此數字指示恢復迭代器前應完成的非同步作業數。ResumeIterator 查詢列舉器物件的 Current 屬性取得此數字。
隨後 ResumeIterator 呼叫 WaitAndInboxCounts.AtomicSetWait 設定迭代器應等待的項目數。若 AtomicSetWait 發現收件箱項目少於等待計數，傳回 false，導致 ResumeIterator 返回（此執行緒無其他工作）。若發現收件箱項目大於或等於等待計數，傳回 true 並繼續迴圈，再次呼叫 MoveNext 讓迭代器恢復執行處理已完成作業。
在迭代器內部，所有 BeginXxx 方法為 AsyncCallback 傳遞相同方法。此私有方法名為 EnqueueAsyncResult，透過呼叫 AsyncEnumerator.End 取得，實作如下：
private void EnqueueAsyncResult(IAsyncResult result) {
// 將此項目加入收件箱
lock (m_inbox) { m_inbox.Add(result); }
// 收件箱計數加 1。若收件箱項目足夠，
// 此執行緒呼叫 ResumeIterator
if (m_waitAndInboxCounts.AtomicIncrementInbox())
ResumeIterator();
}
為確保執行緒安全，此方法將已完成作業的 IAsyncResult 加入收件箱，然後呼叫 WaitAndInboxCounts.AtomicIncrementInbox 使收件箱計數加一。若 AtomicIncrementInbox 發現收件箱項目少於等待計數，傳回 false，執行緒池執行緒返回池中執行其他工作。若發現項目數等於等待計數，傳回 true 並呼叫 ResumeIterator，進而呼叫列舉器 MoveNext 恢復迭代器執行以處理已完成作業。
每個 AsyncEnumerator 物件保留一個 16 位 Wait 計數與一個 16 位 Inbox 計數（範圍 0 到 65535）。Wait 計數指示恢復列舉器前必須完成的非同步作業數，由迭代器 yield return 設定。Inbox 計數維護已完成非同步作業數。因迭代器執行 yield return 時，可能有多執行緒同時執行非同步作業完成程式碼，故必須以執行緒安全方式修改與檢查這兩個計數。
以執行緒安全方式更新計數最快的方法，是將它們置於一個 Int32 中，再使用 Interlocked 方法更新。為提高可讀性與維護性，我定義 WaitAndInboxCounts 結構管理這兩個計數（圖 5）。此結構定義私有 Int32 欄位並提供內部成員，以執行緒安全方式操作 Int32。AsyncEnumerator 程式碼呼叫這些成員。
private struct WaitAndInboxCounts {
private const UInt16 c_MaxWait = 0xFFFF;
// Wait=高 16 位元，Inbox=低 16 位元
private Int32 m_waitAndInboxCounts;
private UInt16 Wait {
get { return (UInt16) (m_waitAndInboxCounts >> 16); }
set { m_waitAndInboxCounts = (Int32) ((value << 16) | Inbox); }
}
private UInt16 Inbox {
get { return (UInt16) m_waitAndInboxCounts; }
set { m_waitAndInboxCounts =
(Int32)((m_waitAndInboxCounts & 0xFFFF0000) | value); }
}
private WaitAndInboxCounts(Int32 waic) { m_waitAndInboxCounts = waic; }
private Int32 ToInt32() { return m_waitAndInboxCounts; }
internal void Initialize() { Wait = c_MaxWait; }
internal Boolean AtomicSetWait(UInt16 numberOpsToWaitFor) {
return InterlockedEx.Morph<Boolean, UInt16>(
ref m_waitAndInboxCounts,
numberOpsToWaitFor, SetWait);
}
private static Int32 SetWait(Int32 i, UInt16 numberOpsToWaitFor,
out Boolean shouldMoveNext) {
WaitAndInboxCounts waic = new WaitAndInboxCounts(i);
// 設定要等待的項目數
waic.Wait = numberOpsToWaitFor;
shouldMoveNext = (waic.Inbox >= waic.Wait);
// 收件箱是否有足夠項目可 MoveNext？
if (shouldMoveNext) {
// 從收件箱減去項目數
waic.Inbox -= waic.Wait;
// 下一次等待為無限期
waic.Wait = c_MaxWait;
}
return waic.ToInt32();
}
internal Boolean AtomicIncrementInbox() {
return InterlockedEx.Morph<Boolean, Object>(
ref m_waitAndInboxCounts,
null, IncrementInbox);
}
private static Int32 IncrementInbox(Int32 i, Object argument,
out Boolean shouldMoveNext) {
WaitAndInboxCounts waic = new WaitAndInboxCounts(i);
// 收件箱計數加 1
waic.Inbox++;
shouldMoveNext = (waic.Inbox == waic.Wait);
// 收件箱是否有足夠項目可 MoveNext？
if (shouldMoveNext) {
// 從收件箱減去項目數
waic.Inbox -= waic.Wait;
// 下一次等待為無限期
waic.Wait = c_MaxWait;
}
return waic.ToInt32();
}
}
兩個可用的內部方法為 AtomicSetWait 與 AtomicIncrementInbox，前文已說明其用途。下面說明其運作原理，這是一種非常有用的技術，適用於需原子方式使用多個值的許多情況。
在第一期「並發事件」專欄中，我介紹過使用 Interlocked.CompareExchange 以更多方式操作 Int32 的知名模式，包括對 Int32 執行原子位元 AND、OR、XOR 操作。擴展此模式執行原子乘、除、最小值、最大值等操作也很簡單。
我將此模式轉換為名為 Morph<TResult, TArgument> 的通用方法：
public static TResult Morph<TResult, TArgument>(ref Int32 target,
TArgument argument, Morpher<TResult, TArgument> morpher) {
TResult morphResult;
Int32 i, j = target;
do {
i = j;
j = Interlocked.CompareExchange(ref target,
morpher(i, argument, out morphResult), i);
} while (i != j);
return morphResult;
}
呼叫此方法時，傳遞（以參考方式）欲以原子方式操作其值的 Int32，另可傳遞供 Morpher 使用的其他參數，最後傳遞簽章符合 Morpher 委派的方法：
delegate Int32 Morpher<TResult, TArgument>(Int32 startValue,
TArgument argument, out TResult morphResult);
Morpher 方法接收目標 Int32 初始值，其他參數由 Morph 傳遞。Morpher 對 Int32 值應用所有演算法並回傳所需新值。若演算法執行期間另一執行緒未修改原始 Int32，則指派 Morpher 回傳值給目標。若另一執行緒修改了原始值，Morph 會再次呼叫 Morpher，傳遞新的 Int32 值（視為已被其他執行緒修改）。
Morpher 還必須設定輸出參數 morphResult。若 Morph 成功修改目標，此輸出參數為 Morph 回傳給呼叫者的值。最大好處在於 Morpher 程式碼不必是執行緒安全的，可有數千行長並多次使用；Morph 負責確保以執行緒安全方式操作原始 Int32 值。
有了 Morph 方法與 Morpher 委派，即可實作 WaitAndInboxCounts.AtomicSetWait 與 AtomicIncrementInbox。
內部上，AtomicSetWait 呼叫 Morph 傳遞 Morpher 方法，將新 Wait 計數設為迭代器 yield return 回傳值。若 Inbox 計數小於新 Wait 計數，原子方式設定新 Wait 計數，並向 Morph 呼叫者傳回 false，告知不應呼叫列舉器 MoveNext（迭代器指示需等待收件箱顯示更多項目）。
若 Morph 發現 Inbox 計數大於或等於新 Wait 計數，則從 Inbox 計數減去 Wait 計數（邏輯上將從 Inbox 刪除這些項），並將 Wait 計數設為最大值 (65535)，使加入收件箱的項（由其他執行緒加入）不會導致迭代器恢復執行。若成功執行所有操作，Morph 向呼叫者傳回 true，此執行緒呼叫列舉器 MoveNext 恢復迭代器執行以處理等待中的已完成項目。
AtomicIncrementInbox 內部呼叫 Morph 傳遞 Morpher 使 Inbox 計數加 1。若 Inbox 計數與目前 Wait 計數不符，原子方式設定 Wait 計數並傳回 false。若 Morpher 發現 Inbox 計數等於目前 Wait 計數，則從 Inbox 減去 Wait，將 Wait 設為最大值，並傳回 true 讓執行緒呼叫 MoveNext 恢復迭代器。
WaitAndInboxCounts 結構還定義其他私有成員供內部使用，以非執行緒安全方式處理 Int32 值。私有建構函式可將 Int32 轉換為 WaitAndInboxCounts 結構。私有 ToInt32 回傳結構內部 Int32 值。另有私有 Wait 與 Inbox 屬性，用於設定 Int32 欄位的最小與最大 16 位值。
使用 C# 迭代器執行非同步處理是平行程式設計相關諸多挑戰之一。Microsoft 已制定多項開發計畫應對這些挑戰，其中之一為並發與協調執行階段 (CCR)，一個更通用的框架，可解決非同步與平行相關的各種問題。CCR 包含使用 C# 迭代器執行非同步列舉的實作。例如，以下程式碼展示在 CCR 中如何使用迭代器非同步處理檔案：
IEnumerator<ITask> CcrIterator(string pathname) {
var resultPort = new Port<IAsyncResult>();
using (FileStream fs = new FileStream(pathname,
FileMode.Open, FileAccess.Read, FileShare.Read,
8192, FileOptions.Asynchronous)) {
Byte[] data = new Byte[fs.Length];
fs.BeginRead(data, 0, data.Length, resultPort.Post, null);
yield return resultPort;
IAsyncResult ar = resultPort;
Int32 bytesRead = fs.EndRead(ar); }
ProcessData(data);
}
除支援迭代器外，CCR 還針對例外處理與協調原語（如仲裁程序）提供擴展支援。
許多開發人員了解非同步程式設計在實現高效能、可延展應用程式與元件方面具有重要價值。但在實踐中，許多程式設計師發現非同步程式設計模型使用起來相當困難，寧願繞開它而接受應用程式的低效能。我將 AsyncEnumerator 類別與 C# 迭代器結合所希望達到的目的，正是使非同步程式設計易於使用，得到程式設計師青睞，進而透過減少執行緒與內容切換數來改進應用程式效能與延展性。
