Streamهای ناهمگام (Asynchronous Streams)

این قابلیت بر دو Interface زیر بنا شده است که همتای ناهمگام Interfaceهای Enumeration هستند که در «Enumeration and Iterators» صفحهٔ 203 توضیح دادیم:

public interface IAsyncEnumerable<out T>
{
  IAsyncEnumerator<T> GetAsyncEnumerator (...);
}
public interface IAsyncEnumerator<out T>: IAsyncDisposable
{
  T Current { get; }
  ValueTask<bool> MoveNextAsync();
}

ValueTask<T> یک Struct است که Task<T> را Wrap می‌کند و از نظر Behavior شبیه آن است، اما وقتی Task به‌صورت همگام Complete شود ــ اتفاقی که هنگام Enumerate کردن Sequence می‌تواند زیاد رخ دهد ــ اجرای کارآمدتری فراهم می‌کند. برای تفاوت‌ها «ValueTask<T>» در صفحهٔ 679 را ببینید. IAsyncDisposable نسخهٔ ناهمگام IDisposable است و اگر Interfaceها را دستی پیاده‌سازی کنید فرصتی برای Cleanup فراهم می‌کند:

public interface IAsyncDisposable
{
  ValueTask DisposeAsync();
}

برای تولید Asynchronous Stream، Methodی می‌نویسید که اصول Iteratorها و Methodهای ناهمگام را ترکیب کند. یعنی Method باید هم yield return و هم await داشته باشد و IAsyncEnumerable<T> برگرداند:

async IAsyncEnumerable<int> RangeAsync (
  int start, int count, int delay)
{
  for (int i = start; i < start + count; i++)
  {
    await Task.Delay (delay);
    yield return i;
  }
}

برای مصرف Asynchronous Stream از Statement با نام await foreach استفاده کنید:

await foreach (var number in RangeAsync (0, 10, 500))
  Console.WriteLine (number);

توجه کنید Data به‌طور پیوسته، هر 500 Millisecond ــ یا در دنیای واقعی هر زمان که در دسترس شود ــ می‌رسد. آن را با Construct مشابهی مبتنی بر Task<IEnumerable<T>> مقایسه کنید که تا آماده‌شدن آخرین قطعهٔ Data چیزی برنمی‌گرداند:

static async Task<IEnumerable<int>> RangeTaskAsync (int start, int count,
                                                    int delay)
{
  List<int> data = new List<int>();
  for (int i = start; i < start + count; i++)
  {
    await Task.Delay (delay);
    data.Add (i);
  }
  return data;
}

روش مصرف آن با Statement از نوع foreach چنین است:

foreach (var data in await RangeTaskAsync(0, 10, 500))
  Console.WriteLine (data);

Query کردن IAsyncEnumerable<T>

Package با نام System.Linq.Async در NuGet، Query Operatorهای LINQ را تعریف می‌کند که روی IAsyncEnumerable<T> کار می‌کنند و اجازه می‌دهند تقریباً همان‌طور که روی IEnumerable<T> Query می‌نویسید، اینجا هم Query بنویسید.

برای نمونه، می‌توانیم روی Method با نام RangeAsync که در بخش قبل تعریف کردیم چنین Query بنویسیم:

IAsyncEnumerable<int> query =
  from i in RangeAsync (0, 10, 500)
  where i % 2 == 0   // Even numbers only.
  select i * 10;     // Multiply by 10.
await foreach (var number in query)
  Console.WriteLine (number);

خروجی 0، 20، 40 و به همین ترتیب است.

IAsyncEnumerable<T> در ASP.NET Core

Controller Actionهای ASP.NET Core اکنون می‌توانند IAsyncEnumerable<T> برگردانند. چنین Methodهایی باید با async علامت‌گذاری شوند. برای مثال:

[HttpGet]
public async IAsyncEnumerable<string> Get()
{

    using var dbContext = new BookContext();
    await foreach (var title in dbContext.Books
                                         .Select(b => b.Title)
                                         .AsAsyncEnumerable())
       yield return title;
}

Methodهای ناهمگام در WinRT

اگر UWP Application توسعه می‌دهید باید با Typeهای WinRT تعریف‌شده در Operating System کار کنید. معادل Task در WinRT، IAsyncAction است و معادل Task<TResult>، IAsyncOperation<TResult>. برای Operationهایی که Progress گزارش می‌کنند، معادل‌ها IAsyncActionWithProgress<TProgress> و IAsyncOperationWithProgress<TResult, TProgress> هستند. همهٔ آن‌ها در Namespace با نام Windows.Foundation تعریف شده‌اند.

با Extension Method با نام AsTask می‌توانید هرکدام را به Task یا Task<TResult> تبدیل کنید:

Task<StorageFile> fileTask = KnownFolders.DocumentsLibrary.CreateFileAsync
                             ("test.txt").AsTask();

یا می‌توانید مستقیم Awaitشان کنید:

StorageFile file = await KnownFolders.DocumentsLibrary.CreateFileAsync
                         ("test.txt");

Method با نام AsTask همچنین Overloadی دارد که Cancellation Token می‌پذیرد؛ «Cancellation» در صفحهٔ 681 را ببینید. هنگام Chain شدن با Variantهای WithProgress نیز می‌تواند Object از نوع IProgress<T> بپذیرد؛ «Progress Reporting» در صفحهٔ 683.

Asynchrony و Synchronization Contextها

قبلاً دیدیم وجود Synchronization Context در Post کردن Continuationها مهم است. چند راه ظریف‌تر دیگر نیز وجود دارد که چنین Contextهایی در توابع ناهمگامِ Void-returning وارد بازی می‌شوند. این رفتار مستقیماً نتیجهٔ بسط Compiler در C# نیست، بلکه حاصل Typeهای Async*MethodBuilder در Namespace با نام System.CompilerServices است که Compiler برای بسط توابع ناهمگام از آن‌ها استفاده می‌کند.

Post کردن Exception

در Rich Client Applicationها معمول است برای پردازش Exceptionهای مدیریت‌نشده‌ای که روی UI Thread پرتاب می‌شوند به Event مرکزی Exception Handling ــ مانند Application.DispatcherUnhandledException در WPF ــ تکیه شود. در ASP.NET Core نیز ExceptionFilterAttribute سفارشی در Method با نام ConfigureServices از Startup.cs کاری مشابه انجام می‌دهد.

در داخل، این مکانیزم‌ها UI Eventها ــ یا در ASP.NET Core، Pipeline مربوط به Methodهای پردازش Page ــ را در Blockهای try/catch خودشان Invoke می‌کنند.

توابع ناهمگام Top-level این موضوع را پیچیده می‌کنند. Event Handler زیر را برای Click یک Button در نظر بگیرید:

async void ButtonClick (object sender, RoutedEventArgs args)
{
  await Task.Delay(1000);
  throw new Exception ("Will this be ignored?");
}

وقتی Button کلیک می‌شود و Event Handler اجرا می‌شود، Execution پس از Statement از نوع await به‌طور عادی به Message Loop برمی‌گردد و Exceptionای که یک ثانیه بعد پرتاب می‌شود دیگر نمی‌تواند توسط Catch Block موجود در Message Loop گرفته شود.

برای کاهش این مشکل، AsyncVoidMethodBuilder در توابع ناهمگام Void-returning، Exceptionهای مدیریت‌نشده را می‌گیرد و اگر Synchronization Context وجود داشته باشد آن‌ها را به Context Post می‌کند؛ در نتیجه Eventهای Global Exception Handling همچنان Fire می‌شوند.

ظرافت جالب این است که فرقی ندارد Exception را پیش از await پرتاب کنید یا پس از آن. بنابراین در مثال زیر Exception به Synchronization Context ــ اگر وجود داشته باشد ــ Post می‌شود و هرگز مستقیم به Caller برنمی‌گردد:

async void Foo() { throw null; await Task.Delay(1000); }

اگر Synchronization Context وجود نداشته باشد، Exception روی Thread Pool Propagate می‌شود و Application را Terminate می‌کند.

دلیل اینکه Exception مستقیم به Caller پرتاب نمی‌شود حفظ Predictability و Consistency است. در مثال زیر، InvalidOperationException بدون توجه به someCondition همیشه اثر یکسانی دارد: Task حاصل را Fault می‌کند:

async Task Foo()
{
  if (someCondition) await Task.Delay (100);
  throw new InvalidOperationException();
}

Iteratorها نیز رفتاری مشابه دارند:

IEnumerable<int> Foo() { throw null; yield return 123; }

در این مثال Exception هرگز مستقیماً به Caller پرتاب نمی‌شود؛ فقط هنگامی که Sequence Enumerate شود Exception پرتاب خواهد شد.

OperationStarted و OperationCompleted

اگر Synchronization Context وجود داشته باشد، توابع ناهمگام Void-returning هنگام ورود به تابع Method با نام OperationStarted و هنگام پایان تابع Method با نام OperationCompleted را نیز فراخوانی می‌کنند.

Override کردن این Methodها هنگام نوشتن Synchronization Context سفارشی برای Unit Test کردن Methodهای ناهمگام Void-returning مفید است. این موضوع در Parallel Programming Blog مایکروسافت توضیح داده شده است.

بهینه‌سازی‌ها (Optimizations)

Complete شدن به‌شکل همگام

یک تابع ناهمگام می‌تواند پیش از Await کردن Return کند. Method زیر را در نظر بگیرید که Download Web Pageها را Cache می‌کند:

static Dictionary<string,string> _cache = new Dictionary<string,string>();
async Task<string> GetWebPageAsync (string uri)
{
  string html;
  if (_cache.TryGetValue (uri, out html)) return html;
  return _cache [uri] = 
    await new WebClient().DownloadStringTaskAsync (uri);
}

اگر URI از قبل در Cache وجود داشته باشد، بدون اینکه Awaitی رخ داده باشد Execution به Caller برمی‌گردد و Method یک Task از قبل Signal‌شده برمی‌گرداند. به این حالت Synchronous Completion گفته می‌شود.

وقتی Taskی را Await می‌کنید که به‌شکل همگام Complete شده، Execution به Caller برنمی‌گردد تا از طریق Continuation دوباره Bounce کند؛ بلکه بلافاصله Statement بعدی را اجرا می‌کند. Compiler این Optimization را با بررسی Property با نام IsCompleted روی Awaiter پیاده‌سازی می‌کند. یعنی هنگامی که چنین می‌نویسید:

Console.WriteLine (await GetWebPageAsync ("http://oreilly.com"));

Compiler کدی تولید می‌کند که در صورت Synchronous Completion، Continuation را میان‌بُر می‌زند:

var awaiter = GetWebPageAsync().GetAwaiter();
if (awaiter.IsCompleted)
  Console.WriteLine (awaiter.GetResult());
else
  awaiter.OnCompleted (() => Console.WriteLine (awaiter.GetResult());

نوشتن Method ناهمگامی که هرگز Await نمی‌کند نیز کاملاً قانونی است، هرچند Compiler Warning می‌دهد:

async Task<string> Foo() { return "abc"; }

چنین Methodهایی هنگام Override کردن Virtual/Abstract Methodها می‌توانند مفید باشند، وقتی Implementation شما اتفاقاً به Asynchrony نیاز ندارد. نمونه‌اش Methodهای ReadAsync/WriteAsync در MemoryStream است؛ فصل 15 را ببینید. راه دیگر برای همان نتیجه استفاده از Task.FromResult است که Task از قبل Signal‌شده برمی‌گرداند:

Task<string> Foo() { return Task.FromResult ("abc"); }

Method با نام GetWebPageAsync اگر از UI Thread فراخوانی شود به‌طور ضمنی Thread-safe است؛ می‌توانید آن را چند بار پشت سر هم Invoke کنید و چند Download Concurrent آغاز شود، بدون اینکه برای محافظت از Cache به Locking نیاز باشد. با این حال اگر Callها برای یک URI یکسان باشند، چند Download تکراری شروع می‌شوند که در نهایت همان Cache Entry را Update می‌کنند و آخرین مورد برنده می‌شود. این خطا نیست، اما کارآمدتر است اگر Callهای بعدی برای همان URI بتوانند به‌صورت ناهمگام منتظر Result مربوط به Request در حال اجرا بمانند.

راه ساده‌ای بدون Lock یا Signaling Construct وجود دارد: به‌جای Cache از Stringها، Cache از «Future»ها یعنی Task<string> می‌سازیم:

static Dictionary<string,Task<string>> _cache = 
   new Dictionary<string,Task<string>>();
Task<string> GetWebPageAsync (string uri)
{
  if (_cache.TryGetValue (uri, out var downloadTask)) return downloadTask;
  return _cache [uri] = new WebClient().DownloadStringTaskAsync (uri);
}

توجه کنید Method را async علامت نمی‌زنیم، چون Taskی را که از Method مربوط به WebClient می‌گیریم مستقیم Return می‌کنیم.

اگر GetWebPageAsync را با URI یکسان چند بار Call کنیم، حالا تضمین شده همان Object از نوع Task<string> را دریافت کنیم. این مزیت اضافه را هم دارد که Load مربوط به Garbage Collection را کم می‌کند. و اگر Task Complete شده باشد، به‌کمک Optimization Compiler که توضیح دادیم Await کردن آن ارزان است.

می‌توانیم مثال را گسترش دهیم تا بدون محافظت Synchronization Context نیز Thread-safe باشد؛ کافی است کل Method Body را Lock کنیم:

lock (_cache)
  if (_cache.TryGetValue (uri, out var downloadTask))
    return downloadTask;
  else
    return _cache [uri] = new WebClient().DownloadStringTaskAsync (uri);
}

این کار جواب می‌دهد چون در تمام مدت Download کردن Page Lock نگه نمی‌داریم ــ که Concurrency را خراب می‌کرد ــ بلکه فقط برای مدت کوتاه بررسی Cache، شروع Task جدید در صورت نیاز و Update کردن Cache با آن Task Lock می‌کنیم.

ValueTask<T>

گفتیم Compiler یک Expression از نوع await روی Taskی که به‌صورت همگام Complete شده را با میان‌بُر زدن Continuation و رفتن مستقیم به Statement بعدی Optimize می‌کند. اگر Synchronous Completion به‌خاطر Cache باشد، Cache کردن خود Task می‌تواند راه‌حل ظریف و کارآمدی باشد.

اما Cache کردن Task در همهٔ سناریوهای Synchronous Completion عملی نیست. گاهی باید Task تازه‌ای Instantiate شود و همین یک Inefficiency بسیار کوچک ایجاد می‌کند. علت این است که Task و Task<T> Reference Type هستند و Instantiate شدنشان Allocation روی Heap و Collection بعدی می‌طلبد.

شکل افراطی Optimization این است که کدی Allocation-free بنویسید؛ یعنی هیچ Reference Typeای Instantiate نکند و هیچ بار اضافه‌ای روی Garbage Collection نگذارد. برای پشتیبانی از این Pattern، Structهای ValueTask و ValueTask<T> معرفی شده‌اند و Compiler اجازه می‌دهد به‌جای Task و Task<T> از آن‌ها استفاده کنید:

async ValueTask<int> Foo() { ... }

اگر Operation به‌صورت همگام Complete شود، Await کردن ValueTask<T> بدون Allocation است:

int answer = await Foo();   // (Potentially) allocation-free

اگر Operation به‌صورت همگام Complete نشود، ValueTask<T> در پشت صحنه یک Task<T> معمولی ایجاد می‌کند و Await را به آن Forward می‌کند؛ در نتیجه چیزی به دست نمی‌آید. با Method با نام AsTask می‌توانید ValueTask<T> را به Task<T> عادی تبدیل کنید.

نسخهٔ Nongeneric با نام ValueTask نیز وجود دارد که مشابه Task است.

احتیاط‌ها هنگام استفاده از ValueTask<T>

ValueTask<T> از این جهت نسبتاً غیرمعمول است که فقط به دلایل Performance به‌شکل Struct تعریف شده و در نتیجه Semanticهای Value Type که برای این کار مناسب نیستند می‌توانند غافلگیرکننده باشند.

برای جلوگیری از رفتار نادرست باید از موارد زیر پرهیز کنید:

• Await کردن یک ValueTask<T> یکسان چند بار.
• فراخوانی .GetAwaiter().GetResult() پیش از Complete شدن Operation.

اگر لازم است این کارها را انجام دهید، .AsTask() را فراخوانی و روی Task حاصل کار کنید.

جلوگیری از Bounce بیش از حد

برای Methodهایی که بارها در Loop فراخوانی می‌شوند، با ConfigureAwait می‌توانید هزینهٔ Bounce مکرر به UI Message Loop را حذف کنید. این کار Task را وادار می‌کند Continuationها را به Synchronization Context برنگرداند و سربار را به هزینهٔ Context Switch نزدیک می‌کند؛ یا اگر Method مورد Await به‌صورت همگام Complete شود، حتی بسیار کمتر:

async void A() { ... await B(); ... }
async Task B()
{
  for (int i = 0; i < 1000; i++)
    await C().ConfigureAwait (false);
}
async Task C() { ... }

در نتیجه، برای Methodهای B و C از Model سادهٔ Thread Safety در UI Applicationها صرف‌نظر می‌کنیم؛ همان Modelی که کد را روی UI Thread اجرا می‌کند و فقط در Statement از نوع await امکان Preemption می‌دهد. اما Method با نام A تحت تأثیر قرار نمی‌گیرد و اگر روی UI Thread شروع شده باشد همان‌جا باقی می‌ماند.

این Optimization به‌ویژه هنگام نوشتن Libraryها مهم است: معمولاً به مزیت Thread Safety ساده‌شده نیاز ندارید، چون کد شما اغلب Shared State با Caller ندارد و به UI Controlها دسترسی نمی‌زند. در مثال ما منطقی است Method با نام C هم اگر بداند Operation احتمالاً Short-running است به‌صورت همگام Complete شود.