<commit_before>using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Quidjibo.Configurations;
using Quidjibo.Factories;
using Quidjibo.Misc;
using Quidjibo.Models;
using Quidjibo.Pipeline;
using Quidjibo.Pipeline.Contexts;
using Quidjibo.Pipeline.Misc;
using Quidjibo.Providers;

namespace Quidjibo.Servers
{
    public class QuidjiboServer : IQuidjiboServer
    {
        private readonly ICronProvider _cronProvider;
        private readonly ILogger _logger;
        private readonly IProgressProviderFactory _progressProviderFactory;
        private readonly IQuidjiboConfiguration _quidjiboConfiguration;
        private readonly IQuidjiboPipeline _quidjiboPipeline;
        private readonly IScheduleProviderFactory _scheduleProviderFactory;
        private readonly IWorkProviderFactory _workProviderFactory;

        public bool IsRunning { get; private set; }

        public QuidjiboServer(
            ILoggerFactory loggerFactory,
            IQuidjiboConfiguration quidjiboConfiguration,
            IWorkProviderFactory workProviderFactory,
            IScheduleProviderFactory scheduleProviderFactory,
            IProgressProviderFactory progressProviderFactory,
            ICronProvider cronProvider,
            IQuidjiboPipeline quidjiboPipeline)
        {
            _logger = loggerFactory.CreateLogger<QuidjiboServer>();
            _workProviderFactory = workProviderFactory;
            _scheduleProviderFactory = scheduleProviderFactory;
            _quidjiboConfiguration = quidjiboConfiguration;
            _cronProvider = cronProvider;
            _quidjiboPipeline = quidjiboPipeline;
            _progressProviderFactory = progressProviderFactory;
            Worker = $"{Environment.GetEnvironmentVariable("COMPUTERNAME")}-{Guid.NewGuid()}";
        }

        public string Worker { get; }

        public void Start()
        {
            lock (_syncRoot)
            {
                if (IsRunning)
                {
                    return;
                }

                _logger.LogInformation("Starting Server {Worker}", Worker);
                _cts = new CancellationTokenSource();
                _loopTasks = new List<Task>();

                _throttle = new SemaphoreSlim(0, _quidjiboConfiguration.Throttle);
                _logger.LogInformation("EnableWorker = {EnableWorker}", _quidjiboConfiguration.EnableWorker);
                if (_quidjiboConfiguration.EnableWorker)
                {
                    if (_quidjiboConfiguration.SingleLoop)
                    {
                        _logger.LogInformation("All queues can share the same loop");
                        var queues = string.Join(",", _quidjiboConfiguration.Queues);
                        _loopTasks.Add(WorkLoopAsync(queues));
                    }
                    else
                    {
                        _logger.LogInformation("Each queue will need a designated loop");
                        _loopTasks.AddRange(_quidjiboConfiguration.Queues.Select(WorkLoopAsync));
                    }
                }

                _logger.LogInformation("EnableScheduler = {EnableScheduler}", _quidjiboConfiguration.EnableScheduler);
                if (_quidjiboConfiguration.EnableScheduler)
                {
                    _logger.LogInformation("Enabling scheduler");
                    _loopTasks.Add(ScheduleLoopAsync(_quidjiboConfiguration.Queues));
                }

                _throttle.Release(_quidjiboConfiguration.Throttle);
                IsRunning = true;
                _logger.LogInformation("Started Worker {Worker}", Worker);
            }
        }

        public void Stop()
        {
            lock (_syncRoot)
            {
                if (!IsRunning)
                {
                    return;
                }

                _cts?.Cancel();
                _cts?.Dispose();
                _loopTasks = null;
                IsRunning = false;
                _logger.LogInformation("Stopped Worker {Worker}", Worker);
            }
        }

        public void Dispose()
        {
            Stop();
        }

        #region Internals

        private readonly object _syncRoot = new object();

        private List<Task> _loopTasks;
        private CancellationTokenSource _cts;
        private SemaphoreSlim _throttle;

        private async Task WorkLoopAsync(string queue)
        {
            var pollingInterval = TimeSpan.FromSeconds(_quidjiboConfiguration.WorkPollingInterval ?? 45);
            var workProvider = await _workProviderFactory.CreateAsync(queue, _cts.Token);
            while (!_cts.IsCancellationRequested)
            {
                var items = new List<WorkItem>(0);
                try
                {
                    _logger.LogTrace("Throttle Count : {ThrottleCount}", _throttle.CurrentCount);

                    // throttle is important when there is more than one listener
                    await _throttle.WaitAsync(_cts.Token);
                    items = await workProvider.ReceiveAsync(Worker, _cts.Token);
                    if (items.Count > 0)
                    {
                        _logger.LogDebug("Received {WorkItemCount} items.", items.Count);
                        var tasks = items.Select(item => InvokePipelineAsync(workProvider, item));
                        await Task.WhenAll(tasks);
                    }
                }
                catch (Exception exception)
                {
                    _logger.LogWarning(0, exception, exception.Message);
                }
                finally
                {
                    _throttle.Release();
                }

                if (items.Count <= 0)
                {
                    await Task.Delay(pollingInterval, _cts.Token);
                }
            }
        }

        private async Task ScheduleLoopAsync(string[] queues)
        {
            var pollingInterval = TimeSpan.FromSeconds(_quidjiboConfiguration.SchedulePollingInterval ?? 45);
            var scheduleProvider = await _scheduleProviderFactory.CreateAsync(queues, _cts.Token);
            while (!_cts.IsCancellationRequested)
            {
                try
                {
                    var items = await scheduleProvider.ReceiveAsync(_cts.Token);
                    foreach (var item in items)
                    {<commit_msg>Fixed issue that causes the Quidjibo to stop when it fails to retrieve work items.<commit_after>using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using Quidjibo.Configurations;
using Quidjibo.Factories;
using Quidjibo.Misc;
using Quidjibo.Models;
using Quidjibo.Pipeline;
using Quidjibo.Pipeline.Contexts;
using Quidjibo.Pipeline.Misc;
using Quidjibo.Providers;

namespace Quidjibo.Servers
{
    public class QuidjiboServer : IQuidjiboServer
    {
        private readonly ICronProvider _cronProvider;
        private readonly ILogger _logger;
        private readonly IProgressProviderFactory _progressProviderFactory;
        private readonly IQuidjiboConfiguration _quidjiboConfiguration;
        private readonly IQuidjiboPipeline _quidjiboPipeline;
        private readonly IScheduleProviderFactory _scheduleProviderFactory;
        private readonly IWorkProviderFactory _workProviderFactory;

        public bool IsRunning { get; private set; }

        public QuidjiboServer(
            ILoggerFactory loggerFactory,
            IQuidjiboConfiguration quidjiboConfiguration,
            IWorkProviderFactory workProviderFactory,
            IScheduleProviderFactory scheduleProviderFactory,
            IProgressProviderFactory progressProviderFactory,
            ICronProvider cronProvider,
            IQuidjiboPipeline quidjiboPipeline)
        {
            _logger = loggerFactory.CreateLogger<QuidjiboServer>();
            _workProviderFactory = workProviderFactory;
            _scheduleProviderFactory = scheduleProviderFactory;
            _quidjiboConfiguration = quidjiboConfiguration;
            _cronProvider = cronProvider;
            _quidjiboPipeline = quidjiboPipeline;
            _progressProviderFactory = progressProviderFactory;
            Worker = $"{Environment.GetEnvironmentVariable("COMPUTERNAME")}-{Guid.NewGuid()}";
        }

        public string Worker { get; }

        public void Start()
        {
            lock (_syncRoot)
            {
                if (IsRunning)
                {
                    return;
                }

                _logger.LogInformation("Starting Server {Worker}", Worker);
                _cts = new CancellationTokenSource();
                _loopTasks = new List<Task>();

                _throttle = new SemaphoreSlim(0, _quidjiboConfiguration.Throttle);
                _logger.LogInformation("EnableWorker = {EnableWorker}", _quidjiboConfiguration.EnableWorker);
                if (_quidjiboConfiguration.EnableWorker)
                {
                    if (_quidjiboConfiguration.SingleLoop)
                    {
                        _logger.LogInformation("All queues can share the same loop");
                        var queues = string.Join(",", _quidjiboConfiguration.Queues);
                        _loopTasks.Add(WorkLoopAsync(queues));
                    }
                    else
                    {
                        _logger.LogInformation("Each queue will need a designated loop");
                        _loopTasks.AddRange(_quidjiboConfiguration.Queues.Select(WorkLoopAsync));
                    }
                }

                _logger.LogInformation("EnableScheduler = {EnableScheduler}", _quidjiboConfiguration.EnableScheduler);
                if (_quidjiboConfiguration.EnableScheduler)
                {
                    _logger.LogInformation("Enabling scheduler");
                    _loopTasks.Add(ScheduleLoopAsync(_quidjiboConfiguration.Queues));
                }

                _throttle.Release(_quidjiboConfiguration.Throttle);
                IsRunning = true;
                _logger.LogInformation("Started Worker {Worker}", Worker);
            }
        }

        public void Stop()
        {
            lock (_syncRoot)
            {
                if (!IsRunning)
                {
                    return;
                }

                _cts?.Cancel();
                _cts?.Dispose();
                _loopTasks = null;
                IsRunning = false;
                _logger.LogInformation("Stopped Worker {Worker}", Worker);
            }
        }

        public void Dispose()
        {
            Stop();
        }

        #region Internals

        private readonly object _syncRoot = new object();

        private List<Task> _loopTasks;
        private CancellationTokenSource _cts;
        private SemaphoreSlim _throttle;

        private async Task WorkLoopAsync(string queue)
        {
            var pollingInterval = TimeSpan.FromSeconds(_quidjiboConfiguration.WorkPollingInterval ?? 45);
            var workProvider = await _workProviderFactory.CreateAsync(queue, _cts.Token);
            while (!_cts.IsCancellationRequested)
            {
                var items = new List<WorkItem>(0);
                try
                {
                    _logger.LogTrace("Throttle Count : {ThrottleCount}", _throttle.CurrentCount);

                    // throttle is important when there is more than one listener
                    await _throttle.WaitAsync(_cts.Token);
                    items = await workProvider.ReceiveAsync(Worker, _cts.Token);
                    if (items.Count > 0)
                    {
                        _logger.LogDebug("Received {WorkItemCount} items.", items.Count);
                        var tasks = items.Select(item => InvokePipelineAsync(workProvider, item));
                        await Task.WhenAll(tasks);
                    }
                }
                catch (Exception exception)
                {
                    _logger.LogWarning(0, exception, exception.Message);
                }
                finally
                {
                    _throttle.Release();
                }

                if (items == null || items.Count <= 0)
                {
                    await Task.Delay(pollingInterval, _cts.Token);
                }
            }
        }

        private async Task ScheduleLoopAsync(string[] queues)
        {
            var pollingInterval = TimeSpan.FromSeconds(_quidjiboConfiguration.SchedulePollingInterval ?? 45);
            var scheduleProvider = await _scheduleProviderFactory.CreateAsync(queues, _cts.Token);
            while (!_cts.IsCancellationRequested)
            {
                try
                {
                    var items = await scheduleProvider.ReceiveAsync(_cts.Token);
                    foreach (var item in items)
                    {
{'fraction_non_alphanumeric': 0.06935928877988964, 'fraction_numerical': 0.0013795217657878603, 'mean_word_length': 1.6849794238683127, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1674134', 'n_tokens_mistral': 3301, 'n_tokens_neox': 3047, 'n_words': 622}