Catalog/imagecatalog/Services/ImageProcessingCoordinator.cs

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

namespace ImageCatalog_2.Services
{
    public class ImageProcessingCoordinator : IImageProcessingCoordinator
    {
        private readonly ImageCreationService _imageCreationService;
        private readonly ILogger<ImageProcessingCoordinator> _logger;

        [CLSCompliant(false)]
        public ImageProcessingCoordinator(
            ImageCreationService imageCreationService,
            ILogger<ImageProcessingCoordinator> logger)
        {
            _imageCreationService = imageCreationService;
            _logger = logger;
        }

        public async Task<ImageProcessingRunResult> RunAsync(
            ImageProcessingRunRequest request,
            CancellationToken token,
            Action<ImageProcessedUpdate> onImageProcessed,
            Action<string> onSpeedUpdated)
        {
            var results = new ConcurrentBag<string>();
            var recentDiffs = new Queue<int>();
            const int recentWindowSize = 5;

            int currentAmount = 0;
            int previousAmount = 0;
            int processedAtomic = 0;

            var speedWatch = Stopwatch.StartNew();
            using var speedTimer = new System.Threading.Timer(_ =>
            {
                try
                {
                    previousAmount = currentAmount;
                    currentAmount = Volatile.Read(ref processedAtomic);
                    int diff = currentAmount - previousAmount;
                    if (diff < 0)
                    {
                        diff = 0;
                    }

                    lock (recentDiffs)
                    {
                        recentDiffs.Enqueue(diff);
                        if (recentDiffs.Count > recentWindowSize)
                        {
                            recentDiffs.Dequeue();
                        }
                    }

                    double avgRecent;
                    lock (recentDiffs)
                    {
                        avgRecent = recentDiffs.Count == 0 ? 0.0 : recentDiffs.Average();
                    }

                    double overall = 0.0;
                    if (speedWatch.Elapsed.TotalSeconds >= 1)
                    {
                        var elapsedSeconds = speedWatch.Elapsed.TotalSeconds;
                        var total = Volatile.Read(ref processedAtomic);
                        overall = elapsedSeconds > 0 ? total / elapsedSeconds : 0.0;
                    }

                    var recentPerMin = avgRecent * 60.0;
                    var elapsed = speedWatch.Elapsed;
                    int hours = (int)elapsed.TotalHours;
                    int minutes = elapsed.Minutes;
                    int seconds = elapsed.Seconds;
                    var elapsedStr = $"{hours}h {minutes}m {seconds}s";

                    var speedText = $"{avgRecent:0.00} f/s (media: {overall:0.00} f/s) - {elapsedStr}{Environment.NewLine}media: {recentPerMin:0.00} f/m";
                    onSpeedUpdated(speedText);
                }
                catch (Exception ex)
                {
                    _logger.LogDebug(ex, "Failed to update speed counter");
                }
            }, null, TimeSpan.FromSeconds(1), TimeSpan.FromSeconds(1));

            EventHandler<Tuple<string, int>> onImageProcessedInternal = (_, args) =>
            {
                var processed = Interlocked.Increment(ref processedAtomic);
                onImageProcessed(new ImageProcessedUpdate(args.Item1, args.Item2, processed));
            };

            await _imageCreationService.CreaCatalogoParallel(
                request.Options,
                results,
                onImageProcessedInternal,
                token).ConfigureAwait(false);

            speedWatch.Stop();

            var finalProcessed = Volatile.Read(ref processedAtomic);
            double overallAvg = 0.0;
            double overallPerMin = 0.0;
            if (speedWatch.Elapsed.TotalSeconds > 0.0)
            {
                overallAvg = finalProcessed / speedWatch.Elapsed.TotalSeconds;
                overallPerMin = overallAvg * 60.0;
            }

            var finalElapsed = speedWatch.Elapsed;
            int finalHours = (int)finalElapsed.TotalHours;
            int finalMinutes = finalElapsed.Minutes;
            int finalSeconds = finalElapsed.Seconds;

            return new ImageProcessingRunResult
            {
                FinalSpeedCounter = $"{finalHours}h {finalMinutes}m {finalSeconds}s{Environment.NewLine}media: {overallAvg:0.00} f/s{Environment.NewLine}media: {overallPerMin:0.00} f/m"
            };
        }
    }
}