主页 / 问题 / 1169690
Accepted
Peter Lavreniuk
Asked:2020-08-24 16:18:02 +0000 UTC2020-08-24 16:18:02 +0000 UTC

两种并行方法的聚合

  • 772

我最近遇到了以下问题：

假设我们有一些方法必须返回一些对象/一些值作为其工作的结果。反过来，这个相同的方法与多个输入源通信（键盘/侦听套接字/其他一些输入/输出设备/等待来自外部的一些回调）。

在我的示例中（我将在下面介绍），假设只有两个。这些源中的每一个都可以等待输入相当长的时间。假设在方法内部RequestFirstInput并且RequestSecondInput有对设备轮询类的对象的调用，这反过来又保证了以下内容：

  • 成功完成工作的结果是一些价值/一些对象
  • 设备轮询类的设计使其了解输入的状态 - ожидание начала ввода/ непосредственно ввод/ ввод завершился/ошибка при вводе
  • 轮询设备不会无限期地完成它的工作。它有一个内部超时。并且在状态没有改变ожидания начала ввода或者输入由于某种原因停止并且没有接收到新数据之后，它可以下降TimeoutException

此外，如果您источник номер 1已经开始接收信息或有输入结果，则无需其他来源，在此示例中источнике номер 2。也就是说——如果一个来源开始从外部接收信息，他有义务通知其他人。

这是一个例子：

    var threadCount = 2;
    var resetEvent = new ManualResetEvent(false);
    int? firstInputResult = 0;
    int? secondInputResult = 0;

    var firstInputCancellationTokenSource = new CancellationTokenSource();
    var secondInputCancellationTokenSource = new CancellationTokenSource();

    ThreadPool.QueueUserWorkItem(o =>
    {
        try
        {
            firstInputResult = RequestFirstInput(source: firstInputCancellationTokenSource,
                                                 token: secondInputCancellationTokenSource.Token);
        }
        catch
        {
            //ignore
        }
        finally
        {
            if (Interlocked.Decrement(ref threadCount) == 0)
                resetEvent.Set();
        }
    });

    ThreadPool.QueueUserWorkItem(o =>
    {
        try
        {
            secondInputResult = RequestSecondInput(source: secondInputCancellationTokenSource,
                                                   token: firstInputCancellationTokenSource.Token);
        }
        catch
        {
            //ignore
        }
        finally
        {
            if (Interlocked.Decrement(ref threadCount) == 0)
                resetEvent.Set();
        }
    });

    resetEvent.WaitOne();

    return ConsolidateInputResult(firstInputResult, secondInputResult);

这是方法签名ConsolidateInputResult，RequestFirstInput以及RequestSecondInput

private int ConsolidateInputResult(int? firstInput, int? secondInput)
private int? RequestFirstInput(CancellationTokenSource source, CancellationToken token)
private int? RequestSecondInput(CancellationTokenSource source, CancellationToken token)

在这个例子中，我大致展示了如何解决这个问题。但我根本不喜欢这个解决方案。将对象转移到某个地方是不对的，CancellationTokenSource并且这个人可以影响拥有属于 的令牌的人CancellationTokenSource。

或者，您可以更改方法的签名RequestFirstInput和RequestSecondInput. 直接从参数中删除自己CancellationTokenSource，并将委托作为新参数传递给那里。由于委托将允许我自己在上面的级别做出决定，而不是将控制限制在其他 N 个输入源。这将允许我从一个角度告诉其他所有人他们工作终止的来源。如果有两个以上的来源，那么参数的数量也会增加（事实上，即使在我遇到这个问题的情况下，也有两个）。

实际上对如何解决这个问题很感兴趣。我对代表的假设是否正确？你有没有同样的问题，你是如何解决的？

c#

1 个回答

  1. Best Answer

    IInputSource.cs

    using System.Threading;

namespace RootNamespace
{
    public interface IInputSource
    {
        event InputReadedCallback InputReaded;

        bool WaitForInput(CancellationToken cancellationToken = default);
        bool TryReadInput(out object? input);
    }
}

    InputReadedCallback.cs

    namespace RootNamespace
{
    public delegate void InputReadedCallback(IInputSource source, object? input);
}

    延迟输入源.cs

    using System;
using System.Threading;

namespace RootNamespace
{
    public class DelayInputSource : IInputSource
    {
        readonly object? input;
        bool inputWaited;

        public TimeSpan Delay { get; }

        public event InputReadedCallback? InputReaded;

        public DelayInputSource(TimeSpan delay, object? input = null)
        {
            Delay = delay;
            this.input = input;
        }

        public bool WaitForInput(CancellationToken cancellationToken = default)
        {
            var rsetEv = new ManualResetEventSlim();

            using var ctr = cancellationToken.Register(e => (e as ManualResetEventSlim)!.Set(), rsetEv);

            inputWaited = !rsetEv.Wait(Delay);

            return inputWaited;
        }
        public bool TryReadInput(out object? input)
        {
            input = null;

            if (!inputWaited) return false;

            input = this.input;

            InputReaded?.Invoke(this, input);

            inputWaited = false;

            return true;
        }
    }
}

    应用程序.cs

    using System;
using System.Threading;
using System.Threading.Tasks;

namespace RootNamespace
{
    static class App
    {
        static void Main()
        {
            var firInputSrc = new DelayInputSource(TimeSpan.FromSeconds(2.0), "first");
            var secInputSrc = new DelayInputSource(Timeout.InfiniteTimeSpan, "second");
            using var firCts = new CancellationTokenSource(TimeSpan.FromSeconds(3.0));
            using var secCts = new CancellationTokenSource(TimeSpan.FromSeconds(10.0));
            firInputSrc.InputReaded += (s, i) => secCts.Cancel();
            secInputSrc.InputReaded += (s, i) => firCts.Cancel();

            var firInputTask = new Task<object?>(InputReadCallback, ((IInputSource)firInputSrc, firCts.Token));
            firInputTask.Start();
            var secInputTask = new Task<object?>(InputReadCallback, ((IInputSource)secInputSrc, secCts.Token));
            secInputTask.Start();

            Console.WriteLine(firInputTask.Result ?? "<null>");
            Console.WriteLine(secInputTask.Result ?? "<null>");

            Console.ReadKey(true);
        }
        static object? InputReadCallback(object? state)
        {
            var stateAsTuple = ((IInputSource, CancellationToken))state!;
            var inputSrc = stateAsTuple.Item1;
            var ct = stateAsTuple.Item2;

            if (inputSrc.WaitForInput(ct))
            {
                if (inputSrc.TryReadInput(out var input))
                    return input;
                else
                    return null;
            }
            else
                return null;
        }
    }
}

    控制台输出

    first
<null>

    笔记。两条线的输出同时发生（在开始后约 2 秒内）。

    • 1

相关问题