3

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

此异常会关闭应用程序，期间。据我所知，没有抓住它。

深入挖掘，我发现在调用期间SemaphoreSlim试图在.AvailableWaitHandle内部访问它自己的属性Release()，它在那里抛出异常，而不是我访问SemaphoreSlim实例本身。（我必须使用 Debug->Exceptions->Common Language Runtime Exceptions->Thrown all check in 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。

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

4

2 回答 2

4

问题出在您正在使用的 RateGate 类中。它有一个内部计时器，即使在处理了 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);
    }
于 2012-05-10T03:03:50.753 回答
1

根据您的评论，听起来您已经产生了一堆线程来处理 RateGate 对象，这些对象在这些线程完成使用之前就被释放了。即代码从 using 块退出后，线程仍在运行。更新：如果您按照评论中的描述进行操作；但不要使用 using 块，您不会遇到问题。我目睹的异常实际上是 ObjectDisposedException；如果 RateGate 在使用它完成代码之前被处置，这将是有意义的......

于 2012-05-10T02:53:03.707 回答