c# - 异步/等待，自定义等待者和垃圾收集器

我正在处理一种情况，即托管对象在async方法的中间过早完成。

这是一个业余爱好家庭自动化项目(Windows 8.1，.NET 4.5.1)，在该项目中，我向非托管的第三方DLL提供了C#回调。回调在某个传感器事件时被调用。

为了处理事件，我使用了async/await和一个简单的自定义等待器(而不是TaskCompletionSource)。我这样做是部分地减少了不必要的分配，但是出于好奇，这主要是出于学习的目的。

下面是我所拥有的非常剥离的版本，使用Win32计时器队列计时器模拟非托管事件源。让我们从输出开始:

Press Enter to exit...
Awaiter()
tick: 0
tick: 1
~Awaiter()
tick: 2
tick: 3
tick: 4

Note how my awaiter gets finalized after the second tick. This is unexpected.

The code (a console app):

using System;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApplication
{
    class Program
    {
        static async Task TestAsync()
        {
            var awaiter = new Awaiter();
            //var hold = GCHandle.Alloc(awaiter);

            WaitOrTimerCallbackProc callback = (a, b) =>
                awaiter.Continue();

            IntPtr timerHandle;
            if (!CreateTimerQueueTimer(out timerHandle,
                    IntPtr.Zero,
                    callback,
                    IntPtr.Zero, 500, 500, 0))
                throw new System.ComponentModel.Win32Exception(
                    Marshal.GetLastWin32Error());

            var i = 0;
            while (true)
            {
                await awaiter;
                Console.WriteLine("tick: " + i++);
            }
        }

        static void Main(string[] args)
        {
            Console.WriteLine("Press Enter to exit...");
            var task = TestAsync();
            Thread.Sleep(1000);
            GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced);
            Console.ReadLine();
        }

        // custom awaiter
        public class Awaiter :
            System.Runtime.CompilerServices.INotifyCompletion
        {
            Action _continuation;

            public Awaiter()
            {
                Console.WriteLine("Awaiter()");
            }

            ~Awaiter()
            {
                Console.WriteLine("~Awaiter()");
            }

            // resume after await, called upon external event
            public void Continue()
            {
                var continuation = Interlocked.Exchange(ref _continuation, null);
                if (continuation != null)
                    continuation();
            }

            // custom Awaiter methods
            public Awaiter GetAwaiter()
            {
                return this;
            }

            public bool IsCompleted
            {
                get { return false; }
            }

            public void GetResult()
            {
            }

            // INotifyCompletion
            public void OnCompleted(Action continuation)
            {
                Volatile.Write(ref _continuation, continuation);
            }
        }

        // p/invoke
        delegate void WaitOrTimerCallbackProc(IntPtr lpParameter, bool TimerOrWaitFired);

        [DllImport("kernel32.dll")]
        static extern bool CreateTimerQueueTimer(out IntPtr phNewTimer,
           IntPtr TimerQueue, WaitOrTimerCallbackProc Callback, IntPtr Parameter,
           uint DueTime, uint Period, uint Flags);
    }
}

我设法通过以下行来抑制awaiter的收集:

var hold = GCHandle.Alloc(awaiter);

但是，我不完全理解为什么我必须创建像这样的强大引用。 awaiter在无限循环内引用。 AFAICT，直到TestAsync返回的任务完成(取消/故障)后，它才不会超出范围。并且任务本身在Main中永远被引用。

最终，我将TestAsync简化为:

static async Task TestAsync()
{
    var awaiter = new Awaiter();
    //var hold = GCHandle.Alloc(awaiter);

    var i = 0;
    while (true)
    {
        await awaiter;
        Console.WriteLine("tick: " + i++);
    }
}

收集仍在进行。我怀疑正在收集整个编译器生成的状态机对象。 有人可以解释为什么会这样吗？

现在，通过以下较小的修改，不再对awaiter进行垃圾收集:

static async Task TestAsync()
{
    var awaiter = new Awaiter();
    //var hold = GCHandle.Alloc(awaiter);

    var i = 0;
    while (true)
    {
        //await awaiter;
        await Task.Delay(500);
        Console.WriteLine("tick: " + i++);
    }
}

更新了，this fiddle显示了如何在没有任何p/调用代码的情况下对awaiter对象进行垃圾回收。我认为，原因可能是在生成的状态机对象的初始状态之外没有对awaiter 的外部引用。我需要研究编译器生成的代码。

更新了，这是编译器生成的代码(对于this fiddle，VS2012)。显然，Task返回的stateMachine.t__builder.Task不会保留对状态机本身(stateMachine)的引用(或更确切地说，是对状态机本身的引用)。我想念什么吗？

    private static Task TestAsync()
    {
      Program.TestAsyncd__0 stateMachine;
      stateMachine.t__builder = AsyncTaskMethodBuilder.Create();
      stateMachine.1__state = -1;
      stateMachine.t__builder.Start<Program.TestAsyncd__0>(ref stateMachine);
      return stateMachine.t__builder.Task;
    }

    [CompilerGenerated]
    [StructLayout(LayoutKind.Auto)]
    private struct TestAsyncd__0 : IAsyncStateMachine
    {
      public int 1__state;
      public AsyncTaskMethodBuilder t__builder;
      public Program.Awaiter awaiter5__1;
      public int i5__2;
      private object u__awaiter3;
      private object t__stack;

      void IAsyncStateMachine.MoveNext()
      {
        try
        {
          bool flag = true;
          Program.Awaiter awaiter;
          switch (this.1__state)
          {
            case -3:
              goto label_7;
            case 0:
              awaiter = (Program.Awaiter) this.u__awaiter3;
              this.u__awaiter3 = (object) null;
              this.1__state = -1;
              break;
            default:
              this.awaiter5__1 = new Program.Awaiter();
              this.i5__2 = 0;
              goto label_5;
          }
label_4:
          awaiter.GetResult();
          Console.WriteLine("tick: " + (object) this.i5__2++);
label_5:
          awaiter = this.awaiter5__1.GetAwaiter();
          if (!awaiter.IsCompleted)
          {
            this.1__state = 0;
            this.u__awaiter3 = (object) awaiter;
            this.t__builder.AwaitOnCompleted<Program.Awaiter, Program.TestAsyncd__0>(ref awaiter, ref this);
            flag = false;
            return;
          }
          else
            goto label_4;
        }
        catch (Exception ex)
        {
          this.1__state = -2;
          this.t__builder.SetException(ex);
          return;
        }
label_7:
        this.1__state = -2;
        this.t__builder.SetResult();
      }

      [DebuggerHidden]
      void IAsyncStateMachine.SetStateMachine(IAsyncStateMachine param0)
      {
        this.t__builder.SetStateMachine(param0);
      }
    }

最佳答案

我删除了所有p/invoke内容，并重新创建了编译器生成的状态机逻辑的简化版本。它表现出相同的行为:第一次调用状态机的awaiter方法后，就会对MoveNext进行垃圾回收。

微软最近在为.NET reference sources提供Web UI方面做得非常出色，这非常有帮助。在研究了 AsyncTaskMethodBuilder 以及最重要的 AsyncMethodBuilderCore.GetCompletionAction 的实现之后，我现在相信我所看到的GC行为完全可以理解。我将在下面解释。

代码:

using System;
using System.Threading;
using System.Threading.Tasks;
using System.Runtime.InteropServices;
using System.Runtime.CompilerServices;

namespace ConsoleApplication
{
    public class Program
    {
        // Original version with async/await

        /*
        static async Task TestAsync()
        {
            Console.WriteLine("Enter TestAsync");
            var awaiter = new Awaiter();
            //var hold = GCHandle.Alloc(awaiter);

            var i = 0;
            while (true)
            {
                await awaiter;
                Console.WriteLine("tick: " + i++);
            }
            Console.WriteLine("Exit TestAsync");
        }
        */

        // Manually coded state machine version

        struct StateMachine: IAsyncStateMachine
        {
            public int _state;
            public Awaiter _awaiter;
            public AsyncTaskMethodBuilder _builder;

            public void MoveNext()
            {
                Console.WriteLine("StateMachine.MoveNext, state: " + this._state);
                switch (this._state)
                {
                    case -1:
                        {
                            this._awaiter = new Awaiter();
                            goto case 0;
                        };
                    case 0:
                        {
                            this._state = 0;
                            var awaiter = this._awaiter;
                            this._builder.AwaitOnCompleted(ref awaiter, ref this);
                            return;
                        };

                    default:
                        throw new InvalidOperationException();
                }
            }

            public void SetStateMachine(IAsyncStateMachine stateMachine)
            {
                Console.WriteLine("StateMachine.SetStateMachine, state: " + this._state);
                this._builder.SetStateMachine(stateMachine);
                // s_strongRef = stateMachine;
            }

            static object s_strongRef = null;
        }

        static Task TestAsync()
        {
            StateMachine stateMachine = new StateMachine();
            stateMachine._state = -1;

            stateMachine._builder = AsyncTaskMethodBuilder.Create();
            stateMachine._builder.Start(ref stateMachine);

            return stateMachine._builder.Task;
        }

        public static void Main(string[] args)
        {
            var task = TestAsync();
            Thread.Sleep(1000);
            GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced);
            Console.WriteLine("Press Enter to exit...");
            Console.ReadLine();
        }

        // custom awaiter
        public class Awaiter :
            System.Runtime.CompilerServices.INotifyCompletion
        {
            Action _continuation;

            public Awaiter()
            {
                Console.WriteLine("Awaiter()");
            }

            ~Awaiter()
            {
                Console.WriteLine("~Awaiter()");
            }

            // resume after await, called upon external event
            public void Continue()
            {
                var continuation = Interlocked.Exchange(ref _continuation, null);
                if (continuation != null)
                    continuation();
            }

            // custom Awaiter methods
            public Awaiter GetAwaiter()
            {
                return this;
            }

            public bool IsCompleted
            {
                get { return false; }
            }

            public void GetResult()
            {
            }

            // INotifyCompletion
            public void OnCompleted(Action continuation)
            {
                Console.WriteLine("Awaiter.OnCompleted");
                Volatile.Write(ref _continuation, continuation);
            }
        }
    }
}

编译器生成的状态机是可变结构，由ref传递。显然，这是为了避免额外分配而进行的优化。

其核心部分发生在AsyncMethodBuilderCore.GetCompletionAction内部，将当前状态机struct装箱，并通过传递给INotifyCompletion.OnCompleted的继续回调保留对装箱副本的引用。

这是对状态机的唯一引用，它有机会经受住GC并在await之后存活。 Task返回的TestAsync对象不会对进行引用，而会保留对它的引用，只有await延续回调可以。我相信这样做是有意的，以保持有效的GC行为。

注意注释行:

// s_strongRef = stateMachine;

如果我取消注释，则状态机的盒装副本不会被GC处理，并且awaiter作为其一部分仍然有效。当然，这不是解决方案，但可以说明问题。

因此，我得出以下结论。当异步操作处于“进行中”状态且当前未执行任何状态机状态(MoveNext)时，继续回调的“守护者”的职责是来对回调本身进行严格控制，确保装箱的状态机副本不会被垃圾收集。

例如，对于 YieldAwaitable (由Task.Yield返回)，由于ThreadPool调用，对继续回调的外部引用由ThreadPool.QueueUserWorkItem任务调度程序保留。如果使用Task.GetAwaiter，则任务对象为indirectly referenced。

就我而言，继续回调的“守护者”就是Awaiter本身。

因此，只要没有外部引用到CLR知道的连续回调(在状态机对象之外)，定制等待者就应采取措施使回调对象保持 Activity 状态。反过来，这将使整个状态机保持 Activity 状态。在这种情况下，必须执行以下步骤:

根据GCHandle.Alloc在回调函数上调用INotifyCompletion.OnCompleted。

在调用继续回调之前，在异步事件实际发生时调用GCHandle.Free。

如果事件从未发生，则实现IDispose来调用GCHandle.Free。

鉴于此，下面是原始计时器回调代码的一个版本，该版本可以正常工作。请注意，无需牢记计时器回调委托(delegate)(WaitOrTimerCallbackProc callback)。它作为状态机的一部分保持 Activity 状态。 更新了:如@svick所指出，此语句可能特定于状态机的当前实现(C#5.0)。我添加了GC.KeepAlive(callback)以消除对此行为的任何依赖，以防将来的编译器版本中它发生更改。

using System;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;

namespace ConsoleApplication
{
    class Program
    {
        // Test task
        static async Task TestAsync(CancellationToken token)
        {
            using (var awaiter = new Awaiter())
            {
                WaitOrTimerCallbackProc callback = (a, b) =>
                    awaiter.Continue();
                try
                {
                    IntPtr timerHandle;
                    if (!CreateTimerQueueTimer(out timerHandle,
                            IntPtr.Zero,
                            callback,
                            IntPtr.Zero, 500, 500, 0))
                        throw new System.ComponentModel.Win32Exception(
                            Marshal.GetLastWin32Error());
                    try
                    {
                        var i = 0;
                        while (true)
                        {
                            token.ThrowIfCancellationRequested();
                            await awaiter;
                            Console.WriteLine("tick: " + i++);
                        }
                    }
                    finally
                    {
                        DeleteTimerQueueTimer(IntPtr.Zero, timerHandle, IntPtr.Zero);
                    }
                }
                finally
                {
                    // reference the callback at the end
                    // to avoid a chance for it to be GC'ed
                    GC.KeepAlive(callback);
                }
            }
        }

        // Entry point
        static void Main(string[] args)
        {
            // cancel in 3s
            var testTask = TestAsync(new CancellationTokenSource(10 * 1000).Token);

            Thread.Sleep(1000);
            GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced, true);

            Thread.Sleep(2000);
            Console.WriteLine("Press Enter to GC...");
            Console.ReadLine();

            GC.Collect(GC.MaxGeneration, GCCollectionMode.Forced);
            Console.WriteLine("Press Enter to exit...");
            Console.ReadLine();
        }

        // Custom awaiter
        public class Awaiter :
            System.Runtime.CompilerServices.INotifyCompletion,
            IDisposable
        {
            Action _continuation;
            GCHandle _hold = new GCHandle();

            public Awaiter()
            {
                Console.WriteLine("Awaiter()");
            }

            ~Awaiter()
            {
                Console.WriteLine("~Awaiter()");
            }

            void ReleaseHold()
            {
                if (_hold.IsAllocated)
                    _hold.Free();
            }

            // resume after await, called upon external event
            public void Continue()
            {
                Action continuation;

                // it's OK to use lock (this)
                // the C# compiler would never do this,
                // because it's slated to work with struct awaiters
                lock (this)
                {
                    continuation = _continuation;
                    _continuation = null;
                    ReleaseHold();
                }

                if (continuation != null)
                    continuation();
            }

            // custom Awaiter methods
            public Awaiter GetAwaiter()
            {
                return this;
            }

            public bool IsCompleted
            {
                get { return false; }
            }

            public void GetResult()
            {
            }

            // INotifyCompletion
            public void OnCompleted(Action continuation)
            {
                lock (this)
                {
                    ReleaseHold();
                    _continuation = continuation;
                    _hold = GCHandle.Alloc(_continuation);
                }
            }

            // IDispose
            public void Dispose()
            {
                lock (this)
                {
                    _continuation = null;
                    ReleaseHold();
                }
            }
        }

        // p/invoke
        delegate void WaitOrTimerCallbackProc(IntPtr lpParameter, bool TimerOrWaitFired);

        [DllImport("kernel32.dll")]
        static extern bool CreateTimerQueueTimer(out IntPtr phNewTimer,
            IntPtr TimerQueue, WaitOrTimerCallbackProc Callback, IntPtr Parameter,
            uint DueTime, uint Period, uint Flags);

        [DllImport("kernel32.dll")]
        static extern bool DeleteTimerQueueTimer(IntPtr TimerQueue, IntPtr Timer,
            IntPtr CompletionEvent);
    }
}