using System.Collections.Concurrent; using qdr.fnd.core.pqueue.Configuration; using qdr.fnd.core.pqueue.Enums; using qdr.fnd.core.pqueue.Exceptions; using qdr.fnd.core.pqueue.Extensions; using qdr.fnd.core.pqueue.Process; using qdr.fnd.core.pqueue.Storages; using Quadarax.Foundation.Core.Data; using Quadarax.Foundation.Core.Logging; using Quadarax.Foundation.Core.Thread; using Quadarax.Foundation.Core.Value.Extensions; using Quadarax.Foundation.Core.Value.Generators; namespace qdr.fnd.core.pqueue { public class PsQueue : LoopWorker, IPsQueue { #region *** Constants *** private const string CS_DEF_QUEUE_NAME = "PSQUEUE"; public const int CN_PRIO_SYSTEM = 0; public const int CN_PRIO_IMMEDIATE = 2; public const int CN_PRIO_CUSTOM = 5; #endregion #region *** Properties *** public PsQueueConfiguration Configuration { get;} public event PsQueueStatusChangedEventHandler? ProcessItemStatusChanged; #endregion #region *** Private Fields *** private TinyHash _hashGenerator; protected IPsQueueStorage _storage; private List> _statusTransitions = new List>(); private ConcurrentDictionary _workingItems = new ConcurrentDictionary(); #endregion #region *** Constructors *** public PsQueue(string queueName, PsQueueConfiguration configuration,IPsQueueStorage storageProvider, ILogger logger) : base(queueName,null) { _hashGenerator = new TinyHash(5); if (string.IsNullOrEmpty(queueName)) queueName =CS_DEF_QUEUE_NAME + _hashGenerator.NewId(); Configuration = configuration ?? throw new ArgumentNullException(nameof(configuration)); _storage = storageProvider ?? throw new ArgumentNullException(nameof(storageProvider)); _statusTransitions.Add(new Tuple(PsStatusEnum.New,new[] { PsStatusEnum.Pending })); _statusTransitions.Add(new Tuple(PsStatusEnum.Pending,new[] { PsStatusEnum.Started, PsStatusEnum.Expired })); _statusTransitions.Add(new Tuple(PsStatusEnum.Started,new[] { PsStatusEnum.Pending, PsStatusEnum.Paused, PsStatusEnum.Stopped, PsStatusEnum.DoneOk, PsStatusEnum.DoneFailed, PsStatusEnum.Expired })); _statusTransitions.Add(new Tuple(PsStatusEnum.Paused,new[] { PsStatusEnum.Started, PsStatusEnum.Stopped, PsStatusEnum.Expired })); _statusTransitions.Add(new Tuple(PsStatusEnum.Stopped,new[] { PsStatusEnum.DoneFailed, PsStatusEnum.Expired })); _statusTransitions.Add(new Tuple(PsStatusEnum.DoneOk,new[] { PsStatusEnum.Deleted })); _statusTransitions.Add(new Tuple(PsStatusEnum.DoneFailed,new[] { PsStatusEnum.Deleted })); _statusTransitions.Add(new Tuple(PsStatusEnum.Expired,new[] { PsStatusEnum.Deleted })); _statusTransitions.Add(new Tuple(PsStatusEnum.Deleted,Array.Empty())); } #endregion #region *** Public Operations *** public IQueryable Query(string? owner, string? ownerRef, params Tuple[] status) { throw new NotImplementedException(); } /// public string EnQueueItem(IPsQueueItemDescriptor item) { ValidateItem(item); Log(LogSeverityEnum.Debug,$"{Name} - Item {item} begins to enqueue."); var itemId = _storage.CreateItem(item.Owner, item.OwnerRef, item.ProcessProviderClass, item.AtteptsInitial, null); foreach (var argument in item.Arguments) { if (argument.ValueType.FullName == null) throw new InvalidOperationException($"Cannot obtain fullName of type '{argument.ValueType.Name}'"); _storage.CreateItemAttribute(itemId, argument.Name, argument.Value, argument.ValueType.FullName, argument.IsOutput); Log(LogSeverityEnum.Debug,$"{Name} - Attribute {argument.Name} was append to item [{itemId}]"); } var newItem = GetItemInternal(itemId); SetItemStatusInternal(newItem, PsStatusEnum.Pending, true, "Item was enqueued."); Log(LogSeverityEnum.Info,$"{Name} - Item [{itemId}] was enqueued as {newItem}"); return itemId; } /// public IPsQueueItem GetQueueItem(string itemId) { ValidateItemId(itemId); Log(LogSeverityEnum.Debug,$"{Name} - Item [{itemId}] begins to get."); var item = GetItemInternal(itemId); Log(LogSeverityEnum.Info,$"{Name} - Item [{itemId}] was fetched as {item}"); return item; } /// public void ForceStartQueueItem(string itemId, string emitter) { ValidateItemId(itemId); ValidateEmitter(emitter); Log(LogSeverityEnum.Debug, $"{Name} - Item [{itemId}] begins to schedule for force start by '{emitter}'."); var item = _storage.GetItem(itemId); EnsureItemStateExpired(item); if (item.Status.In(PsStatusEnum.Started, PsStatusEnum.Stopped, PsStatusEnum.Expired)) throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InvalidStatusForForceStart, item.Status); _storage.UpdateItemValidity(itemId, DateTime.Now, item.ValidTo == null ? null : DateTime.Now.Add(item.ValidTo.Value - item.ValidFrom)); _storage.UpdateItemPriority(itemId, CN_PRIO_IMMEDIATE); SetItemStatusInternal(item, PsStatusEnum.Pending, true, $"Item will reset due force start by emitter '{emitter}'."); Log(LogSeverityEnum.Info, $"{Name} - Item [{itemId}] was scheduled for force start by {emitter}."); } /// public void ForceStopQueueItem(string itemId, string emitter) { ValidateItemId(itemId); ValidateEmitter(emitter); Log(LogSeverityEnum.Debug, $"{Name} - Item [{itemId}] begins to schedule for stop start by '{emitter}'."); var item = _storage.GetItem(itemId); EnsureItemStateExpired(item); if (item.Status.In(PsStatusEnum.Expired, PsStatusEnum.DoneOk, PsStatusEnum.DoneFailed)) throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InvalidStatusForForceStop, item.Status); _storage.UpdateItemValidity(itemId, DateTime.Now, item.ValidTo == null ? null : DateTime.Now.Add(item.ValidTo.Value - item.ValidFrom)); _storage.UpdateItemPriority(itemId, CN_PRIO_IMMEDIATE); SetItemStatusInternal(item, PsStatusEnum.Stopped, true, $"Item will reset due force stop by emitter '{emitter}'."); Log(LogSeverityEnum.Info, $"{Name} - Item [{itemId}] was scheduled for force stop by {emitter}."); } /// public void PauseQueueItem(string itemId, string emitter) { ValidateItemId(itemId); ValidateEmitter(emitter); Log(LogSeverityEnum.Debug, $"{Name} - Item [{itemId}] begins to schedule for pause by '{emitter}'."); var item = _storage.GetItem(itemId); EnsureItemStateExpired(item); if (item.Status.In(PsStatusEnum.Expired, PsStatusEnum.DoneOk, PsStatusEnum.DoneFailed)) throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InvalidStatusForPause, item.Status); _storage.UpdateItemValidity(itemId, DateTime.Now, item.ValidTo == null ? null : DateTime.Now.Add(item.ValidTo.Value - item.ValidFrom)); _storage.UpdateItemPriority(itemId, CN_PRIO_IMMEDIATE); SetItemStatusInternal(item, PsStatusEnum.Paused, true, $"Item will reset due pause by emitter '{emitter}'."); Log(LogSeverityEnum.Info, $"{Name} - Item [{itemId}] was scheduled for pause by {emitter}."); } public void StartQueueItem(string itemId, string emmiter, bool isImmediate) { throw new NotImplementedException(); } /// public IQueryable QueryQueueItems(IPaging paging, string? owner, string? ownerRef, bool includeTransient = false, params PsStatusEnum[] statuses) { if (paging == null) throw new ArgumentNullException(nameof(paging)); var statusesForQuery = new List>(); foreach (var status in statuses) { statusesForQuery.Add(new Tuple(false, status)); if (includeTransient) statusesForQuery.Add(new Tuple(true, status)); } return _storage.Query(owner, ownerRef, statusesForQuery.ToArray()); } public void Start() { base.Start(); } public void Stop() { base.Stop(); } #endregion #region *** Private Operations *** #region Overrides of LoopWorker /// protected override Task DoIteration(IWorkerContext? context) { // fetch items to immediate stop - frees workers var itemsToStop = Query(null, null, new Tuple[] { new(true, PsStatusEnum.Stopped),new(true, PsStatusEnum.Paused) }) .OrderBy(x=>x.Priority).ThenBy(x=>x.ValidFrom); foreach(var item in itemsToStop){ var process = _workingItems.TryGetValue(item.Id, out var processProvider) ? processProvider : null; if (process == null) Log(LogSeverityEnum.Warn, $"{Name} - Cannot find process provider in working cache for item [{item.Id}]. Skiped."); if (process != null) { process.Stop(); SetItemStatusInternal(item, item.Status, false, "Item was stopped by schedule."); _workingItems.TryRemove(item.Id, out _); } } // fetch working, stopped, paused items which currently expired - frees workers var itemsToExpire = Query(null, null, new Tuple[] { new(true, PsStatusEnum.Stopped),new(true, PsStatusEnum.Paused), new(true, PsStatusEnum.Started) }) .Where(x=>x.IsValiditySpanExpired()); foreach(var item in itemsToExpire) { EnsureItemStateExpired(item); } // fetch items to immediate start var workersLeft = Configuration.MaxConcurrentProcesses - _workingItems.Count; if (workersLeft>0) { // fetch var itemsToProcess = Query(null, null, new Tuple[] { new(true, PsStatusEnum.Started) }) .Where(x => !x.IsValiditySpanExpired()).Take(workersLeft); foreach(var item in itemsToProcess) { if (item.AttemptsLeft <= 0) { SetItemStatusInternal(item, PsStatusEnum.DoneFailed, true, "Item will be marked as failed due no attempts left."); SetItemStatusInternal(item, PsStatusEnum.DoneFailed, false, "Item will be marked as failed due no attempts left."); continue; } _storage.UpdateItemAttempts(item.Id, item.AttemptsLeft - 1); var provider = ResolveProvider(item.ProcessProviderClass, item.Arguments); _storage.UpdateItemStarted(item.Id, DateTime.Now); _storage.UpdateItemFinished(item.Id, null); _workingItems.TryAdd(item.Id, provider); provider.Start(); SetItemStatusInternal(item, PsStatusEnum.Started, false, "Item was started by schedule."); } } // fetch items to close expired var itemsToClose = Query(null, null, new Tuple[] { new(true, PsStatusEnum.Expired) }); foreach (var item in itemsToClose) { SetItemStatusInternal(item, PsStatusEnum.Expired, false, "Item will be expired by schedule."); } // fetch pending items to initiate to expired var itemsPendingToExpire = Query(null, null, new Tuple[] { new(true, PsStatusEnum.Pending), new(false, PsStatusEnum.Pending) }); foreach(var item in itemsPendingToExpire) { SetItemStatusInternal(item, PsStatusEnum.Expired, true, "Item will be expired by schedule."); } return base.DoIteration(context); } #endregion private void ValidateItemId(string itemId) { if (string.IsNullOrEmpty(itemId)) throw new ArgumentNullException(nameof(itemId)); } private void ValidateEmitter(string emitter) { if (string.IsNullOrEmpty(emitter)) throw new ArgumentNullException(nameof(emitter)); } private void ValidateItem(IPsQueueItemDescriptor item) { if (item is null) throw new ArgumentNullException(nameof(item)); if (string.IsNullOrEmpty(item.Owner)) throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.OwnerRequired); if (string.IsNullOrEmpty(item.OwnerRef)) throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.OwnerRefRequired); var provider = ResolveProvider(item.ProcessProviderClass, item.Arguments); provider.Validate(); if (item.AtteptsInitial <= 0) throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InitialAttemptMustBePositive); } private void SetItemStatusInternal(IPsQueueItem item, PsStatusEnum? newStatus, bool isTransient, string message) { var oldStatus = item.Status; var oldIsTransient = item.IsStatusTransient; if (!isTransient && newStatus == null) throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InvalidStatusTransientCombination); if (newStatus != null) { if (!_statusTransitions.First(x => x.Item1 == oldStatus).Item2.Contains(newStatus.Value)) throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.InvalidStatusTransition, oldStatus,newStatus ?? oldStatus, string.Join(",", _statusTransitions.First(x => x.Item1 == oldStatus).Item2.Select(x=>x.ToString()))); } _storage.AppendStatusJournal(item.Id, newStatus ?? oldStatus, isTransient, message); _storage.UpdateItemStatus(item.Id, newStatus, isTransient); ((PsQueueItem)item).SetStatus(newStatus, isTransient, message); ProcessItemStatusChanged?.Invoke(this, new PsQueueStatusChangedEventArgs(item.Id) { CurrentStatus = newStatus ?? oldStatus, CurrentIsStatusTransient = isTransient, PreviousStatus = oldStatus, PreviousIsStatusTransient = oldIsTransient }); Log(LogSeverityEnum.Debug, $"{Name} - Item [{item.Id}] changed status '{oldStatus}[{oldIsTransient}]' to '{(newStatus ?? oldStatus)}[{isTransient}]'."); } private IPsProcessProvider ResolveProvider(string providerClass, IEnumerable arguments) { if (string.IsNullOrEmpty(providerClass)) throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.ProcessProviderRequired); var providerType = Type.GetType(providerClass); if (providerType == null) throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.ProcessProviderCannotResolve, providerClass); var provider = Activator.CreateInstance(providerType, arguments) as IPsProcessProvider; if (provider == null) throw new PSQueueValidationException(PSQueueValidationException.ErrorCodes.ProcessProviderCannotResolve, providerClass); return provider; } private void EnsureItemStateExpired(IPsQueueItem item) { if (item==null) throw new ArgumentNullException(nameof(item)); if (item.IsValiditySpanExpired() && item.Status != PsStatusEnum.Expired) { if (item.Status == PsStatusEnum.Started) { if (!_workingItems.TryGetValue(item.Id, out var process)) throw new InvalidOperationException($"Cannot find process provider in working cache for item [{item.Id}]"); SetItemStatusInternal(item, PsStatusEnum.Stopped, true, "Item will be stopped because expired."); SetItemStatusInternal(item, PsStatusEnum.Stopped, false, "Item was stopped by system because expired."); // NOTE: fast switch to stopped due prevents main loop to handle - stops later _workingItems.TryRemove(item.Id, out _); process.Stop(); SetItemStatusInternal(item, PsStatusEnum.Expired, true, "Item will be expired by process."); } else { SetItemStatusInternal(item, PsStatusEnum.Expired, true, "Item will be expired by process."); } item = GetItemInternal(item.Id); } } private IPsQueueItem GetItemInternal(string itemId) { ValidateItemId(itemId); var item = _storage.GetItem(itemId); return item; } protected override void OnDisposing() { base.OnDisposing(); ProcessItemStatusChanged = null; _storage.Dispose(); //TODO: Implement OnDisposing foreach (var item in _workingItems.Values) item.Stop(); _workingItems.Clear(); _workingItems = null!; } #endregion } }