mcs/class/referencesource/System.Web/Util/SynchronizationHelper.cs

   1 //------------------------------------------------------------------------------
   2 // <copyright file="SynchronizationHelper.cs" company="Microsoft">
   3 //     Copyright (c) Microsoft Corporation.  All rights reserved.
   4 // </copyright>
   5 //------------------------------------------------------------------------------
   6
   7 namespace System.Web.Util {
   8     using System;
   9     using System.Runtime.ExceptionServices;
  10     using System.Threading;
  11     using System.Threading.Tasks;
  12
  13     // This class is used by the AspNetSynchronizationContext to assist with scheduling tasks in a non-blocking fashion.
  14     // Asynchronous work will be queued and will execute sequentially, never consuming more than a single thread at a time.
  15     // Synchronous work will block and will execute on the current thread.
  16
  17     internal sealed class SynchronizationHelper {
  18
  19         private Task _completionTask; // the Task that will run when all in-flight operations have completed
  20         private Thread _currentThread; // the Thread that's running the current Task; all threads must see the same value for this field
  21         private Task _lastScheduledTask = CreateInitialTask(); // the last Task that was queued to this helper, used to hook future Tasks (not volatile since always accessed under lock)
  22         private Task _lastScheduledTaskAsync = CreateInitialTask(); // the last async Task that was queued to this helper
  23         private readonly object _lockObj = new object(); // synchronizes access to _lastScheduledTask
  24         private int _operationsInFlight; // operation counter
  25         private readonly ISyncContext _syncContext; // a context that wraps an operation with pre- and post-execution phases
  26         private readonly Action<bool> _appVerifierCallback; // for making sure that developers don't try calling us after the request has completed
  27
  28         public SynchronizationHelper(ISyncContext syncContext) {
  29             _syncContext = syncContext;
  30             _appVerifierCallback = AppVerifier.GetSyncContextCheckDelegate(syncContext);
  31         }
  32
  33         // If an operation results in an exception, this property will provide access to it.
  34         public ExceptionDispatchInfo Error { get; set; }
  35
  36         // Helper to access the _currentThread field in a thread-safe fashion.
  37         // It is not enough to mark the _currentThread field volatile, since that only guarantees
  38         // read / write ordering and doesn't ensure that each thread sees the same value.
  39         private Thread CurrentThread {
  40             get { return Interlocked.CompareExchange(ref _currentThread, null, null); }
  41             set { Interlocked.Exchange(ref _currentThread, value); }
  42         }
  43
  44         // Returns the number of pending operations
  45         public int PendingCount { get { return ChangeOperationCount(0); } }
  46
  47         public int ChangeOperationCount(int addend) {
  48             int newOperationCount = Interlocked.Add(ref _operationsInFlight, addend);
  49             if (newOperationCount == 0) {
  50                 // if an asynchronous completion operation is queued, run it
  51                 Task completionTask = Interlocked.Exchange(ref _completionTask, null);
  52                 if (completionTask != null) {
  53                     completionTask.Start();
  54                 }
  55             }
  56
  57             return newOperationCount;
  58         }
  59
  60         private void CheckForRequestStateIfRequired(bool checkForReEntry) {
  61             if (_appVerifierCallback != null) {
  62                 _appVerifierCallback(checkForReEntry);
  63             }
  64         }
  65
  66         // Creates the initial hook that future operations can ride off of
  67         private static Task CreateInitialTask() {
  68             return Task.FromResult<object>(null);
  69         }
  70
  71         // Takes control of this SynchronizationHelper instance synchronously. Asynchronous operations
  72         // will be queued but will not be dispatched until control is released (by disposing of the
  73         // returned object). This operation might block if a different thread is currently in
  74         // control of the context.
  75         public IDisposable EnterSynchronousControl() {
  76             if (CurrentThread == Thread.CurrentThread) {
  77                 // If the current thread already has control of this context, there's nothing extra to do.
  78                 return DisposableAction.Empty;
  79             }
  80
  81             // used to mark the end of the synchronous task
  82             TaskCompletionSource<object> tcs = new TaskCompletionSource<object>();
  83             Task lastTask;
  84             lock (_lockObj) {
  85                 lastTask = _lastScheduledTask;
  86                 _lastScheduledTask = tcs.Task; // future work can be scheduled off this Task
  87             }
  88
  89             // The original task may end up Faulted, which would make its Wait() method throw an exception.
  90             // To avoid this, we instead wait on a continuation which is always guaranteed to complete successfully.
  91             if (!lastTask.IsCompleted) { lastTask.ContinueWith(_ => { }, TaskContinuationOptions.ExecuteSynchronously).Wait(); }
  92             CurrentThread = Thread.CurrentThread;
  93
  94             // synchronous control is released by marking the Task as complete
  95             return new DisposableAction(() => {
  96                 CurrentThread = null;
  97                 tcs.TrySetResult(null);
  98             });
  99         }
 100
 101         public void QueueAsynchronous(Action action) {
 102             CheckForRequestStateIfRequired(checkForReEntry: true);
 103             ChangeOperationCount(+1);
 104
 105             // This method only schedules work; it doesn't itself do any work. The lock is held for a very
 106             // short period of time.
 107             lock (_lockObj) {
 108                 Task newTask = _lastScheduledTask.ContinueWith(_ => SafeWrapCallback(action), TaskScheduler.Default);
 109                 _lastScheduledTask = newTask; // the newly-created task is now the last one
 110             }
 111         }
 112
 113         // QueueAsynchronousAsync and SafeWrapCallbackAsync guarantee:
 114         // 1. For funcs posted here, it's would first come, first complete.
 115         // 2. There is no overlapping execution.
 116         public void QueueAsynchronousAsync(Func<object, Task> func, object state) {
 117             CheckForRequestStateIfRequired(checkForReEntry: true);
 118             ChangeOperationCount(+1);
 119
 120             // This method only schedules work; it doesn't itself do any work. The lock is held for a very
 121             // short period of time.
 122             lock (_lockObj) {
 123                 // 1. Note that we are chaining newTask with _lastScheduledTaskAsync, not _lastScheduledTask.
 124                 // Chaining newTask with _lastScheduledTask would cause deadlock.
 125                 // 2. Unwrap() is necessary to be called here. When chaining multiple tasks using the ContinueWith
 126                 // method, your return type will be Task<T> whereas T is the return type of the delegate/method
 127                 // passed to ContinueWith. As the return type of an async delegate is a Task, you will end up with
 128                 // a Task<Task> and end up waiting for the async delegate to return you the Task which is done after
 129                 // the first await.
 130                 Task newTask = _lastScheduledTaskAsync.ContinueWith(
 131                     async _ => { await SafeWrapCallbackAsync(func, state); }).Unwrap();
 132                 _lastScheduledTaskAsync = newTask; // the newly-created task is now the last one
 133             }
 134         }
 135
 136         public void QueueSynchronous(Action action) {
 137             CheckForRequestStateIfRequired(checkForReEntry: false);
 138             if (CurrentThread == Thread.CurrentThread) {
 139                 // current thread already owns the context, so just execute inline to prevent deadlocks
 140                 action();
 141                 return;
 142             }
 143
 144             ChangeOperationCount(+1);
 145             using (EnterSynchronousControl()) {
 146                 SafeWrapCallback(action);
 147             }
 148         }
 149
 150         private void SafeWrapCallback(Action action) {
 151             // This method will try to catch exceptions so that they don't bubble up to our
 152             // callers. However, ThreadAbortExceptions will continue to bubble up.
 153             try {
 154                 CurrentThread = Thread.CurrentThread;
 155                 ISyncContextLock syncContextLock = null;
 156                 try {
 157                     syncContextLock = (_syncContext != null) ? _syncContext.Enter() : null;
 158                     try {
 159                         action();
 160                     }
 161                     catch (Exception ex) {
 162                         Error = ExceptionDispatchInfo.Capture(ex);
 163                     }
 164                 }
 165                 finally {
 166                     if (syncContextLock != null) {
 167                         syncContextLock.Leave();
 168                     }
 169                 }
 170             }
 171             finally {
 172                 CurrentThread = null;
 173                 ChangeOperationCount(-1);
 174             }
 175         }
 176
 177         // This method does not run the func by itself. It simply queues the func into the existing
 178         // syncContext queue.
 179         private async Task SafeWrapCallbackAsync(Func<object, Task> func, object state) {
 180             try {
 181                 TaskCompletionSource<Task> tcs = new TaskCompletionSource<Task>();
 182                 QueueAsynchronous(() => {
 183                     var t = func(state);
 184                     t.ContinueWith((_) => {
 185                         if (t.IsFaulted) {
 186                             tcs.TrySetException(t.Exception.InnerExceptions);
 187                         }
 188                         else if (t.IsCanceled) {
 189                             tcs.TrySetCanceled();
 190                         }
 191                         else {
 192                             tcs.TrySetResult(t);
 193                         }
 194                     }, TaskContinuationOptions.ExecuteSynchronously);
 195                 });
 196                 await tcs.Task;
 197             }
 198             catch (Exception ex) {
 199                 Error = ExceptionDispatchInfo.Capture(ex);
 200             }
 201             finally {
 202                 ChangeOperationCount(-1);
 203             }
 204         }
 205
 206         // Sets the continuation that will asynchronously execute when the pending operation counter
 207         // hits zero. Returns true if asynchronous execution is expected, false if the operation
 208         // counter is already at zero and the caller should run the continuation inline.
 209         public bool TrySetCompletionContinuation(Action continuation) {
 210             int newOperationCount = ChangeOperationCount(+1); // prevent the operation counter from hitting zero while we're setting the field
 211             bool scheduledAsynchronously = (newOperationCount > 1);
 212             if (scheduledAsynchronously) {
 213                 Interlocked.Exchange(ref _completionTask, new Task(continuation));
 214             }
 215
 216             ChangeOperationCount(-1);
 217             return scheduledAsynchronously;
 218         }
 219
 220     }
 221 }