Агрегация двух параллельных методов

Question

Недавно столкнулся со следующей проблемой:

Допустим у нас есть некий метод, который обязан вернуть некий объект/некоторое значение как результат выполнения своей работы. В свою очередь - этот самый метод общается с несколькими источниками ввода (клавиатура/слушает сокет/какое либо другое устройство ввода вывода/ждет какого либо колбека из вне).

В моем примере (который я представлю ниже) пускай их будет только два. Каждый из этих источников может ожидать ввод достаточно продолжительное время. Допустим внутри методов 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 источникам ввода. Что позволит мне из одной точки говорить всем другим о источникам прекращение своей работы. Если источников будет более чем 2, то количество аргументов тоже возрастет (на самом деле даже в том случае в котором я встретился с этой проблемой их два).

Собственно интересует как решить данную проблему. Мое предположение с делегатом верное? Случались ли такие же проблемы у Вас и как вы их решили?

Answer 1

_{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);
}

_{DelayInputSource.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;
        }
    }
}

_App.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;
        }
    }
}

_{Console output}

first
<null>

^{Примечание. Вывод обеих строчек происходит одновременно (через ~2 секунды после запуска).}

БЛОГ НА HUSL

Агрегация двух параллельных методов

Ответы (1 шт):