Async Coordination Primitives

ในบทความ Building Async Coordination Primitives, Part 1: AsyncManualResetEvent พูดถึงการใช้ TAP ทำ ManualResetEvent ซึ่งสามารถ Wait ได้แบบ asynchronous! ซึ่งหลายๆคนถ้าเริ่มใช้ async/await แล้วก็จะพบปัญหาเดียวกันว่า ทำไมมันไม่มี WaitAsync() ใน ManualResetEvent/AutoResetEvent ฟะ ... (มีแต่ใน SemaphoreSlim ซึ่ง เอิ่ม... Event กับ Semaphore มันก็แทนกันไม่ได้ 100% น่ะนะ)

โดยการใช้ TaskCompletionSource<bool> สำหรับ wait ก็สามารถ ทำ WaitAsync() กับ Set() ได้ง่ายๆ (แต่เป็น WaitAsync ที่ใส่ timeout ไม่ได้ :s )

TaskCompletionSource<bool> src = new TaskCompletionSource<bool>();  
 Task WaitAsync() { return src.Task; }
 void Set() { src.TrySetResult(true); }

ส่วนของ Reset() มีโจทย์ที่ต้องคิดเล็กน้อย คือเนื่องจากวิธีนี้ TaskCompletionSource ที่ set ไปแล้ว มัน reset อีกไม่ได้ เราต้อง new ขึ้นมาใหม่ ดังนั้นถ้ามีการ access Set() กับ WaitAsync() และ Reset() พร้อมๆกัน ต้องมั่นใจว่าตัวที่ new นั้นจะไม่เป็น"ลูกกำพร้า" 

ตัวอย่างเช่น มี 3 threads: t1, t2, t3 โดยที่ t1 กำลังเรียก WaitAsync() ขณะที่ t2 กับ t3 กำลังเรียก Reset() ... โดยสมมติว่าลำดับการ execute ออกมาเป็น t1 t2 t3 ตามลำดับ ต้องแน่ใจว่า instance ที่ reset โดย t2 จะต้องไม่ถูก replace โดย t3 ถ้ามันยังไม่ถูก set (มิฉะนั้นแล้ว t1 ก็จะรอตลอดไป เพราะ Set() จะมีผลกับ instance ใหม่ที่สร้างโดย t3)

ดังนั้น code ส่วนของ Reset จริงเป็น spin exchange โดยที่ check completion ของ Task ปัจจุบันด้วย

void Reset(){  
 TaskCompletionSource&lt;bool&gt; current;
 while((current = source).Task.IsCompleted
 && Interlocked.CompareExchange(ref source, new TaskCompletionSource<bool>(), current) != current){}
 }

แต่มีอีกวิธีนึง ซึ่งสามารถทำให้ WaitHandle สามารถทำ WaitOneAsync ได้เลย เป็นวิธีที่กล่าวไว้ในเอกสาร TAP หน้า 33 โดยใช้ ThreadPool.RegisterWaitForSingleObject() ซึ่งวิธีนี้ทำให้เขียน WaitAsync() ได้ทั้งแบบมี timeout และไม่มี timeout!

   Task WaitAsyncWaitHandle wait, int milliseconds = -1){
            var result = new TaskCompletionSource();
            var registration = ThreadPool.RegisterWaitForSingleObject(wait,
                                                                      (state,timedout) => result.TrySetResult(!timedout),
                                                                      null, milliseconds,
                                                                      executeOnlyOnce: true);
            result.Task.ContinueWith(_ => registration.Unregister(null));
            return result.Task;
    }

ส่วนถ้าเป็น ManualResetEventSlim เนื่องจากมันไม่ได้ inherit จาก WaitHandle ตรงๆ และการทำงานของมันผ่าน spin check ก่อนที่จะใช้ kernel Wait object ดังนั้น WaitAsync() ที่จะใช้กับ ManualResetEventSlim ต้องดัดแปลงอีกนิดหน่อย โดยการทำ spin check เองเลย :D

   Task WaitAsync(ManualResetEventSlim wait, int milliseconds){
            var counter = wait.SpinCount;
            var watch = Stopwatch.StartNew();
            while(!wait.IsSet && counter-- > 0 && watch.ElapsedMilliseconds &lt;= milliseconds){}
            if (wait.IsSet)
                return Task.FromResult(true);
            var elasped = (int) watch.ElapsedMilliseconds;
            if (elasped > milliseconds)
                return Task.FromResult(false);
            else
                return WaitAsync(wait.WaitHandle, milliseconds - elasped); // กลับไปใช้ Kernel wait object
        }

สรุปว่าเทคนิคอันหลัง work สุด ^^