.NET TPL CancellationToken内存泄漏

Question

我开发了一个库,它实现了工作项的生产者/消费者模式.工作已经出列,并且每个出列的工作项都会出现一个单独的任务,其中包含失败和成功的延续.

任务继续在工作项完成(或失败)后重新排队.

整个库共享一个中心CancellationTokenSource,在应用程序关闭时触发.

我现在面临重大的内存泄漏.如果使用取消令牌作为参数创建任务,则任务似乎保留在内存中,直到触发取消源(并稍后处置).

这可以在此示例代码(VB.NET)中重现.主要任务是包装工作项的任务,继续任务将处理重新安排.

Dim oCancellationTokenSource As New CancellationTokenSource
Dim oToken As CancellationToken = oCancellationTokenSource.Token
Dim nActiveTasks As Integer = 0

Dim lBaseMemory As Long = GC.GetTotalMemory(True)

For iteration = 0 To 100 ' do this 101 times to see how much the memory increases

  Dim lMemory As Long = GC.GetTotalMemory(True)

  Console.WriteLine("Memory at iteration start: " & lMemory.ToString("N0"))
  Console.WriteLine("  to baseline: " & (lMemory - lBaseMemory).ToString("N0"))

  For i As Integer = 0 To 1000 ' 1001 iterations to get an immediate, measurable impact
    Interlocked.Increment(nActiveTasks)
    Dim outer As Integer = i
    Dim oMainTask As New Task(Sub()
                                ' perform some work
                                Interlocked.Decrement(nActiveTasks)
                              End Sub, oToken)
    Dim inner As Integer = 1
    Dim oFaulted As Task = oMainTask.ContinueWith(Sub()
                                                    Console.WriteLine("Failed " & outer & "." & inner)
                                                    ' if failed, do something with the work and re-queue it, if possible
                                                    ' (imagine code for re-queueing - essentially just a synchronized list.add)

                                                                                                            ' Does not help:
                                                    ' oMainTask.Dispose()
                                                  End Sub, oToken, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default)
    ' if not using token, does not cause increase in memory:
    'End Sub, TaskContinuationOptions.OnlyOnFaulted)

            ' Does not help:
    ' oFaulted.ContinueWith(Sub()
    '                         oFaulted.Dispose()
    '                       End Sub, TaskContinuationOptions.NotOnFaulted)


    Dim oSucceeded As Task = oMainTask.ContinueWith(Sub()
                                                      ' success
                                                      ' re-queue for next iteration
                                                      ' (imagine code for re-queueing - essentially just a synchronized list.add)

                                                                                                                ' Does not help:
                                                      ' oMainTask.Dispose()
                                                    End Sub, oToken, TaskContinuationOptions.OnlyOnRanToCompletion, TaskScheduler.Default)
    ' if not using token, does not cause increase in memory:
    'End Sub, TaskContinuationOptions.OnlyOnRanToCompletion)

            ' Does not help:
    ' oSucceeded.ContinueWith(Sub()
    '                           oSucceeded.Dispose()
    '                         End Sub, TaskContinuationOptions.NotOnFaulted)


    ' This does not help either and makes processing much slower due to the thrown exception (at least one of these tasks is cancelled)
    'Dim oDisposeTask As New Task(Sub()
    '                               Try
    '                                 Task.WaitAll({oMainTask, oFaulted, oSucceeded, oFaultedFaulted, oSuccededFaulted})
    '                               Catch ex As Exception

    '                               End Try
    '                               oMainTask.Dispose()
    '                               oFaulted.Dispose()
    '                               oSucceeded.Dispose()                                     
    '                             End Sub)

    oMainTask.Start()
    '  oDisposeTask.Start()
  Next

  Console.WriteLine("Memory after creating tasks: " & GC.GetTotalMemory(True).ToString("N0"))

  ' Wait until all main tasks are finished (may not mean that continuations finished)

  Dim previousActive As Integer = nActiveTasks
  While nActiveTasks > 0
    If previousActive <> nActiveTasks Then
      Console.WriteLine("Active: " & nActiveTasks)
      Thread.Sleep(500)
      previousActive = nActiveTasks
    End If

  End While

  Console.WriteLine("Memory after tasks finished: " & GC.GetTotalMemory(True).ToString("N0"))

Next

我使用ANTS Memory Profiler测量了内存使用情况,并看到System.Threading.ExecutionContext大幅增加,后者追溯到任务延续和CancellationCallbackInfo.

如您所见,我已经尝试处理使用取消令牌的任务,但这似乎没有任何效果.

编辑

我正在使用.NET 4.0

更新

即使只是在主要任务的链接中继续失败,内存使用也会不断增加.任务继续似乎阻止取消令牌注册的取消注册.

因此,如果一个任务被连接一个没有运行的延续(由于TaskContinuationOptions),那么似乎存在内存泄漏.如果只有一个延续,运行,那么我没有观察到内存泄漏.

解决方法

作为一种解决方法,我可以在没有任何TaskContinuationOptions处理的情况下执行单个延续,并在那里处理父任务的状态:

oMainTask.ContinueWith(Sub(t)
                     If t.IsCanceled Then
                       ' ignore
                     ElseIf t.IsCompleted Then
                       ' reschedule

                     ElseIf t.IsFaulted Then
                       ' error handling

                     End If
                   End Sub)

我将不得不检查在取消的情况下它的表现如何,但这似乎可以解决问题.我几乎怀疑.NET Framework中有一个错误.具有相互排斥条件的任务取消并非如此罕见.

Answer 1

一些观察

1. 潜在的泄漏仅在存在未运行的任务“分支”的情况下才出现。在您的示例中，如果您注释掉该oFaulted任务，泄漏就会消失。如果您更新代码以出现oMainTask错误，从而使oFaulted任务运行并且oSucceeded任务不运行，则注释掉oSucceeded可以防止泄漏。
2. 也许没有帮助，但如果您oCancellationTokenSource.Cancel()在所有任务运行后调用，内存就会释放。处置没有帮助，处置取消源与任务的任何组合也没有帮助。
3. 我看了一下http://referencesource.microsoft.com/，它是4.5.2（有没有办法查看早期框架？）我知道它不一定相同，但它有助于了解正在发生什么类型的事情。基本上，当您将取消令牌传递给任务时，任务会向取消令牌的取消源注册自身。因此，取消源保存了对您所有任务的引用。我还不清楚为什么你的场景似乎正在泄漏。如果我发现任何东西，我会在有机会更深入地研究后进行更新。

解决方法

将分支逻辑移至始终运行的延续。

Dim continuation As Task =
    oMainTask.ContinueWith(
        Sub(antecendent)
            If antecendent.Status = TaskStatus.Faulted Then
                'Handle errors
            ElseIf antecendent.Status = TaskStatus.RanToCompletion Then
                'Do something else
            End If
        End Sub,
        oToken,
        TaskContinuationOptions.None,
        TaskScheduler.Default)

无论如何，这很可能比其他方法更轻。在这两种情况下，始终会运行一个延续任务，但使用此代码时，只会创建 1 个延续任务，而不是 2 个。

Answer 2

我能够通过移动这两行来解决 .net 4.0 下的问题

Dim oCancellationTokenSource As New CancellationTokenSource
Dim oToken As CancellationToken = oCancellationTokenSource.Token

在第一个循环内

然后在循环结束时

oToken = Nothing
oCancellationTokenSource.Dispose()

我也移动了

Interlocked.Decrement(nActiveTasks)

自以来的每个“最终”任务中

While nActiveTasks > 0

不会准确。

这是有效的代码

Imports System.Threading.Tasks
Imports System.Threading

Module Module1

Sub Main()
    Dim nActiveTasks As Integer = 0

    Dim lBaseMemory As Long = GC.GetTotalMemory(True)

    For iteration = 0 To 100 ' do this 101 times to see how much the memory increases
        Dim oCancellationTokenSource As New CancellationTokenSource
        Dim oToken As CancellationToken = oCancellationTokenSource.Token
        Dim lMemory As Long = GC.GetTotalMemory(True)

        Console.WriteLine("Memory at iteration start: " & lMemory.ToString("N0"))
        Console.WriteLine("  to baseline: " & (lMemory - lBaseMemory).ToString("N0"))

        For i As Integer = 0 To 1000 ' 1001 iterations to get an immediate, measurable impact
            Dim outer As Integer = iteration
            Dim inner As Integer = i

            Interlocked.Increment(nActiveTasks)

            Dim oMainTask As New Task(Sub()
                                          ' perform some work
                                      End Sub, oToken, TaskCreationOptions.None)

            oMainTask.ContinueWith(Sub()
                                       Console.WriteLine("Failed " & outer & "." & inner)
                                       Interlocked.Decrement(nActiveTasks)
                                   End Sub, oToken, TaskContinuationOptions.OnlyOnFaulted, TaskScheduler.Default)


            oMainTask.ContinueWith(Sub()
                                       If inner Mod 250 = 0 Then Console.WriteLine("Success " & outer & "." & inner)
                                       Interlocked.Decrement(nActiveTasks)
                                   End Sub, oToken, TaskContinuationOptions.OnlyOnRanToCompletion, TaskScheduler.Default)


            oMainTask.Start()
        Next

        Console.WriteLine("Memory after creating tasks: " & GC.GetTotalMemory(True).ToString("N0"))


        Dim previousActive As Integer = nActiveTasks
        While nActiveTasks > 0
            If previousActive <> nActiveTasks Then
                Console.WriteLine("Active: " & nActiveTasks)
                Thread.Sleep(500)
                previousActive = nActiveTasks
            End If

        End While

        oToken = Nothing
        oCancellationTokenSource.Dispose()

        Console.WriteLine("Memory after tasks finished: " & GC.GetTotalMemory(True).ToString("N0"))

    Next

    Console.WriteLine("Final Memory after finished: " & GC.GetTotalMemory(True).ToString("N0"))

    Console.Read()
End Sub

End Module