c# - 如何在不冒应用程序失败风险的情况下调用 SemaphoreSlim.Release()？

Question

我正在尝试使用新的 SemaphoreSlim .NET 4.0 中的类来限制可以无限期运行的快节奏循环的速率。在对此进行单元测试时，我发现如果循环足够紧密且并行度足够高，SemaphoreSlim当你调用 Release() 时会抛出一个无法捕获的异常, 即使您检查了 .Count属性优先，并在整个检查计数/释放序列期间锁定信号量实例本身。

此异常会导致应用程序停止运行。据我所知，没有捕获它。

深入挖掘，我发现 SemaphoreSlim正在尝试访问它自己的 .AvailableWaitHandle Release()期间的内部属性(property)调用，它在那里抛出异常，而不是我访问 SemaphoreSlim实例本身。 (我必须使用调试-> 异常-> 公共(public)语言运行时异常-> 在 Visual Studio 中抛出所有检查来发现这一点；您无法在运行时捕获它。有关更多详细信息，请参见 The Uncatchable Exception。)

我的问题是，在这种情况下，有没有人知道使用此类而不会冒应用立即终止风险的安全方法？

注意:信号量实例被包裹在一个RateGate实例中，其代码可以在这篇文章中找到:Better Rate Limiting in .NET .

更新: 我正在添加完整的控制台应用程序代码来重现。两个答案都有助于解决问题；请参阅下面的解释。

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading.Tasks;
using System.Linq;
using System.Text;
using System.Threading;
using PennedObjects.RateLimiting;

namespace RateGateForceTerminateDemo
{
    class Program
    {
        static int _secondsToRun = 10;
        static void Main(string[] args) {
            AppDomain.CurrentDomain.UnhandledException += new UnhandledExceptionEventHandler(CurrentDomain_UnhandledException);
            OptimizeMaxThreads();
            Console.WriteLine();
            Console.WriteLine("Press any key to exit.");
            Console.ReadKey(true);
        }

        static void CurrentDomain_UnhandledException(object sender, UnhandledExceptionEventArgs e) {
            Console.WriteLine("Unhandled exception, terminating={0}:{1}", e.IsTerminating, e.ExceptionObject.ToString());
            Console.WriteLine("Press any key to terminate app.");
            Console.ReadKey(true);
        }

        static void OptimizeMaxThreads() {
            int processors = Environment.ProcessorCount;
            int processorsSq = Convert.ToInt32(Math.Pow(processors,2));
            int threads = 1;
            double result;
            Tuple<int, double> maxResult = new Tuple<int, double>(threads, 0);
            while (threads <= processorsSq) {
                Console.WriteLine("Running for {0}s with upper limit of {1} threads... ", _secondsToRun, threads);
                result = TestThrottling(10000000, threads, _secondsToRun);
                Console.WriteLine("Ok. Result is {0:N0} ops/s", result);
                Console.WriteLine();
                if(result > maxResult.Item2)
                    maxResult = new Tuple<int, double>(threads, result);
                threads *= 2;
            }
            Console.WriteLine("{0} threads achieved max throughput of {1:N0}", maxResult.Item1, maxResult.Item2);
        }
        static double TestThrottling(int limitPerSecond, int maxThreads, int maxRunTimeSeconds) {
            int completed = 0;
            RateGate gate = new RateGate(limitPerSecond, TimeSpan.FromSeconds(1));
            ParallelLoopResult res = new ParallelLoopResult();
            ParallelOptions parallelOpts = new ParallelOptions() { MaxDegreeOfParallelism = maxThreads };
            Stopwatch sw = Stopwatch.StartNew();
            try {
                res = Parallel.For<int>(0, 1000000000, parallelOpts, () => 0, (num, state, subtotal) =>
                {
                    bool succeeded = gate.WaitToProceed(10000);
                    if (succeeded) {
                        subtotal++;
                    }
                    else {
                        Console.WriteLine("Gate timed out for thread {0}; {1:N0} iterations, elapsed {2}.", Thread.CurrentThread.ManagedThreadId, subtotal, sw.Elapsed);
                        // return subtotal;
                    }
                    if (sw.Elapsed.TotalSeconds > maxRunTimeSeconds) {
                        Console.WriteLine("MaxRunTime expired for thread {0}, last succeeded={1}, iterations={2:N0}, elapsed={3}.", Thread.CurrentThread.ManagedThreadId, succeeded, subtotal, sw.Elapsed);
                        state.Break();
                    }
                    return subtotal;
                }, (subtotal) => Interlocked.Add(ref completed, subtotal));
            }
            catch (AggregateException aggEx) {
                Console.WriteLine(aggEx.Flatten().ToString());
            }
            catch (Exception ex) {
                Console.WriteLine(ex);
            }
            sw.Stop();
            double throughput = completed / Math.Max(sw.Elapsed.TotalSeconds, 1);
            Console.WriteLine("Done at {0}, finished {1:N0} iterations, IsCompleted={2}, LowestBreakIteration={3:N0}, ",
                sw.Elapsed,
                completed,
                res.IsCompleted,
                (res.LowestBreakIteration.HasValue ? res.LowestBreakIteration.Value : double.NaN));
            Console.WriteLine();
            //// Uncomment the following 3 lines to stop prevent the ObjectDisposedException:
            //Console.WriteLine("We should not hit the dispose statement below without a console pause.");
            //Console.Write("Hit any key to continue... ");
            //Console.ReadKey(false);
            gate.Dispose();
            return throughput;
        }
    }
}

因此使用@dtb 的解决方案，线程“a”仍然有可能通过 _isDisposed检查，但线程“b”在线程“a”命中 Release() 之前处理信号量.我发现在 ExitTimerCallback 和 Dispose 方法中都在 _semaphore 实例周围添加了一个锁。 @Peter Ritchie 的建议让我在处理信号量之前取消并处理计时器。这两件事的结合让程序完成并正确处理 RateGate，无一异常(exception)。

因为如果没有那个输入我就不会得到这个我不想回答自己。但是，由于 StackOverflow 在提供完整答案时更有用，所以我会接受首先发布补丁或伪补丁并在上述控制台应用程序中成功存活的人。

Answer 1

问题出在您使用的 RateGate 类中。它有一个内部 Timer，即使在 RateGate 实例被释放后也会运行它的代码。此代码包括对已处置的 SemaphoreSlim 的 Release 调用。

修复:

@@ -88,7 +88,8 @@
    int exitTime;
    while (_exitTimes.TryPeek(out exitTime)
            && unchecked(exitTime - Environment.TickCount) <= 0)
    {
+       if (_isDisposed) return;
        _semaphore.Release();
        _exitTimes.TryDequeue(out exitTime);
    }