فصل ۲۱ — بخش دوم: Nonexclusive Locking و Semaphore

Semaphore برای محدودکردن concurrency مفید است؛ یعنی از اجرای هم‌زمان تعداد بیش از حد threadها در یک قسمت کد جلوگیری می‌کند. مثال زیر پنج thread را به یک «باشگاه» می‌فرستد که فقط سه thread هم‌زمان ظرفیت دارد:

class TheClub      // No door lists!
{
  static SemaphoreSlim _sem = new SemaphoreSlim (3); // Capacity of 3

  static void Main()
  {
    for (int i = 1; i <= 5; i++) new Thread (Enter).Start (i);
  }

  static void Enter (object id)
  {
    Console.WriteLine (id + " wants to enter");
    _sem.Wait();
    Console.WriteLine (id + " is in!");           // Only three threads
    Thread.Sleep (1000 * (int) id);               // can be here at
    Console.WriteLine (id + " is leaving");       // a time.
    _sem.Release();
  }
}

1 wants to enter
1 is in!
2 wants to enter
2 is in!
3 wants to enter
3 is in!
4 wants to enter
5 wants to enter
1 is leaving
4 is in!
2 is leaving
5 is in!

ساخت Semaphore با initial count صفر نیز قانونی است و بعداً می‌توان با Release count را بالا برد. دو نمونهٔ زیر از نظر رفتار معادل‌اند:

var semaphore1 = new SemaphoreSlim (3);
var semaphore2 = new SemaphoreSlim (0); semaphore2.Release (3);

یک Semaphore نام‌دار مانند Mutex می‌تواند میان processها مشترک باشد. متن منبع یادآوری می‌کند که named Semaphore فقط روی Windows در دسترس است، در حالی که named Mutex روی Unix نیز کار می‌کند.

Semaphore و Lock در کد Async

نگه‌داشتن یک lock در طول یک await غیرقانونی است:

lock (_locker)
{
  await Task.Delay (1000); // Compilation error
  ...
}

این کار از نظر مفهومی هم مناسب نیست، زیرا lock به thread تعلق دارد و thread پس از برگشت از await ممکن است عوض شود. Locking همچنین block می‌کند و در asynchronous programming دقیقاً می‌خواهیم از blockشدن طولانی جلوگیری کنیم.

با این حال گاهی لازم است عملیات asynchronous به‌صورت sequential اجرا شوند یا parallelism به حداکثر n عملیات محدود شود. مثلاً مرورگر می‌تواند downloadها را asynchronous و parallel اجرا کند اما نخواهد بیش از 10 download هم‌زمان داشته باشد. SemaphoreSlim این کار را انجام می‌دهد:

SemaphoreSlim _semaphore = new SemaphoreSlim (10);

async Task<byte[]> DownloadWithSemaphoreAsync (string uri)
{
  await _semaphore.WaitAsync();
  try { return await new WebClient().DownloadDataTaskAsync (uri); }
  finally { _semaphore.Release(); }
}

اگر initial count را به 1 کاهش دهیم، حداکثر parallelism نیز 1 می‌شود و در عمل یک asynchronous lock خواهیم داشت.

نوشتن Extension Method با نام EnterAsync

متن منبع با استفاده از کلاس Disposable نوشته‌شده در بخش Anonymous Disposal، یک extension method برای ساده‌کردن استفادهٔ async از SemaphoreSlim پیشنهاد می‌کند:

public static async Task<IDisposable> EnterAsync (this SemaphoreSlim ss)
{
  await ss.WaitAsync().ConfigureAwait (false);
  return Disposable.Create (() => ss.Release());
}

سپس متد download چنین می‌شود:

async Task<byte[]> DownloadWithSemaphoreAsync (string uri)
{
  using (await _semaphore.EnterAsync())
    return await new WebClient().DownloadDataTaskAsync (uri);
}

Parallel.ForEachAsync

از .NET 6 راه دیگر برای محدودکردن concurrency asynchronous، متد Parallel.ForEachAsync است. اگر uris آرایه‌ای از URIها باشد، مثال زیر با حداکثر 10 download موازی آن‌ها را دریافت می‌کند:

await Parallel.ForEachAsync (uris,
  new ParallelOptions { MaxDegreeOfParallelism = 10 },
  async (uri, cancelToken) =>
  {
    var download = await new HttpClient().GetByteArrayAsync (uri);
    Console.WriteLine ($"Downloaded {download.Length} bytes");
  });

سایر متدهای کلاس Parallel بیشتر برای parallel programming از نوع compute-bound هستند و در فصل ۲۲ بررسی می‌شوند.

Reader/Writer Lockها

بسیاری از typeها برای read هم‌زمان thread-safe هستند، اما برای update هم‌زمان یا read و update هم‌زمان نیستند؛ resourceهایی مانند file نیز می‌توانند همین ویژگی را داشته باشند. یک exclusive lock ساده همیشه می‌تواند همهٔ accessها را serialize کند، اما اگر reader زیاد و update کم باشد concurrency را بی‌جهت محدود می‌کند. ReaderWriterLockSlim برای همین سناریو طراحی شده است.

دو نوع اصلی lock وجود دارد:

• Write lock کاملاً exclusive است.
• Read lock با read lockهای دیگر سازگار است.

Thread دارای write lock همهٔ threadهای خواهان read یا write را block می‌کند و بالعکس. اگر write lock وجود نداشته باشد هر تعداد thread می‌توانند هم‌زمان read lock بگیرند.

متدهای اصلی ReaderWriterLockSlim:

public void EnterReadLock();
public void ExitReadLock();
public void EnterWriteLock();
public void ExitWriteLock();

برای همهٔ متدهای Enter نسخه‌های Try با timeout وجود دارد. در type قدیمی، متدهای مشابه با نام‌های AcquireXXX/ReleaseXXX وجود دارند و در timeout به‌جای false یک ApplicationException پرتاب می‌کنند.

برنامهٔ زیر سه thread reader و دو thread writer دارد؛ readerها دائماً list را enumerate می‌کنند و writerها هر 100 ms عدد تصادفی می‌افزایند:

class SlimDemo
{
  static ReaderWriterLockSlim _rw = new ReaderWriterLockSlim();
  static List<int> _items = new List<int>();
  static Random _rand = new Random();

  static void Main()
  {
    new Thread (Read).Start();
    new Thread (Read).Start();
    new Thread (Read).Start();
    new Thread (Write).Start ("A");
    new Thread (Write).Start ("B");
  }

  static void Read()
  {
    while (true)
    {
      _rw.EnterReadLock();
      foreach (int i in _items) Thread.Sleep (10);
      _rw.ExitReadLock();
    }
  }

  static void Write (object threadID)
  {
    while (true)
    {
      int newNumber = GetRandNum (100);
      _rw.EnterWriteLock();
      _items.Add (newNumber);
      _rw.ExitWriteLock();
      Console.WriteLine ("Thread " + threadID + " added " + newNumber);
      Thread.Sleep (100);
    }
  }

  static int GetRandNum (int max) { lock (_rand) return _rand.Next(max); }
}

Thread B added 61
Thread A added 83
Thread B added 55
Thread A added 33
...

ReaderWriterLockSlim readهای concurrent بیشتری نسبت به lock ساده اجازه می‌دهد. افزودن خط زیر در متد Write معمولاً «3 concurrent readers» را نشان می‌دهد:

Console.WriteLine (_rw.CurrentReadCount + " concurrent readers");

Propertyهای مانیتورینگ دیگر:

public bool IsReadLockHeld            { get; }
public bool IsUpgradeableReadLockHeld { get; }
public bool IsWriteLockHeld           { get; }

public int WaitingReadCount           { get; }
public int WaitingUpgradeCount        { get; }
public int WaitingWriteCount          { get; }

public int RecursiveReadCount         { get; }
public int RecursiveUpgradeCount      { get; }
public int RecursiveWriteCount        { get; }

Upgradeable Lock

گاهی لازم است read lock در یک عملیات atomic به write lock تبدیل شود. فرض کنید می‌خواهید فقط اگر item از قبل در list وجود ندارد آن را اضافه کنید. اگر read lock را آزاد و سپس write lock بگیرید، thread دیگری می‌تواند میان این دو مرحله list را تغییر دهد. ReaderWriterLockSlim نوع سومی به نام upgradeable lock دارد؛ مانند read lock است ولی می‌تواند atomically به write lock ارتقا یابد.

1. EnterUpgradeableReadLock را فراخوانی کنید.
2. عملیات read انجام دهید.
3. EnterWriteLock را فراخوانی کنید.
4. عملیات write را انجام دهید.
5. ExitWriteLock؛ سپس در صورت نیاز read بیشتر.
6. ExitUpgradeableReadLock.

از دید caller شبیه nested locking است، ولی در مرحلهٔ upgrade، Slim به‌طور atomic read lock را با write lock عوض می‌کند. هر تعداد read lock می‌توانند هم‌زمان باشند، اما در هر لحظه فقط یک upgradeable lock مجاز است؛ این محدودیت conversion deadlock را مهار می‌کند. متن منبع آن را با lockهای SQL Server مقایسه می‌کند:

while (true)
{
  int newNumber = GetRandNum (100);
  _rw.EnterUpgradeableReadLock();
  if (!_items.Contains (newNumber))
  {
    _rw.EnterWriteLock();
    _items.Add (newNumber);
    _rw.ExitWriteLock();
    Console.WriteLine ("Thread " + threadID + " added " + newNumber);
  }
  _rw.ExitUpgradeableReadLock();
  Thread.Sleep (100);
}

Lock Recursion

به‌طور پیش‌فرض nested/recursive locking در ReaderWriterLockSlim ممنوع است و کد زیر exception می‌دهد:

var rw = new ReaderWriterLockSlim();
rw.EnterReadLock();
rw.EnterReadLock();
rw.ExitReadLock();
rw.ExitReadLock();

اگر recursion عمداً موردنیاز باشد:

var rw = new ReaderWriterLockSlim (LockRecursionPolicy.SupportsRecursion);

Recursive locking complexity را بالا می‌برد، زیرا می‌توان بیش از یک نوع lock داشت:

rw.EnterWriteLock();
rw.EnterReadLock();
Console.WriteLine (rw.IsReadLockHeld);  // True
Console.WriteLine (rw.IsWriteLockHeld); // True
rw.ExitReadLock();
rw.ExitWriteLock();

قاعدهٔ پایه این است که پس از گرفتن lock، recursive lock بعدی فقط می‌تواند روی مقیاس زیر «کمتر» یا مساوی باشد، نه قوی‌تر:

Read Lock → Upgradeable Lock → Write Lock

اما ارتقای upgradeable lock به write lock همیشه قانونی است.

Signaling با Event Wait Handleها

ساده‌ترین constructهای signaling، event wait handleها هستند و ارتباطی با eventهای C# ندارند. سه گونهٔ اصلی‌اند: AutoResetEvent، ManualResetEvent/ManualResetEventSlim و CountdownEvent. دو مورد نخست از کلاس پایهٔ EventWaitHandle استفاده می‌کنند.

AutoResetEvent

AutoResetEvent را می‌توان مانند turnstile بلیت‌دار تصور کرد: هر بلیت دقیقاً یک نفر را عبور می‌دهد و پس از عبور، دروازه خودکار بسته یا reset می‌شود. thread با WaitOne منتظر می‌ماند و thread دیگری با Set یک blocked thread را آزاد می‌کند. اگر چند thread منتظر باشند queue تشکیل می‌شود؛ هر thread آزاد و دارای reference به همان object می‌تواند Set را صدا بزند.

دو روش ساخت:

var auto = new AutoResetEvent (false);
var auto2 = new EventWaitHandle (false, EventResetMode.AutoReset);

ارسال true به constructor معادل Set فوری است.

مثال پایه:

class BasicWaitHandle
{
  static EventWaitHandle _waitHandle = new AutoResetEvent (false);

  static void Main()
  {
    new Thread (Waiter).Start();
    Thread.Sleep (1000);
    _waitHandle.Set();
  }

  static void Waiter()
  {
    Console.WriteLine ("Waiting...");
    _waitHandle.WaitOne();
    Console.WriteLine ("Notified");
  }
}
// Output:
// Waiting... (pause) Notified.

اگر زمانی که هیچ threadی منتظر نیست Set فراخوانی شود، handle باز می‌ماند تا اولین thread بعدی WaitOne را فراخوانی کند. این رفتار race «سیگنال کمی زودتر رسید و برای همیشه گم شد» را کاهش می‌دهد. اما چند Set پیاپی زمانی که هیچ‌کس منتظر نیست جمع نمی‌شوند؛ فقط نفر بعدی عبور می‌کند و signalهای اضافه هدر می‌روند.

Dispose کردن Wait Handle

پس از پایان کار می‌توان Close را برای آزادکردن resource سیستم‌عامل فراخوانی کرد. همچنین finalizer الگوی disposal را تکمیل می‌کند و در صورت رهاشدن همهٔ referenceها GC در آینده handle را آزاد خواهد کرد. متن منبع این مورد را از معدود مواردی می‌داند که اتکا به finalizer به‌عنوان backup معمولاً قابل قبول است، چون burden سیستم‌عامل سبک است. با پایان process نیز wait handleها خودکار آزاد می‌شوند.

Reset یک AutoResetEvent باز را می‌بندد بدون آن‌که منتظر شود. WaitOne timeout اختیاری می‌پذیرد و اگر timeout علت پایان انتظار باشد false برمی‌گرداند. timeout برابر صفر فقط بازبودن handle را test می‌کند و اگر AutoResetEvent باز باشد آن را reset نیز می‌کند.

Two-way Signaling

اگر main thread بخواهد worker را سه بار پشت‌سرهم signal کند، چند Set سریع ممکن است signal دوم یا سوم را از دست بدهند چون worker برای پردازش هر signal زمان می‌خواهد. راه‌حل این است که main thread پیش از signal بعدی صبر کند تا worker اعلام آمادگی کند. برای این کار AutoResetEvent دوم استفاده می‌شود:

class TwoWaySignaling
{
  static EventWaitHandle _ready = new AutoResetEvent (false);
  static EventWaitHandle _go = new AutoResetEvent (false);
  static readonly object _locker = new object();
  static string _message;

  static void Main()
  {
    new Thread (Work).Start();

    _ready.WaitOne();
    lock (_locker) _message = "ooo";
    _go.Set();

    _ready.WaitOne();
    lock (_locker) _message = "ahhh";
    _go.Set();

    _ready.WaitOne();
    lock (_locker) _message = null;
    _go.Set();
  }

  static void Work()
  {
    while (true)
    {
      _ready.Set();
      _go.WaitOne();
      lock (_locker)
      {
        if (_message == null) return;
        Console.WriteLine (_message);
      }
    }
  }
}
// Output:
// ooo
// ahhh

در مثال، پیام null به worker اعلام می‌کند باید پایان یابد. برای threadهایی که طولانی یا نامحدود اجرا می‌شوند، داشتن exit strategy مهم است.

ManualResetEvent

ManualResetEvent مانند دروازهٔ ساده است: Set دروازه را باز می‌کند و هر تعداد thread منتظر WaitOne را عبور می‌دهد؛ Reset دروازه را می‌بندد. threadهایی که هنگام بسته‌بودن WaitOne می‌کنند block می‌شوند و دفعهٔ بعد که دروازه باز شود همه با هم آزاد می‌شوند.

var manual1 = new ManualResetEvent (false);
var manual2 = new EventWaitHandle (false, EventResetMode.ManualReset);

Performance ابزارهای Signaling

طبق متن منبع، Wait یا Signal کردن AutoResetEvent یا ManualResetEvent بدون blocking حدود یک microsecond زمان می‌برد. ManualResetEventSlim و CountdownEvent در waitهای کوتاه می‌توانند تا حدود 50 برابر سریع‌تر باشند، چون کمتر به OS وابسته‌اند و هوشمندانه spin می‌کنند. با این حال در بیشتر سناریوها سربار خود signaling classها bottleneck اصلی نیست.

CountdownEvent

CountdownEvent اجازه می‌دهد روی بیش از یک thread منتظر بمانید. هنگام ساخت تعداد countهایی که باید تکمیل شوند مشخص می‌شود:

var countdown = new CountdownEvent (3);

Signal count را کم می‌کند و Wait تا رسیدن count به صفر block می‌شود:

new Thread (SaySomething).Start ("I am thread 1");
new Thread (SaySomething).Start ("I am thread 2");
new Thread (SaySomething).Start ("I am thread 3");

countdown.Wait();
Console.WriteLine ("All threads have finished speaking!");

void SaySomething (object thing)
{
  Thread.Sleep (1000);
  Console.WriteLine (thing);
  countdown.Signal();
}

AddCount count را دوباره افزایش می‌دهد، اما اگر count قبلاً صفر شده باشد exception می‌دهد؛ TryAddCount در این حالت false برمی‌گرداند. برای unsignal کردن باید Reset را فراخوانی کرد که هم state را reset و هم count را به مقدار اولیه بازمی‌گرداند. مانند ManualResetEventSlim، این class نیز property WaitHandle دارد.

ساخت EventWaitHandle بین Processها

Constructorِ EventWaitHandle نام اختیاری می‌پذیرد و می‌تواند handle مشترک میان processها بسازد. اگر همان نام از قبل روی رایانه وجود داشته باشد reference به همان handle زیرین برگردانده می‌شود، وگرنه OS نمونهٔ جدیدی می‌سازد:

EventWaitHandle wh = new EventWaitHandle (false, EventResetMode.AutoReset,
                                      @"Global\MyCompany.MyApp.SomeName");

دو application که این کد را اجرا کنند می‌توانند یکدیگر را signal کنند. Named event wait handle فقط روی Windows در دسترس است.

Wait Handleها و Continuation

به‌جای block کردن thread با wait روی handle، می‌توان با ThreadPool.RegisterWaitForSingleObject continuation به آن متصل کرد. delegate زمانی اجرا می‌شود که handle signal شود:

var starter = new ManualResetEvent (false);

RegisteredWaitHandle reg = ThreadPool.RegisterWaitForSingleObject
 (starter, Go, "Some Data", -1, true);

Thread.Sleep (5000);
Console.WriteLine ("Signaling worker...");
starter.Set();
Console.ReadLine();
reg.Unregister (starter);

void Go (object data, bool timedOut)
{
  Console.WriteLine ("Started - " + data);
  // Perform task...
}

// Output:
// (5 second delay)
// Signaling worker...
// Started - Some Data

وقتی handle signal شود یا timeout برسد، delegate روی pooled thread اجرا می‌شود. سپس برای آزادکردن unmanaged handle مربوط به callback باید Unregister فراخوانی شود. علاوه بر wait handle و delegate، این متد یک object دلخواه برای انتقال به callback، timeout برحسب millisecond ــ مقدار -1 یعنی بدون timeout ــ و flagی برای one-off یا recurring بودن درخواست می‌پذیرد.

WaitAny، WaitAll و SignalAndWait

کلاس WaitHandle علاوه بر Set، WaitOne و Reset متدهای static برای synchronization پیچیده‌تر دارد. WaitAny، WaitAll و SignalAndWait روی چند handle کار می‌کنند و handleها می‌توانند از typeهای مختلف، حتی Mutex و Semaphore، باشند. ManualResetEventSlim و CountdownEvent نیز از طریق property WaitHandle قابل استفاده‌اند.

WaitAny منتظر یکی از آرایهٔ handleهاست. WaitAll به‌صورت atomic روی همهٔ آن‌ها منتظر می‌ماند. اگر دو AutoResetEvent داشته باشید، WaitAny هرگز هر دو را با هم latch نمی‌کند و WaitAll نیز فقط یکی را latch نمی‌کند.

SignalAndWait روی یک WaitHandle Set و سپس روی handle دیگر WaitOne می‌کند. بعد از signal اول، thread برای handle دوم به ابتدای queue می‌رود و شانس موفقیت بالا می‌رود، هرچند کل عملیات واقعاً atomic نیست. می‌توان آن را «تعویض یک signal با signal دیگر» تصور کرد و برای rendezvous دو thread استفاده نمود:

// First thread:
WaitHandle.SignalAndWait (wh1, wh2);

// Second thread:
WaitHandle.SignalAndWait (wh2, wh1);