<commit_before>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Signum.Engine.Authorization;
using Signum.Engine.Basics;
using Signum.Engine.Cache;
using Signum.Engine.Operations;
using Signum.Entities;
using Signum.Entities.Authorization;
using Signum.Entities.Processes;
using Signum.Utilities;
using Signum.Utilities.ExpressionTrees;
using System.Data.SqlClient;

namespace Signum.Engine.Processes
{
    public static class ProcessRunnerLogic
    {
        static Dictionary<Lite<ProcessDN>, ExecutingProcess> executing = new Dictionary<Lite<ProcessDN>, ExecutingProcess>();

        static Timer timer;

        internal static DateTime? nextPlannedExecution;

        public static int MaxDegreeOfParallelism = 2;

        static bool running = false;

        static int initialDelayMiliseconds;

        static CancellationTokenSource CancelNewProcesses;

        static AutoResetEvent autoResetEvent = new AutoResetEvent(false);

        public static ProcessLogicState ExecutionState()
        {
            return new ProcessLogicState
            {
                Running = running,
                InitialDelayMiliseconds = initialDelayMiliseconds,
                MaxDegreeOfParallelism = MaxDegreeOfParallelism,
                NextPlannedExecution = nextPlannedExecution,
                JustMyProcesses = ProcessLogic.JustMyProcesses,
                Executing = executing.Values.Select(p => new ExecutionState
                {
                    IsCancellationRequested = p.CancelationSource.IsCancellationRequested,
                    Process = p.CurrentExecution.ToLite(),
                    State = p.CurrentExecution.State,
                    Progress = p.CurrentExecution.Progress,
                    MachineName = p.CurrentExecution.MachineName,
                }).ToList()
            };
        }

        public static void StartRunningProcesses(int delayMilliseconds)
        {
            initialDelayMiliseconds = delayMilliseconds;

            if (initialDelayMiliseconds == 0)
                StartRunningProcesses();

            Task.Factory.StartNew(() =>
            {
                Thread.Sleep(initialDelayMiliseconds);
                StartRunningProcesses();
            });
        }

        static int SetAsQueued(this IQueryable<ProcessDN> query)
        {
            return query.UnsafeUpdate(p => new ProcessDN
            {
                State = ProcessState.Queued,
                QueuedDate = TimeZoneManager.Now,
                ExecutionStart = null,
                ExecutionEnd = null,
                SuspendDate = null,
                Progress = null,
                Exception = null,
                ExceptionDate = null,
                MachineName = ProcessLogic.JustMyProcesses ? Environment.MachineName : ProcessDN.None
            });
        }

        internal static void SetAsQueued(this ProcessDN process)
        {
            process.State = ProcessState.Queued;
            process.QueuedDate = TimeZoneManager.Now;
            process.ExecutionStart = null;
            process.ExecutionEnd = null;
            process.SuspendDate = null;
            process.Progress = null;
            process.Exception = null;
            process.ExceptionDate = null;
            process.MachineName = ProcessLogic.JustMyProcesses ? Environment.MachineName : ProcessDN.None;
        }

        public static void StartRunningProcesses()
        {
            if (running)
                throw new InvalidOperationException("ProcessLogic is running");


            Task.Factory.StartNew(() =>
            {
                using (AuthLogic.Disable())
                {
                    try
                    {
                        running = true;

                        (from p in Database.Query<ProcessDN>()
                         where
                         p.State == ProcessState.Executing && p.MachineName == Environment.MachineName ||
                         p.State == ProcessState.Suspending && p.MachineName == Environment.MachineName ||
                         p.State == ProcessState.Suspended && (p.MachineName == Environment.MachineName || !ProcessLogic.JustMyProcesses && p.MachineName == ProcessDN.None)
                         select p).SetAsQueued();

                        CancelNewProcesses = new CancellationTokenSource();
                        autoResetEvent.Set();
                        timer = new Timer(ob => WakeUp("Timer", null), // main timer
                             null,
                             Timeout.Infinite,
                             Timeout.Infinite);

                        while (autoResetEvent.WaitOne())
                        {
                            if (CancelNewProcesses.IsCancellationRequested)
                                return;

                            using (HeavyProfiler.Log("PWL", ()=> "Process Runner"))
                            {
                                (from p in Database.Query<ProcessDN>()
                                 where p.State == ProcessState.Planned && p.PlannedDate <= TimeZoneManager.Now
                                 orderby p.PlannedDate
                                 select p).SetAsQueued();

                                var list = Database.Query<ProcessDN>()
                                        .Where(a => ProcessLogic.JustMyProcesses ? a.MachineName == Environment.MachineName : a.MachineName == ProcessDN.None)
                                        .Where(a => a.State == ProcessState.Planned)
                                        .Select(a => a.PlannedDate)
                                        .ToListWithInvalidation("Process", args => WakeUp("Planned dependency", args));

                                SetNextPannedExecution(list.Min());

                                lock (executing)
                                {
                                    int remaining = MaxDegreeOfParallelism - executing.Count;

                                    if (remaining > 0)
                                    {
                                        var afordable = new Transaction().Using(tr =>
                                        {
                                            while (true)
                                            {
                                                var queued = Database.Query<ProcessDN>()
                                                    .Where(p => p.State == ProcessState.Queued)
                                                    .Where(p => p.MachineName == Environment.MachineName || (!ProcessLogic.JustMyProcesses && p.MachineName == ProcessDN.None))
                                                    .Select(a => new { Process = a.ToLite(), a.QueuedDate, a.MachineName })
                                                    .ToListWithInvalidation("", args => WakeUp("Queued dependency", args));

                                                var result = queued
                                                    .OrderByDescending(p => p.MachineName == Environment.MachineName)
                                                    .OrderBy(a => a.QueuedDate)
                                                    .Take(remaining).ToList();

                                                var taken = result.Where(p => p.MachineName == ProcessDN.None).Select(a => a.Process).ToList();

                                                if (taken.IsNullOrEmpty())
                                                    return tr.Commit(result);

                                                Database.Query<ProcessDN>()
                                                    .Where(p => taken.Contains(p.ToLite()) && p.MachineName == ProcessDN.None)
                                                    .UnsafeUpdate(p => new ProcessDN { MachineName = Environment.MachineName });
                                            }
                                        });

                                        foreach (var pair in afordable)
                                        {
                                            ProcessDN pro = pair.Process.Retrieve();

                                            IProcessAlgorithm algorithm = ProcessLogic.GetProcessAlgorithm(MultiEnumLogic<ProcessAlgorithmDN>.ToEnum(pro.Algorithm));

                                            ExecutingProcess executingProcess = new ExecutingProcess(algorithm, pro);

                                            executing.Add(pro.ToLite(), executingProcess);

                                            executingProcess.TakeForThisMachine();

                                            Task.Factory.StartNew(() =>
                                            {
                                                try
                                                {
                                                    executingProcess.Execute();
                                                }
                                                finally
                                                {
                                                    lock (executing)
                                                    {
                                                        executing.Remove(pro.ToLite());
                                                        WakeUp("Process ended", null);
                                                    }
                                                }
                                            });<commit_msg>refix ProcessRunnerLogic<commit_after>﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Signum.Engine.Authorization;
using Signum.Engine.Basics;
using Signum.Engine.Cache;
using Signum.Engine.Operations;
using Signum.Entities;
using Signum.Entities.Authorization;
using Signum.Entities.Processes;
using Signum.Utilities;
using Signum.Utilities.ExpressionTrees;
using System.Data.SqlClient;

namespace Signum.Engine.Processes
{
    public static class ProcessRunnerLogic
    {
        static Dictionary<Lite<ProcessDN>, ExecutingProcess> executing = new Dictionary<Lite<ProcessDN>, ExecutingProcess>();

        static Timer timer;

        internal static DateTime? nextPlannedExecution;

        public static int MaxDegreeOfParallelism = 2;

        static bool running = false;

        static int initialDelayMiliseconds;

        static CancellationTokenSource CancelNewProcesses;

        static AutoResetEvent autoResetEvent = new AutoResetEvent(false);

        public static ProcessLogicState ExecutionState()
        {
            return new ProcessLogicState
            {
                Running = running,
                InitialDelayMiliseconds = initialDelayMiliseconds,
                MaxDegreeOfParallelism = MaxDegreeOfParallelism,
                NextPlannedExecution = nextPlannedExecution,
                JustMyProcesses = ProcessLogic.JustMyProcesses,
                Executing = executing.Values.Select(p => new ExecutionState
                {
                    IsCancellationRequested = p.CancelationSource.IsCancellationRequested,
                    Process = p.CurrentExecution.ToLite(),
                    State = p.CurrentExecution.State,
                    Progress = p.CurrentExecution.Progress,
                    MachineName = p.CurrentExecution.MachineName,
                }).ToList()
            };
        }

        public static void StartRunningProcesses(int delayMilliseconds)
        {
            initialDelayMiliseconds = delayMilliseconds;

            if (initialDelayMiliseconds == 0)
                StartRunningProcesses();

            Task.Factory.StartNew(() =>
            {
                Thread.Sleep(initialDelayMiliseconds);
                StartRunningProcesses();
            });
        }

        static int SetAsQueued(this IQueryable<ProcessDN> query)
        {
            return query.UnsafeUpdate(p => new ProcessDN
            {
                State = ProcessState.Queued,
                QueuedDate = TimeZoneManager.Now,
                ExecutionStart = null,
                ExecutionEnd = null,
                SuspendDate = null,
                Progress = null,
                Exception = null,
                ExceptionDate = null,
                MachineName = ProcessLogic.JustMyProcesses ? Environment.MachineName : ProcessDN.None
            });
        }

        internal static void SetAsQueued(this ProcessDN process)
        {
            process.State = ProcessState.Queued;
            process.QueuedDate = TimeZoneManager.Now;
            process.ExecutionStart = null;
            process.ExecutionEnd = null;
            process.SuspendDate = null;
            process.Progress = null;
            process.Exception = null;
            process.ExceptionDate = null;
            process.MachineName = ProcessLogic.JustMyProcesses ? Environment.MachineName : ProcessDN.None;
        }

        public static void StartRunningProcesses()
        {
            if (running)
                throw new InvalidOperationException("ProcessLogic is running");


            Task.Factory.StartNew(() =>
            {
                using (AuthLogic.Disable())
                {
                    try
                    {
                        running = true;

                        (from p in Database.Query<ProcessDN>()
                         where
                         p.State == ProcessState.Executing && p.MachineName == Environment.MachineName ||
                         p.State == ProcessState.Suspending && p.MachineName == Environment.MachineName ||
                         p.State == ProcessState.Suspended && (p.MachineName == Environment.MachineName || !ProcessLogic.JustMyProcesses && p.MachineName == ProcessDN.None)
                         select p).SetAsQueued();

                        CancelNewProcesses = new CancellationTokenSource();
                        autoResetEvent.Set();
                        timer = new Timer(ob => WakeUp("Timer", null), // main timer
                             null,
                             Timeout.Infinite,
                             Timeout.Infinite);

                        while (autoResetEvent.WaitOne())
                        {
                            if (CancelNewProcesses.IsCancellationRequested)
                                return;

                            using (HeavyProfiler.Log("PWL", ()=> "Process Runner"))
                            {
                                (from p in Database.Query<ProcessDN>()
                                 where p.State == ProcessState.Planned && p.PlannedDate <= TimeZoneManager.Now
                                 orderby p.PlannedDate
                                 select p).SetAsQueued();

                                var list = Database.Query<ProcessDN>()
                                        .Where(a => ProcessLogic.JustMyProcesses ? a.MachineName == Environment.MachineName : a.MachineName == ProcessDN.None)
                                        .Where(a => a.State == ProcessState.Planned)
                                        .Select(a => a.PlannedDate)
                                        .ToListWithInvalidation("Process", args => WakeUp("Planned dependency", args));

                                SetNextPannedExecution(list.Min());

                                lock (executing)
                                {
                                    int remaining = MaxDegreeOfParallelism - executing.Count;

                                    if (remaining > 0)
                                    {

                                    retry:
                                        var queued = Database.Query<ProcessDN>()
                                            .Where(p => p.State == ProcessState.Queued)
                                            .Where(p => p.MachineName == Environment.MachineName || (!ProcessLogic.JustMyProcesses && p.MachineName == ProcessDN.None))
                                            .Select(a => new { Process = a.ToLite(), a.QueuedDate, a.MachineName })
                                            .ToListWithInvalidation("", args => WakeUp("Queued dependency", args));

                                        var afordable = queued
                                            .OrderByDescending(p => p.MachineName == Environment.MachineName)
                                            .OrderBy(a => a.QueuedDate)
                                            .Take(remaining).ToList();

                                        var taken = afordable.Where(p => p.MachineName == ProcessDN.None).Select(a => a.Process).ToList();

                                        if (taken.Any())
                                        {
                                            Database.Query<ProcessDN>()
                                            .Where(p => taken.Contains(p.ToLite()) && p.MachineName == ProcessDN.None)
                                            .UnsafeUpdate(p => new ProcessDN { MachineName = Environment.MachineName });

                                            goto retry;
                                        }

                                        foreach (var pair in afordable)
                                        {
                                            ProcessDN pro = pair.Process.Retrieve();

                                            IProcessAlgorithm algorithm = ProcessLogic.GetProcessAlgorithm(MultiEnumLogic<ProcessAlgorithmDN>.ToEnum(pro.Algorithm));

                                            ExecutingProcess executingProcess = new ExecutingProcess(algorithm, pro);

                                            executing.Add(pro.ToLite(), executingProcess);

                                            executingProcess.TakeForThisMachine();

                                            Task.Factory.StartNew(() =>
                                            {
                                                try
                                                {
                                                    executingProcess.Execute();
                                                }
                                                finally
                                                {
                                                    lock (executing)
                                                    {
                                                        executing.Remove(pro.ToLite());
                                                        WakeUp("Process ended", null);
                                                    }
                                                }
                                            });
{'fraction_non_alphanumeric': 0.07282164202579428, 'fraction_numerical': 0.00031456432840515884, 'mean_word_length': 1.0572691975841242, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 70, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 181, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3105006', 'n_tokens_mistral': 3754, 'n_tokens_neox': 3585, 'n_words': 732}