目錄
使用 AsyncEnumerator 類
AsyncEnumerator 的體系結構
使用 Wait 和 Inbox 計數器
線程安全變形
了解更多信息
在上期專欄 (並發事件: 通過 C# 簡化 APM) 中，，我講述了一些有關如何使用新的 C# 語言功能（ 匿名方法、lambda 表達式和迭代器）簡化異步編程的理念。在專欄最後，我說明了如何借助 C# 迭代器 使用同步編程模型完成異步編程。圖 1 顯示了示例迭代器。不過，僅使用常規的 C# foreach 語句是無 法執行迭代器的代碼的，因為此代碼始終由調用 foreach 的線程執行，並且是連續執行，不會暫停來完 成異步 I/O 操作。
在本期專欄中，我將介紹我的 AsyncEnumerator 類，它可以智能地驅動迭代器，使不同的線程池線程 能夠在不同的時間執行代碼，我可以確保迭代器只有在完成異步 I/O 操作後才會執行下一次迭代。此外 ，我還會介紹 AsyncEnumerator 類的體系結構及其工作原理。
使用 AsyncEnumerator 類
下面是 AsyncEnumerator 類的定義：
public class AsyncEnumerator {
// Methods called by code outside of the iterator
public AsyncEnumerator();
public void Execute(IEnumerator<Int32> enumerator);
// Methods called by code inside the iterator
public AsyncCallback End();
public IAsyncResult DequeueAsyncResult();
}
AsyncEnumerator 類的用法相當簡單。首先，我們來介紹如何實現您的迭代器成員，然後再介紹如何 調用。
將迭代器成員定義為可以接受所需的任何參數，並添加一個附加參數，該參數是對 AsyncEnumerator 對象的引用。返回 Int32 集合時您的迭代器成員必須設為原型；換句話說，它的返回類型必須是 IEnumerator<Int32>。
然後，在迭代器成員內部，通過調用相應的 BeginXxx 方法啟動每個異步操作。您知道，如果調用 BeginXxx 方法，則必須將完成異步操作時應調用的方法的名稱傳遞給它。
您可以調用 AsyncEnumerator 對象的 End 方法，而不需要定義自己的方法。End 方法會返回一個 AsyncCallback 代理，此代理標識在 AsyncEnumerator 類中定義的私有方法。因此，每個異步操作完成 時，AsyncEnumerator 對象中的代碼都會得到通知。然後，此代碼會將已完成的操作的 IAsyncResult 對 象放到 List<IAsyncResult> 對象（我稱它為 inbox，即收件箱）中。
在您的代碼中，在調用 BeginXxx 方法之後，放置一個 yield return 語句，用於返回排隊等候的異 步操作數。在圖 1 中，因為我只調用了一個 BeginXxx 方法 (BeginRead)，所以 yield return 語句返 回值 1。yield return 語句將暫停您的迭代器方法，並停止執行其中的代碼。
圖 1 C# 迭代器
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 語句返回非 1 值的其他示例，但對於許多應用程序而言，還是 返回 1 比較合適。在 yield return 語句中指定的數字用於告知 AsyncEnumerator 對象需要完成多少個 異步操作才能恢復迭代器的執行。因此，當迭代器暫停時，您啟動的所有異步操作都會繼續完成。完成每 個異步操作後，AsyncEnumerator 的收件箱中都會增加一個條目。當收件箱中的項目數等於您在 yield return 語句中指定的數字時，AsyncEnumerator 對象將恢復迭代器的執行，並允許繼續執行其代碼。
深入分析：AsyncEnumeration 環境
使用 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 還針對異常處理和協調原語（如仲裁程序）提供了擴展支持。
— 由 Howard Dierking 主編
在 yield return 語句後，應調用相應的 EndXxx 方法。調用 EndXxx 方法時，需要為其傳遞一個 IAsyncResult 對象。幸運的是，AsyncEnumerator 對象已在其收件箱中將這些對象排隊，因此，您只需 通過調用 AsyncEnumerator 對象的 DequeueAsyncResult 方法取出這些對象即可。這些項目將使用先進 先出 (FIFO) 算法從收件箱中取出。
EndXxx 方法返回值後，即可直接在迭代器內部處理異步操作的結果！這太棒了，因為盡管您正在執行 異步操作，但卻擁有了一個同步編程模型。
在許多情況下，都最好是現在通過調用其他 BeginXxx 方法對另一個異步操作進行排隊。在迭代器中 ，您可以調用另一個 BeginXxx 方法，然後在調用相應的 EndXxx 方法後添加一個 yield return 1 語句 。這樣，您就可以輕松地逐個執行一系列異步操作，而不阻止任何線程。此外，您還可以使用同步編程模 型編寫所有代碼！
例如，圖 2 中顯示的 HtmlToFile 迭代器可從 Web 服務器中讀取以 1 千字節 (KB) 為單位的 HTML 塊，並將其保存到本地文件中。調用迭代器非常簡單：先構建一個 AsyncEnumerator 對象，然後調用其 Execute 方法，向其傳遞迭代器成員的名稱即可。您可以向迭代器成員傳遞參數，還會向迭代器傳遞對 AsyncEnumerator 對象的引用，使迭代器可以調用 End 和 DequeueAsyncResult 方法。調用 HtmlToFile 迭代器的代碼如下所示：
AsyncEnumerator ae = new AsyncEnumerator();
ae.Execute(HtmlToFile(
ae,
"http://www.Wintellect.com/",
"LocalFile.html"));
圖 2 HtmlToFile 迭代器
private static IEnumerator<Int32> HtmlToFile(AsyncEnumerator ae,
String url, String file) {
// Issue asynchronous web request operation
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();
// Read the stream data and write it to a file in 1KB chunks
Byte[] data = new Byte[1024];
using (FileStream fs = new FileStream(file, FileMode.Create,
FileAccess.Write, FileShare.None, data.Length,
FileOptions.Asynchronous)) {
// See support.microsoft.com/kb/156932
fs.SetLength(webResponse.ContentLength);
while (true) {
// Issue asynchronous web response stream read operation
webResponseStream.BeginRead(data, 0, data.Length,
ae.End(), null);
yield return 1;
// Get result of web response stream read operation
Int32 bytesRead = webResponseStream.EndRead(
ae.DequeueAsyncResult());
if (bytesRead == 0) break; // Stream end: close file & exit
// Issue asynchronous file write operation
fs.BeginWrite(data, 0, bytesRead, ae.End(), null);
yield return 1;
// Get result of file write operation
fs.EndWrite(ae.DequeueAsyncResult());
}
}
}
}
在迭代器成員退出或執行 yield break 語句前，AsyncEnumerator 的 Execute 方法不會返回值。當 主線程受阻時，線程池線程將在異步操作完成時繼續執行迭代器。上述所有情形都意味著，迭代器的每部 分代碼理論上都可以由其他線程執行，因此迭代器代碼不應依賴於線程特定的狀態，如線程本地存儲、區 域、UI 區域、主體或線程優先級。
順便說一下，我並不傾向於讓線程直到迭代器完成運行時才調 用 Execute 塊。但這樣卻可以更簡單地說明 AsyncEnumerator 類的工作原理，並可以簡化試驗和調試。 但需要指出的是，有一種方法可以讓 AsyncEnumerator 在執行迭代器時不進行阻止。在 Windows® Form 或 Windows Presentation Foundation (WPF) 應用程序中，如果希望從 GUI 線程調用迭代器來獲 得可伸縮性和響應性，此方法將非常重要。在本專欄下一部分中，我將為您介紹如何異步執行迭代器成員 。
AsyncEnumerator 的體系結構
AsyncEnumerator 類可定義圖 3 中顯示的私有字段。 m_enumerator 字段指您的迭代器成員，構造函數會將此字段初始化為空，然後 Execute 會將此字段設置 為傳遞給它的值。m_inbox 字段指 IAsyncResult 對象的集合。m_waitAndInboxCounts 字段是一個包含 有兩個 16 位整數值的結構；在本專欄的稍後部分，我將介紹如何使用這些值。
圖 3 AsyncEnumerator 字段
public partial class AsyncEnumerator {
// Refers to the iterator member's code
private IEnumerator<Int32> m_enumerator;
// Collection of completed async operations (the inbox)
private List<IAsyncResult> m_inbox =
new List<IAsyncResult>();
// Structure with Wait & Inbox counters
private WaitAndInboxCounts m_waitAndInboxCounts;
}
正如我在上期專欄中所介紹的，要使編譯器創建用於實現 IEnumerator<T> 接口的類，使用迭 代器確實是一種簡便方式。通常，您會使用 foreach 語句遍歷 IEnumerator<T> 對象。不過，無 論何時您都可以顯式調用代碼中的 IEnumerator<T> MoveNext 方法和 Current 屬性，這正是 AsyncEnumerator 類的作用。
AsyncEnumerator 的 Execute 方法將在內部調用名為 ResumeIterator 的私有方法（如圖 4 中所示 ）。此方法用於啟動迭代器，還可用於使迭代器從暫停狀態恢復執行。
圖 4 ResumeIterator
private void ResumeIterator() {
Boolean continueIterating;
// While there are more operations to perform...
while (continueIterating = m_enumerator.MoveNext()) {
// Get the value returned from the enumerator
UInt16 numberOpsToWaitFor = checked((UInt16) m_enumerator.Current);
// If inbox has fewer items than requested, keep iterator suspended
if (!m_waitAndInboxCounts.AtomicSetWait(numberOpsToWaitFor)) break;
// Inbox has enough items, loop to resume the iterator
}
// The iterator is suspended, just return
if (continueIterating) return;
// The iterator has exited, execute the iterator's finally code
m_enumerator.Dispose();
}
當線程調用 ResumeIterator 時，它將調用 IEnumerator<T> MoveNext 方法，用於喚醒並運行 迭代器。如果迭代器退出或執行 yield break 語句，IEnumerator<T> MoveNext 方法將返回 false，表明它沒有可執行的操作。如果將 ResumeIterator 方法的 continueIterating 變量設置為 false，它將退出此循環，並調用 IEnumerator<T> 對象的 Dispose 方法，允許執行清理。然後， ResumeIterator 方法將返回值，因為迭代器已完成其所有工作。
另一方面，如果枚舉器對象的 MoveNext 方法返回 true，則表明它已啟動了一些異步操作，並已通過 執行 yield return 語句暫停運行。現在，ResumeIterator 方法需要知道 IEnumerator<T> 對象 在其 yield return 語句中指定的數字。此數字指示在恢復運行迭代器之前應完成的異步操作數。為了獲 得此數字，ResumeIterator 將查詢枚舉器對象的 Current 屬性；此屬性將返回上次通過 yield return 語句指定的值。
隨後，ResumeIterator 將調用 WaitAndInboxCounts AtomicSetWait 方法，該方法用於設置迭代器應 等待的項目數。如果 AtomicSetWait 發現收件箱中的項目數少於此等待計數，則 AtomicSetWait 將返回 false，這會導致 ResumeIterator 返回值，因為此線程沒有其他任務可執行。如果 AtomicSetWait 發現 收件箱中的項目數大於或等於等待計數，則 AtomicSetWait 將返回 true 並繼續循環，從而再次調用枚 舉器對象的 MoveNext 方法，並允許迭代器恢復執行，以便處理已完成的操作。
在迭代器內部，所有 BeginXxx 方法都會為 AsyncCallback 方法傳遞相同的方法。此私有方法稱為 EnqueueAsyncResult，可通過調用 AsyncEnumerator 對象的 End 方法獲取，並按以下方式實現：
private void EnqueueAsyncResult(IAsyncResult result) {
// Add this item to the inbox
lock (m_inbox) { m_inbox.Add(result); }
// Add 1 to inbox count. If inbox has enough items
// in it; this thread calls ResumeIterator
if (m_waitAndInboxCounts.AtomicIncrementInbox())
ResumeIterator();
}
在線程安全方面，此方法會向收件箱中添加已完成操作的 IAsyncResult 對象，然後調用 WaitAndInboxCounts AtomicIncrementInbox 方法，使收件箱中的項目數加一。如果 AtomicIncrementInbox 發現收件箱中的項目數少於等待計數，則 AtomicIncrementInbox 將返回 false ，這將使線程池線程返回池中以便用來執行其他工作。
如果 AtomicIncrementInbox 發現收件箱中的項目數等於等待計數，則 AtomicIncrementInbox 將返 回 true 並調用 AsyncEnumerator 的 ResumeIterator 方法，該方法隨後即會調用枚舉器對象的 MoveNext 方法恢復執行迭代器，以便接下來能夠處理已完成的操作。
使用 Wait 和 Inbox 計數器
每個 AsyncEnumerator 對象都可保留一個 16 位 Wait 計數和一個 16 位 Inbox 計數；當然，兩種 計數的范圍都是從 0 到 65535。Wait 計數指示在恢復枚舉器對象之前必須完成的異步操作數。迭代器通 過其 yield return 語句設置此值。Inbox 計數維護已完成的異步操作數。必須使用線程安全的技術修改 和檢查 Wait 和 Inbox 計數，因為在迭代器執行其 yield return 語句時，可能有多個線程也在同時執 行異步操作完成代碼。
以線程安全的方式更新這些計數的最快方法是將它們置於一個 Int32 中，然後使用 Interlocked 方 法更新此 Int32。為了使此代碼更具可讀性並且更易維護，我決定定義一個 WaitAndInboxCounts 結構來 管理這兩種計數（請參見圖 5）。此結構可定義一個私有 Int32 字段並提供內部成員，從而以線程安全 的方式使用 Int32。AsyncEnumerator 代碼將調用這些成員。
圖 5 線程安全的 Wait 和 Inbox 計數
private struct WaitAndInboxCounts {
private const UInt16 c_MaxWait = 0xFFFF;
// Wait=High 16 bits, Inbox=low-16 bits
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);
// Set the number of items to wait for
waic.Wait = numberOpsToWaitFor;
shouldMoveNext = (waic.Inbox >= waic.Wait);
// Does the inbox contain enough items to MoveNext?
if (shouldMoveNext) {
// Subtract the number of items from the inbox
waic.Inbox -= waic.Wait;
// The next wait is indefinite
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);
// Add 1 to the inbox count
waic.Inbox++;
shouldMoveNext = (waic.Inbox == waic.Wait);
// Does the inbox contain enough items to MoveNext?
if (shouldMoveNext) {
// Subtract the number of items from the inbox
waic.Inbox -= waic.Wait;
// The next wait is indefinite
waic.Wait = c_MaxWait;
}
return waic.ToInt32();
}
}
這兩種可用的內部方法是 AtomicSetWait 和 AtomicIncrementInbox。在本專欄的前面部分，我介紹 了這些方法的用途。下面，我介紹一下它們的工作原理，因為這是一種非常有用的技術，適用於需要以原 子方式使用多個值的許多情況。
我相信，大部分人都知道 Microsoft® .NET Framework 類庫 的 Interlocked 類提供了以線程安全方式使用 Int32 值的多種方法，這些方法可以使 Int32 按步長 1 進行遞增、使 Int32 按步長 1 進行遞減、向 Int32 添加正值或負值、將 Int32 更改為新值、檢查 Int32 是否為特定值以及將其更改為另一個值（如果是特定值）。簡而言之，Interlocked 方法極為有用 ，但它們只提供了幾種使用 Int32 的方法。
線程安全變形
在我第一次的“並發事件 ”專欄(並發事件: 通過 C# 簡化 APM)中，針對如何使用 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 方法返回的值。如果當算法執行另一線程的代碼時該線程更改了原始 Int32 值，則 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 計數，則 Morph 方法將從 Inbox 計數中減 去 Wait 計數（因為邏輯上將從 Inbox 中刪除這些項），並會將 Wait 計數設置為最大值 (65535)，以 便添加到收件箱中的項（由其他線程添加）不會導致迭代器恢復執行。不過，如果成功執行了上述所有操 作，Morph 方法將向其調用方返回 true，並且此線程將調用枚舉器對象的 MoveNext 方法，恢復迭代器 的執行，以便處理正在等待的已完成項目。
在內部，AtomicIncrementInbox 將調用 Morph，從而向其傳遞 Morpher 方法，使 Inbox 計數加 1。 如果 Inbox 計數與當前的 Wait 計數不符，則將以原子方式設置 Wait 計數，並向 Morph 的調用方返回 false，告知調用方不應調用枚舉器對象的 MoveNext 方法，因為迭代器指示它需要等待收件箱中顯示更 多的項目。
對於 Morph，如果 Morpher 方法發現 Inbox 計數等於當前的 Wait 計數，則 Morpher 方法將從 Inbox 計數中減去 Wait 計數（因為邏輯上將從 Inbox 中刪除這些項），並會將 Wait 計數設置為最大 值，以便添加到收件箱中的項（由其他線程添加）不會導致迭代器恢復執行。不過，如果成功執行了上述 所有操作，Morph 方法將向其調用方返回 true，並且此線程將調用枚舉器對象的 MoveNext 方法，恢復 迭代器的執行，以便處理正在等待的已完成項目。
WaitAndInboxCounts 結構還定義了其他一些私有成員供內部使用，用於以非線程安全方式處理 Int32 值。私有構造函數可有效地將 Int32 轉換為 WaitAndInboxCounts 結構。私有的 ToInt32 方法將返回 WaitAndInboxCounts 結構內部的 Int32 值。此外，還有一些私有的 Wait 和 Inbox 屬性，用於設置 Int32 字段的最小和最大 16 位值。
了解更多信息
許多開發人員都了解異步編程在實現高性能、可伸縮的應用程序和組件方面具有重要價值。但在實踐 中，許多編程人員都發現異步編程模型使用起來相當困難，因此寧願繞開它而接受應用程序的低性能；而 我將 AsyncEnumerator 類與 C# 迭代器結合所希望達到的目的，正是使異步編程易於使用，以得到編程 人員的青睐，進而通過減少線程和上下文開關數來改進其應用程序的性能和可伸縮性。
