AsyncProcessor.cs

using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Waher.Events;
 
namespace Waher.Runtime.Queue
{
    public class AsyncProcessor<T> : IAsyncProcessor
        where T : class, IWorkItem
    {
        private readonly CancellationTokenSource[] cancelWaitSources;
        private readonly CancellationTokenSource[] cancelWorkSources;
        private readonly T[] currentWorkItem;
        private readonly Task[] processors;
        private readonly string name;
        private readonly int nrProcessors;
        private LinkedListNode<IAsyncProcessor> processorNode;
        private AsyncQueue<T> queue = new AsyncQueue<T>();
        private int nrProcessorsRunning;
        private bool terminating = false;
        private bool terminated = false;
        private bool idle = true;
 
        [Obsolete("Use the constructor with a Name argument.")]
        public AsyncProcessor(int NrProcessors)
            : this(NrProcessors, Guid.NewGuid().ToString())
        {
        }
 
        public AsyncProcessor(int NrProcessors, string Name)
        {
            if (NrProcessors <= 0)
                throw new ArgumentException("Number of processors must be positive.", nameof(NrProcessors));
 
            int i;
 
            this.name = Name;
            this.nrProcessors = this.nrProcessorsRunning = NrProcessors;
            this.processors = new Task[NrProcessors];
 
            this.cancelWaitSources = new CancellationTokenSource[NrProcessors];
            this.cancelWorkSources = new CancellationTokenSource[NrProcessors];
            this.currentWorkItem = new T[NrProcessors];
            this.processorNode = AsyncProcessors.RegisterProcessor(this);
 
            for (i = 0; i < this.nrProcessors; i++)
                this.processors[i] = this.PerformWork(i);
        }
 
        [Obsolete("Use the DisposeAsync() method.")]
        public void Dispose()
        {
            this.DisposeAsync().Wait();
        }
 
        public async Task DisposeAsync()
        {
            this.terminating = true;
 
            if (!(this.processorNode is null))
            {
                AsyncProcessors.UnregisterProcessor(this.processorNode);
                this.processorNode = null;
            }
 
            if (!this.terminated)
            {
                for (int i = 0; i < this.nrProcessors; i++)
                {
                    this.cancelWaitSources[i]?.Cancel();
                    this.cancelWorkSources[i]?.Cancel();
                }
 
                await Task.WhenAll(this.processors);
                this.terminated = true;
            }
 
            this.queue?.Dispose();
            this.queue = null;
        }
 
        public void CloseForTermination()
        {
            _ = this.CloseForTermination(false, 0);
        }
 
        public Task CloseForTermination(bool WaitForCompletion)
        {
            return this.CloseForTermination(WaitForCompletion, int.MaxValue);
        }
 
        public async Task CloseForTermination(bool WaitForCompletion, int Timeout)
        {
            try
            {
                this.terminating = true;
 
                for (int i = 0; i < this.nrProcessors; i++)
                    this.cancelWaitSources[i]?.Cancel();
 
                if (!this.terminated && Timeout > 0)
                {
                    if (Timeout > 0 && Timeout < int.MaxValue)
                    {
                        _ = Task.Delay(Timeout).ContinueWith((_) =>
                        {
                            for (int i = 0; i < this.nrProcessors; i++)
                                this.cancelWorkSources[i]?.Cancel();
                        });
                    }
 
                    if (WaitForCompletion)
                        await Task.WhenAll(this.processors);
 
                    this.terminated = true;
                }
            }
            catch (Exception ex)
            {
                Log.Exception(ex);
            }
        }
 
        public void Queue(T Work)
        {
            if (!this.terminating)
            {
                this.idle = false;
                this.queue?.Queue(Work);
            }
        }
 
        public Task<bool> Forward(T Work)
        {
            if (!this.terminating)
            {
                this.idle = false;
                return this.queue?.Forward(Work) ?? Task.FromResult(false);
            }
            else
                return Task.FromResult(false);
        }
 
        public string Name => this.name;
 
        public bool Terminating => this.terminating;
 
        public bool Terminated => this.terminated;
 
        public int QueueSize => this.queue?.CountItems ?? 0;
 
        public bool Idle => this.idle;
 
        private async Task PerformWork(int ProcessorIndex)
        {
            try
            {
                ProcessorEventArgs e = new ProcessorEventArgs(ProcessorIndex);
                T Item;
                Task<T> Task;
 
                while (true)
                {
                    if (this.terminating)
                    {
                        if (!(this.queue?.TryGetItem(out Item) ?? false))
                            break;
                    }
                    else
                    {
                        using (CancellationTokenSource Cancel = new CancellationTokenSource())
                        {
                            this.cancelWaitSources[ProcessorIndex] = Cancel;
                            try
                            {
                                Task = this.queue?.Wait(Cancel.Token);
                                if (Task is null)
                                    break;
 
                                if (!Task.IsCompleted && this.queue.CountSubscribers == this.nrProcessors)
                                {
                                    this.idle = true;
                                    await this.OnIdle.Raise(this, e);
                                }
 
                                Item = await Task;
                                if (Item is null)
                                    break;
                            }
                            finally
                            {
                                this.cancelWaitSources[ProcessorIndex] = null;
                            }
                        }
                    }
 
                    this.idle = false;
 
                    using (CancellationTokenSource Cancel = new CancellationTokenSource())
                    {
                        this.cancelWorkSources[ProcessorIndex] = Cancel;
                        this.currentWorkItem[ProcessorIndex] = Item;
                        try
                        {
                            await Item.Execute(Cancel.Token);
                            Item.Processed(!Cancel.IsCancellationRequested);
                        }
                        catch (Exception ex)
                        {
                            Item.Processed(false);
                            Log.Exception(ex);
                        }
                        finally
                        {
                            this.cancelWorkSources[ProcessorIndex] = null;
                            this.currentWorkItem[ProcessorIndex] = default;
                        }
                    }
                }
            }
            catch (Exception ex)
            {
                Log.Exception(ex);
            }
 
            if (--this.nrProcessorsRunning == 0)
                this.terminated = true;
        }
 
        public event EventHandlerAsync<ProcessorEventArgs> OnIdle;
 
        public async Task WaitUntilIdle()
        {
            TaskCompletionSource<bool> Idle = new TaskCompletionSource<bool>();
 
            Task OnIdle(object Sender, EventArgs e)
            {
                Task.Run(() => Idle.TrySetResult(true));
                return Task.CompletedTask;
            }
            ;
 
            this.OnIdle += OnIdle;
            try
            {
                if (this.idle)
                    return;
 
                await Idle.Task;
            }
            finally
            {
                this.OnIdle -= OnIdle;
            }
        }
    }
}