< Summary - CoreWCF Coverage — PR #1766

Information
Class: CoreWCF.Runtime.OutWrapper<T>
Assembly: CoreWCF.Primitives
File(s): /home/runner/work/CoreWCF/CoreWCF/src/Common/src/CoreWCF/Runtime/TaskHelpers.cs
Line coverage
0%
Covered lines: 0
Uncovered lines: 5
Coverable lines: 5
Total lines: 531
Line coverage: 0%
Branch coverage
N/A
Covered branches: 0
Total branches: 0
Branch coverage: N/A
Method coverage

Feature is only available for sponsors

Upgrade to PRO version

Metrics

MethodBranch coverage Cyclomatic complexity NPath complexity Sequence coverage
.ctor()100%110%
op_Implicit(...)100%110%

File(s)

/home/runner/work/CoreWCF/CoreWCF/src/Common/src/CoreWCF/Runtime/TaskHelpers.cs

#LineLine coverage
 1// Licensed to the .NET Foundation under one or more agreements.
 2// The .NET Foundation licenses this file to you under the MIT license.
 3
 4using System;
 5using System.Diagnostics;
 6using System.Diagnostics.Contracts;
 7using System.Runtime.CompilerServices;
 8using System.Threading;
 9using System.Threading.Tasks;
 10using CoreWCF;
 11using CoreWCF.Dispatcher;
 12
 13namespace CoreWCF.Runtime
 14{
 15    internal static class TaskHelpers
 16    {
 17        //This replaces the Wait<TException>(this Task task) method as we want to await and not Wait()
 18        public static async Task AsyncWait<TException>(this Task task)
 19        {
 20            try
 21            {
 22                await task;
 23            }
 24            catch
 25            {
 26                throw Fx.Exception.AsError<TException>(task.Exception);
 27            }
 28        }
 29
 30        // Helper method when implementing an APM wrapper around a Task based async method which returns a result.
 31        // In the BeginMethod method, you would call use ToApm to wrap a call to MethodAsync:
 32        //     return MethodAsync(params).ToApm(callback, state);
 33        // In the EndMethod, you would use ToApmEnd<TResult> to ensure the correct exception handling
 34        // This will handle throwing exceptions in the correct place and ensure the IAsyncResult contains the provided
 35        // state object
 36        public static IAsyncResult ToApm<T>(this Task<T> task, AsyncCallback callback, object state)
 37            => ToApm<T>(new ValueTask<T>(task), callback, state);
 38
 39        /// <summary>
 40        /// Helper method to convert from Task async method to "APM" (IAsyncResult with Begin/End calls)
 41        /// </summary>
 42        public static IAsyncResult ToApm<T>(this ValueTask<T> valueTask, AsyncCallback callback, object state)
 43        {
 44            var result = new AsyncResult<T>(valueTask, callback, state);
 45            if (result.CompletedSynchronously)
 46            {
 47                result.ExecuteCallback();
 48            }
 49            else if (callback != null)
 50            {
 51                // We use OnCompleted rather than ContinueWith in order to avoid running synchronously
 52                // if the task has already completed by the time we get here.
 53                // This will allocate a delegate and some extra data to add it as a TaskContinuation
 54                valueTask.ConfigureAwait(false)
 55                    .GetAwaiter()
 56                    .OnCompleted(result.ExecuteCallback);
 57            }
 58
 59            return result;
 60        }
 61
 62        /// <summary>
 63        /// Helper method to convert from Task async method to "APM" (IAsyncResult with Begin/End calls)
 64        /// </summary>
 65        public static IAsyncResult ToApm(this Task task, AsyncCallback callback, object state)
 66        {
 67            var result = new AsyncResult(task, callback, state);
 68            if (result.CompletedSynchronously)
 69            {
 70                result.ExecuteCallback();
 71            }
 72            else if (callback != null)
 73            {
 74                // We use OnCompleted rather than ContinueWith in order to avoid running synchronously
 75                // if the task has already completed by the time we get here.
 76                // This will allocate a delegate and some extra data to add it as a TaskContinuation
 77                task.ConfigureAwait(false)
 78                    .GetAwaiter()
 79                    .OnCompleted(result.ExecuteCallback);
 80            }
 81
 82            return result;
 83        }
 84
 85        public static T ToApmEnd<T>(this IAsyncResult asyncResult)
 86        {
 87            if (asyncResult is AsyncResult<T> asyncResultInstance)
 88            {
 89                return asyncResultInstance.GetResult();
 90            }
 91            else
 92            {
 93                throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new ArgumentException(SRCommon.SFxInvalidCallb
 94            }
 95        }
 96
 97        public static void ToApmEnd(this IAsyncResult asyncResult)
 98        {
 99            if (asyncResult is AsyncResult asyncResultInstance)
 100            {
 101                asyncResultInstance.GetResult();
 102            }
 103            else
 104            {
 105                throw DiagnosticUtility.ExceptionUtility.ThrowHelperError(new ArgumentException(SRCommon.SFxInvalidCallb
 106            }
 107        }
 108
 109        private class AsyncResult : IAsyncResult
 110        {
 111            private readonly Task _task;
 112            private readonly AsyncCallback _asyncCallback;
 113
 114            public AsyncResult(Task task, AsyncCallback asyncCallback, object asyncState)
 115            {
 116                _task = task;
 117                _asyncCallback = asyncCallback;
 118                AsyncState = asyncState;
 119                CompletedSynchronously = task.IsCompleted;
 120            }
 121
 122            public void GetResult() => _task.GetAwaiter().GetResult();
 123
 124            public void ExecuteCallback() => _asyncCallback?.Invoke(this);
 125
 126            public object AsyncState { get; }
 127            WaitHandle IAsyncResult.AsyncWaitHandle => ((IAsyncResult)_task).AsyncWaitHandle;
 128
 129            public bool CompletedSynchronously { get; }
 130            public bool IsCompleted => _task.IsCompleted;
 131        }
 132
 133        internal class AsyncResult<T> : IAsyncResult
 134        {
 135            private readonly ValueTask<T> _task;
 136            private readonly AsyncCallback _asyncCallback;
 137
 138            public AsyncResult(ValueTask<T> task, AsyncCallback asyncCallback, object asyncState)
 139            {
 140                _task = task;
 141                _asyncCallback = asyncCallback;
 142                AsyncState = asyncState;
 143                CompletedSynchronously = task.IsCompleted;
 144            }
 145
 146            public T GetResult() => _task.GetAwaiter().GetResult();
 147
 148            public bool IsFaulted => _task.IsFaulted;
 149            public AggregateException Exception => _task.AsTask().Exception;
 150
 151            // Calls the async callback with this as parameter
 152            public void ExecuteCallback() => _asyncCallback?.Invoke(this);
 153            public object AsyncState { get; }
 154            WaitHandle IAsyncResult.AsyncWaitHandle => !CompletedSynchronously ? ((IAsyncResult)_task.AsTask()).AsyncWai
 155
 156            public bool CompletedSynchronously { get; }
 157            public bool IsCompleted => _task.IsCompleted;
 158        }
 159
 160        // Awaitable helper to await a maximum amount of time for a task to complete. If the task doesn't
 161        // complete in the specified amount of time, returns false. This does not modify the state of the
 162        // passed in class, but instead is a mechanism to allow interrupting awaiting a task if a timeout
 163        // period passes.
 164        // TODO: When we move away from netstandard, we can switch to the new Task.WaitAsync method
 165        public static async Task<bool> AwaitWithTimeout(this Task task, TimeSpan timeout)
 166        {
 167            if (task.IsCompleted)
 168            {
 169                return true;
 170            }
 171
 172            if (timeout == TimeSpan.MaxValue || timeout == Timeout.InfiniteTimeSpan)
 173            {
 174                await task;
 175                return true;
 176            }
 177
 178            using (CancellationTokenSource cts = new CancellationTokenSource())
 179            {
 180                Task completedTask = await Task.WhenAny(task, Task.Delay(timeout, cts.Token));
 181                if (completedTask == task)
 182                {
 183                    cts.Cancel();
 184                    return true;
 185                }
 186                else
 187                {
 188                    return (task.IsCompleted);
 189                }
 190            }
 191        }
 192
 193        // Task.GetAwaiter().GetResult() calls an internal variant of Wait() which doesn't wrap exceptions in
 194        // an AggregateException. It does spinwait so if it's expected that the Task isn't about to complete,
 195        // then use the NoSpin variant.
 196        public static void WaitForCompletion(this Task task)
 197        {
 198            Fx.Assert(task.IsCompleted || !IOThreadScheduler.IsRunningOnIOThread, "Waiting on an IO Thread might cause p
 199            // Waiting on an IO Thread can cause performance problems as we might block the IOThreadScheduler
 200            // dequeuing loop.
 201            task.GetAwaiter().GetResult();
 202        }
 203
 204        // If the task is about to complete, this method will be more expensive than the regular method as it
 205        // always causes a WaitHandle to be allocated. If it is expected that the task will take longer than
 206        // the time of a spin wait, then a WaitHandle will be allocated anyway and this method avoids the CPU
 207        // cost of the spin wait.
 208        public static void WaitForCompletionNoSpin(this Task task)
 209        {
 210            if (!task.IsCompleted)
 211            {
 212                Fx.Assert(!IOThreadScheduler.IsRunningOnIOThread, "Waiting on an IO Thread might cause problems");
 213                // Waiting on an IO Thread can cause performance problems as we might block the IOThreadScheduler
 214                // dequeuing loop.
 215                ((IAsyncResult)task).AsyncWaitHandle.WaitOne();
 216            }
 217
 218            // Call GetResult() to get any exceptions that were thrown
 219            task.GetAwaiter().GetResult();
 220        }
 221
 222        public static TResult WaitForCompletion<TResult>(this Task<TResult> task)
 223        {
 224            Fx.Assert(task.IsCompleted || !IOThreadScheduler.IsRunningOnIOThread, "Waiting on an IO Thread might cause p
 225            // Waiting on an IO Thread can cause performance problems as we might block the IOThreadScheduler
 226            // dequeuing loop.
 227            return task.GetAwaiter().GetResult();
 228        }
 229
 230        public static TResult WaitForCompletionNoSpin<TResult>(this Task<TResult> task)
 231        {
 232            if (!task.IsCompleted)
 233            {
 234                Fx.Assert(!IOThreadScheduler.IsRunningOnIOThread, "Waiting on an IO Thread might cause problems");
 235                // Waiting on an IO Thread can cause performance problems as we might block the IOThreadScheduler
 236                // dequeuing loop.
 237                ((IAsyncResult)task).AsyncWaitHandle.WaitOne();
 238            }
 239
 240            return task.GetAwaiter().GetResult();
 241        }
 242
 243        public static bool WaitForCompletionNoSpin(this Task task, TimeSpan timeout)
 244        {
 245            if (timeout >= TimeoutHelper.MaxWait)
 246            {
 247                task.WaitForCompletionNoSpin();
 248                return true;
 249            }
 250
 251            bool completed = true;
 252            if (!task.IsCompleted)
 253            {
 254                Fx.Assert(!IOThreadScheduler.IsRunningOnIOThread, "Waiting on an IO Thread might cause problems");
 255                // Waiting on an IO Thread can cause performance problems as we might block the IOThreadScheduler
 256                // dequeuing loop.
 257                completed = ((IAsyncResult)task).AsyncWaitHandle.WaitOne(timeout);
 258            }
 259
 260            if (completed)
 261            {
 262                // Throw any exceptions if there are any
 263                task.GetAwaiter().GetResult();
 264            }
 265
 266            return completed;
 267        }
 268
 269        // Used by WebSocketTransportDuplexSessionChannel on the sync code path.
 270        // TODO: Try and switch as many code paths as possible which use this to async
 271        public static void Wait(this Task task, TimeSpan timeout, Action<Exception, TimeSpan, string> exceptionConverter
 272        {
 273            bool timedOut = false;
 274
 275            try
 276            {
 277                timedOut = !task.WaitForCompletionNoSpin(timeout);
 278            }
 279            catch (Exception ex)
 280            {
 281                if (Fx.IsFatal(ex) || exceptionConverter == null)
 282                {
 283                    throw;
 284                }
 285
 286                exceptionConverter(ex, timeout, operationType);
 287            }
 288
 289            if (timedOut)
 290            {
 291                throw Fx.Exception.AsError(new TimeoutException(SRCommon.Format(SRCommon.TaskTimedOutError, timeout)));
 292            }
 293        }
 294
 295        public static Task CompletedTask()
 296        {
 297            return Task.FromResult(true);
 298        }
 299
 300        public static DefaultTaskSchedulerAwaiter EnsureDefaultTaskScheduler()
 301        {
 302            return DefaultTaskSchedulerAwaiter.Singleton;
 303        }
 304
 305        public static Action<object> OnAsyncCompletionCallback = OnAsyncCompletion;
 306
 307        // Method to act as callback for asynchronous code which uses AsyncCompletionResult as the return type when used
 308        // a Task based async method. These methods require a callback which is called in the case of the IO completing 
 309        // This pattern still requires an allocation, whereas the purpose of using the AsyncCompletionResult enum is to 
 310        // In the future, this pattern should be replaced with a reusable awaitable object, potentially with a global po
 311        private static void OnAsyncCompletion(object state)
 312        {
 313            var tcs = state as TaskCompletionSource<bool>;
 314            Fx.Assert(tcs != null, "Async state should be of type TaskCompletionSource<bool>");
 315            tcs.TrySetResult(true);
 316        }
 317
 318        public static IDisposable RunTaskContinuationsOnOurThreads()
 319        {
 320            if (SynchronizationContext.Current == ServiceModelSynchronizationContext.Instance)
 321            {
 322                return null; // No need to save and restore state as we're already using the correct sync context
 323            }
 324
 325            return new SyncContextScope();
 326        }
 327
 328        // Calls the given Action asynchronously on the ThreadPool.
 329        public static async Task CallActionAsync<TArg>(Action<TArg> action, TArg argument)
 330        {
 331            // Make sure any async tasks started from the action have their continuation
 332            // execute on the IOThreadScheduler, but make sure the action itself is running
 333            // on the thread pool.
 334            if (!Thread.CurrentThread.IsThreadPoolThread)
 335            {
 336                // Switch to a thread pool thread to run passed action
 337                SynchronizationContext.SetSynchronizationContext(null);
 338                await Task.Yield();
 339            }
 340
 341            // Now we're running on the ThreadPool, we reset the SynchronizationContext to
 342            // our sync context which posts to the IOThreadScheduler. We're not hopping threads
 343            // so any synchronous blocking will occur on the current thread pool thread.
 344            Fx.Assert(Thread.CurrentThread.IsThreadPoolThread, "We should be running on the thread pool");
 345            using (var scope = RunTaskContinuationsOnOurThreads())
 346            {
 347                action(argument);
 348            }
 349        }
 350
 351        public static Task CompletedOrCanceled(CancellationToken token)
 352        {
 353            if (token.IsCancellationRequested)
 354            {
 355                return Task.FromCanceled(token);
 356            }
 357
 358            return Task.CompletedTask;
 359        }
 360
 361        private class SyncContextScope : IDisposable
 362        {
 363            private readonly SynchronizationContext _prevContext;
 364
 365            public SyncContextScope()
 366            {
 367                _prevContext = SynchronizationContext.Current;
 368                SynchronizationContext.SetSynchronizationContext(ServiceModelSynchronizationContext.Instance);
 369            }
 370
 371            public void Dispose()
 372            {
 373                SynchronizationContext.SetSynchronizationContext(_prevContext);
 374            }
 375        }
 376
 377        public static Task<TResult> CancellableAsyncWait<TResult>(this Task<TResult> task, CancellationToken token)
 378        {
 379            if (!token.CanBeCanceled)
 380            {
 381                return task;
 382            }
 383
 384            object[] state = new object[2];
 385            var tcs = new TaskCompletionSource<TResult>(state);
 386            state[0] = tcs;
 387            state[1] = default(CancellationTokenRegistration);
 388            CancellationTokenRegistration registration = token.Register(OnCancellation<TResult>, state);
 389            state[1] = registration;
 390            if (token.IsCancellationRequested)
 391            {
 392                registration.Dispose();
 393                tcs.TrySetCanceled();
 394            }
 395            else
 396            {
 397                task.ContinueWith((antecedent, obj) =>
 398                {
 399                    object[] stateArr = (object[])obj;
 400                    var tcsObj = (TaskCompletionSource<TResult>)stateArr[0];
 401                    var tokenRegistration = (CancellationTokenRegistration)stateArr[1];
 402                    tokenRegistration.Dispose();
 403                    if (antecedent.IsFaulted)
 404                    {
 405                        tcsObj.TrySetException(antecedent.Exception.InnerException);
 406                    }
 407                    else if (antecedent.IsCanceled)
 408                    {
 409                        tcsObj.TrySetCanceled();
 410                    }
 411                    else
 412                    {
 413                        tcsObj.TrySetResult(antecedent.Result);
 414                    }
 415                }, state, CancellationToken.None, TaskContinuationOptions.HideScheduler, TaskScheduler.Default);
 416            }
 417
 418            return tcs.Task;
 419        }
 420
 421        private static void OnCancellation<TResult>(object state)
 422        {
 423            object[] stateArr = (object[])state;
 424            var tcsObj = (TaskCompletionSource<TResult>)stateArr[0];
 425            var tokenRegistration = (CancellationTokenRegistration)stateArr[1];
 426            tcsObj.TrySetCanceled();
 427            tokenRegistration.Dispose();
 428        }
 429
 430        internal static SynchronizationContextAwaiter GetAwaiter(this SynchronizationContext syncContext)
 431        {
 432            return new SynchronizationContextAwaiter(syncContext);
 433        }
 434    }
 435
 436    // This awaiter causes an awaiting async method to continue on the same thread if using the
 437    // default task scheduler, otherwise it posts the continuation to the ThreadPool. While this
 438    // does a similar function to Task.ConfigureAwait, this code doesn't require a Task to function.
 439    // With Task.ConfigureAwait, you would need to call it on the first task on each potential code
 440    // path in a method. This could mean calling ConfigureAwait multiple times in a single method.
 441    // This awaiter can be awaited on at the beginning of a method a single time and isn't dependant
 442    // on running other awaitable code.
 443    internal struct DefaultTaskSchedulerAwaiter : INotifyCompletion
 444    {
 445        public static DefaultTaskSchedulerAwaiter Singleton = new DefaultTaskSchedulerAwaiter();
 446
 447        // If the current TaskScheduler is the default, if we aren't currently running inside a task and
 448        // the default SynchronizationContext isn't current, when a Task starts, it will change the TaskScheduler
 449        // to one based off the current SynchronizationContext. Also, any async api's that WCF consumes will
 450        // post back to the same SynchronizationContext as they were started in which could cause WCF to deadlock
 451        // on our Sync code path.
 452        public bool IsCompleted
 453        {
 454            get
 455            {
 456                return (TaskScheduler.Current == TaskScheduler.Default) &&
 457                    (SynchronizationContext.Current == null ||
 458                    (SynchronizationContext.Current.GetType() == typeof(SynchronizationContext)));
 459            }
 460        }
 461
 462        // Only called when IsCompleted returns false, otherwise the caller will call the continuation
 463        // directly causing it to stay on the same thread.
 464        public void OnCompleted(Action continuation)
 465        {
 466            Task.Run(continuation);
 467        }
 468
 469        // Awaiter is only used to control where subsequent awaitable's run so GetResult needs no
 470        // implementation. Normally any exceptions would be thrown here, but we have nothing to throw
 471        // as we don't run anything, only control where other code runs.
 472        public void GetResult() { }
 473
 474        public DefaultTaskSchedulerAwaiter GetAwaiter()
 475        {
 476            return this;
 477        }
 478    }
 479
 480    internal struct SynchronizationContextAwaiter : INotifyCompletion
 481    {
 482        private readonly SynchronizationContext _syncContext;
 483
 484        public SynchronizationContextAwaiter(SynchronizationContext syncContext)
 485        {
 486            _syncContext = syncContext;
 487        }
 488
 489        public bool IsCompleted => _syncContext == SynchronizationContext.Current;
 490
 491        public void OnCompleted(Action continuation)
 492        {
 493            // _syncContext will be null if it's the default sync context
 494            // This method is being called because IsCompleted returned false
 495            // This means we need to run the continuation on a regular thread pool
 496            // thread.
 497            if (_syncContext == null)
 498            {
 499                ThreadPool.UnsafeQueueUserWorkItem(PostCallback, continuation);
 500                return;
 501            }
 502
 503            _syncContext.Post(PostCallback, continuation);
 504        }
 505
 506
 507        public void GetResult() { }
 508
 509        internal static void PostCallback(object state)
 510        {
 511            ((Action)state)();
 512        }
 513    }
 514
 515    // Async methods can't take an out (or ref) argument. This wrapper allows passing in place of an out argument
 516    // and can be used to return a value via a method argument.
 517    internal class OutWrapper<T>
 518    {
 0519        public OutWrapper()
 520        {
 0521            Value = default;
 0522        }
 523
 0524        public T Value { get; set; }
 525
 526        public static implicit operator T(OutWrapper<T> wrapper)
 527        {
 0528            return wrapper.Value;
 529        }
 530    }
 531}