EVCB_OCPP.OCPPServer.16J/EVCB_OCPP.WSServer/Helper/GroupHandlerIO.cs

using Microsoft.Extensions.Logging;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace EVCB_OCPP.WSServer.Helper;

public class GroupHandler<TI,TO> where TI : class
{
    public GroupHandler(
        Func<BundleHandlerData<TI,TO>, Task> handleFunc,
        ILogger logger, int workerCnt = 1, int maxRetry = 10)
    {
        this.handleFunc = handleFunc;
        this.logger = logger;

        this.maxRetry = maxRetry;
        workersLock = new SemaphoreSlim(workerCnt);
        //singleWorkLock = new(_WorkerCnt);
    }

    private readonly Func<BundleHandlerData<TI, TO>, Task> handleFunc;
    private readonly ILogger logger;
    private readonly int maxRetry;
    private readonly ConcurrentQueue<WaitParam<TI, TO>> waitList = new();

    private SemaphoreSlim workersLock;// = new SemaphoreSlim(1);

    public async Task<TO> HandleAsync(TI param)
    {
        var waitData = new WaitParam<TI,TO>() { Data = param, Waiter = new SemaphoreSlim(0), Exception = null };
        waitList.Enqueue(waitData);
        TryStartHandler();
        await waitData.Waiter.WaitAsync();
        if (waitData.Exception is not null)
        {
            throw waitData.Exception;
        }
        return waitData.Result;
    }

    private void TryStartHandler()
    {
        if (!workersLock.Wait(0))
        {
            return;
        }

        if (waitList.Count == 0)
        {
            workersLock.Release();
            return;
        }

        _ = StartHandleTask();
    }

    private async Task StartHandleTask()
    {
        var timer = Stopwatch.StartNew();
        List<long> times = new();

        var requests = new List<WaitParam<TI,TO>>();

        while (waitList.TryDequeue(out var handle))
        {
            requests.Add(handle);
        }
        times.Add(timer.ElapsedMilliseconds);

        int cnt = 0;
        Exception lastException = null;
        var datas = requests.Select(x => x.Data).ToList();

        for (; cnt < maxRetry; cnt++)
        {
            var bundleHandledata = new BundleHandlerData<TI, TO>(datas);
            try
            {
                await handleFunc(bundleHandledata);
            }
            catch (Exception e)
            {
                lastException = e;
            }

            var completedKeys = bundleHandledata.CompletedDatas.Select(x => x.Key).ToList();
            var completedRequests = requests.Where(x => completedKeys.Any(y=> y == x.Data)).ToList();
            foreach (var request in completedRequests)
            {
                var result = bundleHandledata.CompletedDatas.FirstOrDefault(x => x.Key == request.Data);
                request.Result = result.Value;
                request.Waiter.Release();
            }

            //datas = datas.Except(bundleHandledata.CompletedDatas).ToList();
            datas = datas.Where(x => !completedKeys.Contains(x)).ToList();

            if (datas == null || datas.Count == 0)
            {
                break;
            }
            logger.LogError(lastException?.Message);
            times.Add(timer.ElapsedMilliseconds);
        }

        var uncompletedRequests = requests.Where(x => datas.Contains(x.Data)).ToList();
        foreach (var request in uncompletedRequests)
        {
            request.Exception = lastException;
            request.Waiter.Release();
        }
        workersLock.Release();

        timer.Stop();
        if (timer.ElapsedMilliseconds > 1000)
        {
            logger.LogWarning($"StartHandleTask {string.Join("/", times)}");
        }

        TryStartHandler();
    }
}

public class BundleHandlerData<TI,TO>
{
    public List<TI> Datas { get; set; }
    public List<KeyValuePair<TI,TO>> CompletedDatas { get; set; }

    public BundleHandlerData(List<TI> datas)
    {
        Datas = datas;
        CompletedDatas = new();
    }

    public void AddCompletedData(TI competedData, TO result)
    {
        CompletedDatas.Add(new KeyValuePair<TI, TO>(competedData, result)); ;
    }
}

internal class WaitParam<TI,TO>
{
    public TI Data { get; init; }
    public TO Result { get; set; }
    public SemaphoreSlim Waiter { get; init; }
    public Exception Exception { get; set; }
}