<commit_before>using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Text;
using System.Threading.Tasks;

using Orleans.Runtime.Configuration;


namespace Orleans.Runtime.Scheduler
{
    [DebuggerDisplay("WorkItemGroup Name={Name} State={state}")]
    internal class WorkItemGroup : IWorkItem
    {
        private enum WorkGroupStatus
        {
            Waiting = 0,
            Runnable = 1,
            Running = 2,
            Shutdown = 3
        }

        private static readonly Logger appLogger = LogManager.GetLogger("Scheduler.WorkItemGroup", LoggerType.Runtime);
        private readonly Logger log;
        private readonly OrleansTaskScheduler masterScheduler;
        private WorkGroupStatus state;
        private readonly Object lockable;
        private readonly Queue<ActivationTask> workItems;

        private long totalItemsEnQueued;    // equals total items queued, + 1
        private long totalItemsProcessed;
        private readonly QueueTrackingStatistic queueTracking;
        private TimeSpan totalQueuingDelay;
        private readonly long quantumExpirations;
        private readonly int workItemGroupStatisticsNumber;

        internal ActivationTaskScheduler TaskRunner { get; private set; }
        internal ActivationTaskScheduler RcTaskRunner { get; private set; }

        public DateTime TimeQueued { get; set; }

        public TimeSpan TimeSinceQueued
        {
            get { return Utils.Since(TimeQueued); } 
        }

        public ISchedulingContext SchedulingContext { get; set; }

        public bool IsSystemPriority
        {
            get { return SchedulingUtils.IsSystemPriorityContext(SchedulingContext); }
        }

        internal bool IsSystemGroup
        {
            get { return SchedulingUtils.IsSystemContext(SchedulingContext); }
        }

        public string Name { get { return SchedulingContext == null ? "unknown" : SchedulingContext.Name; } }

        internal int ExternalWorkItemCount
        {
            get { lock (lockable) { return WorkItemCount; } }
        }

        private int WorkItemCount
        {
            get { return workItems.Count; } 
        }

        internal float AverageQueueLenght
        {
            get 
            {
#if TRACK_DETAILED_STATS
                if (StatisticsCollector.CollectShedulerQueuesStats)
                {
                    return queueTracking.AverageQueueLength;
                }
#endif
                return 0;
            }
        }

        internal float NumEnqueuedRequests
        {
            get
            {
#if TRACK_DETAILED_STATS
                if (StatisticsCollector.CollectShedulerQueuesStats)
                {
                    return queueTracking.NumEnqueuedRequests;
                }
#endif
                return 0;
            }
        }

        internal float ArrivalRate
        {
            get
            {
#if TRACK_DETAILED_STATS
                if (StatisticsCollector.CollectShedulerQueuesStats)
                {
                    return queueTracking.ArrivalRate;
                }
#endif
                return 0;
            }
        }

        private bool IsActive
        {
            get
            {
                return WorkItemCount != 0;
            }
        }

        // This is the maximum number of work items to be processed in an activation turn. 
        // If this is set to zero or a negative number, then the full work queue is drained (MaxTimePerTurn allowing).
        private const int MaxWorkItemsPerTurn = 0; // Unlimited
        // This is a soft time limit on the duration of activation macro-turn (a number of micro-turns). 
        // If a activation was running its micro-turns longer than this, we will give up the thread.
        // If this is set to zero or a negative number, then the full work queue is drained (MaxWorkItemsPerTurn allowing).
        public static TimeSpan ActivationSchedulingQuantum { get; set; }
        // This is the maximum number of waiting threads (blocked in WaitForResponse) allowed
        // per ActivationWorker. An attempt to wait when there are already too many threads waiting
        // will result in a TooManyWaitersException being thrown.
        //private static readonly int MaxWaitingThreads = 500;


        internal WorkItemGroup(OrleansTaskScheduler sched, ISchedulingContext schedulingContext)
        {
            masterScheduler = sched;
            SchedulingContext = schedulingContext;
            state = WorkGroupStatus.Waiting;
            workItems = new Queue<ActivationTask>();
            lockable = new Object();
            totalItemsEnQueued = 0;
            totalItemsProcessed = 0;
            totalQueuingDelay = TimeSpan.Zero;
            quantumExpirations = 0;
            TaskRunner = new ActivationTaskScheduler(this);
            RcTaskRunner = new ActivationTaskScheduler(this, true);
            log = IsSystemPriority ? LogManager.GetLogger("Scheduler." + Name + ".WorkItemGroup", LoggerType.Runtime) : appLogger;

            if (StatisticsCollector.CollectShedulerQueuesStats)
            {
                queueTracking = new QueueTrackingStatistic("Scheduler." + SchedulingContext.Name);
                queueTracking.OnStartExecution();
            }

            if (StatisticsCollector.CollectPerWorkItemStats)
            {
                workItemGroupStatisticsNumber = SchedulerStatisticsGroup.RegisterWorkItemGroup(SchedulingContext.Name, SchedulingContext,
                    () =>
                    {
                        var sb = new StringBuilder();
                        lock (lockable)
                        {
                                    
                            sb.Append("QueueLength = " + WorkItemCount);
                            sb.Append(String.Format(", State = {0}", state));
                            if (state == WorkGroupStatus.Runnable)
                                sb.Append(String.Format("; oldest item is {0} old", workItems.Count >= 0 ? workItems.Peek().ToString() : "null"));
                        }
                        return sb.ToString();
                    });
            }
        }<commit_msg>Create a Reactive ActivationTaskRunner when the  WorkItemGroup needs one by lazily creating it, instead of always creating one.<commit_after>using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Text;
using System.Threading.Tasks;

using Orleans.Runtime.Configuration;


namespace Orleans.Runtime.Scheduler
{
    [DebuggerDisplay("WorkItemGroup Name={Name} State={state}")]
    internal class WorkItemGroup : IWorkItem
    {
        private enum WorkGroupStatus
        {
            Waiting = 0,
            Runnable = 1,
            Running = 2,
            Shutdown = 3
        }

        private static readonly Logger appLogger = LogManager.GetLogger("Scheduler.WorkItemGroup", LoggerType.Runtime);
        private readonly Logger log;
        private readonly OrleansTaskScheduler masterScheduler;
        private WorkGroupStatus state;
        private readonly Object lockable;
        private readonly Queue<ActivationTask> workItems;

        private long totalItemsEnQueued;    // equals total items queued, + 1
        private long totalItemsProcessed;
        private readonly QueueTrackingStatistic queueTracking;
        private TimeSpan totalQueuingDelay;
        private readonly long quantumExpirations;
        private readonly int workItemGroupStatisticsNumber;

        internal ActivationTaskScheduler TaskRunner { get; private set; }

        // Lazily create a Reactive Task Runner when this group requires one.
        internal ActivationTaskScheduler _RcTaskRunner;
        internal ActivationTaskScheduler RcTaskRunner
        {
            get
            {
                if (_RcTaskRunner == null)
                {
                    _RcTaskRunner = new ActivationTaskScheduler(this, true);
                }
                return _RcTaskRunner;
            }
        }

        public DateTime TimeQueued { get; set; }

        public TimeSpan TimeSinceQueued
        {
            get { return Utils.Since(TimeQueued); } 
        }

        public ISchedulingContext SchedulingContext { get; set; }

        public bool IsSystemPriority
        {
            get { return SchedulingUtils.IsSystemPriorityContext(SchedulingContext); }
        }

        internal bool IsSystemGroup
        {
            get { return SchedulingUtils.IsSystemContext(SchedulingContext); }
        }

        public string Name { get { return SchedulingContext == null ? "unknown" : SchedulingContext.Name; } }

        internal int ExternalWorkItemCount
        {
            get { lock (lockable) { return WorkItemCount; } }
        }

        private int WorkItemCount
        {
            get { return workItems.Count; } 
        }

        internal float AverageQueueLenght
        {
            get 
            {
#if TRACK_DETAILED_STATS
                if (StatisticsCollector.CollectShedulerQueuesStats)
                {
                    return queueTracking.AverageQueueLength;
                }
#endif
                return 0;
            }
        }

        internal float NumEnqueuedRequests
        {
            get
            {
#if TRACK_DETAILED_STATS
                if (StatisticsCollector.CollectShedulerQueuesStats)
                {
                    return queueTracking.NumEnqueuedRequests;
                }
#endif
                return 0;
            }
        }

        internal float ArrivalRate
        {
            get
            {
#if TRACK_DETAILED_STATS
                if (StatisticsCollector.CollectShedulerQueuesStats)
                {
                    return queueTracking.ArrivalRate;
                }
#endif
                return 0;
            }
        }

        private bool IsActive
        {
            get
            {
                return WorkItemCount != 0;
            }
        }

        // This is the maximum number of work items to be processed in an activation turn. 
        // If this is set to zero or a negative number, then the full work queue is drained (MaxTimePerTurn allowing).
        private const int MaxWorkItemsPerTurn = 0; // Unlimited
        // This is a soft time limit on the duration of activation macro-turn (a number of micro-turns). 
        // If a activation was running its micro-turns longer than this, we will give up the thread.
        // If this is set to zero or a negative number, then the full work queue is drained (MaxWorkItemsPerTurn allowing).
        public static TimeSpan ActivationSchedulingQuantum { get; set; }
        // This is the maximum number of waiting threads (blocked in WaitForResponse) allowed
        // per ActivationWorker. An attempt to wait when there are already too many threads waiting
        // will result in a TooManyWaitersException being thrown.
        //private static readonly int MaxWaitingThreads = 500;


        internal WorkItemGroup(OrleansTaskScheduler sched, ISchedulingContext schedulingContext)
        {
            masterScheduler = sched;
            SchedulingContext = schedulingContext;
            state = WorkGroupStatus.Waiting;
            workItems = new Queue<ActivationTask>();
            lockable = new Object();
            totalItemsEnQueued = 0;
            totalItemsProcessed = 0;
            totalQueuingDelay = TimeSpan.Zero;
            quantumExpirations = 0;
            TaskRunner = new ActivationTaskScheduler(this);
            
            log = IsSystemPriority ? LogManager.GetLogger("Scheduler." + Name + ".WorkItemGroup", LoggerType.Runtime) : appLogger;
      
            if (StatisticsCollector.CollectShedulerQueuesStats)
            {
                queueTracking = new QueueTrackingStatistic("Scheduler." + SchedulingContext.Name);
                queueTracking.OnStartExecution();
            }

            if (StatisticsCollector.CollectPerWorkItemStats)
            {
                workItemGroupStatisticsNumber = SchedulerStatisticsGroup.RegisterWorkItemGroup(SchedulingContext.Name, SchedulingContext,
                    () =>
                    {
                        var sb = new StringBuilder();
                        lock (lockable)
                        {
                                    
                            sb.Append("QueueLength = " + WorkItemCount);
                            sb.Append(String.Format(", State = {0}", state));
                            if (state == WorkGroupStatus.Runnable)
                                sb.Append(String.Format("; oldest item is {0} old", workItems.Count >= 0 ? workItems.Peek().ToString() : "null"));
                        }
                        return sb.ToString();
                    });
            }
        }
{'fraction_non_alphanumeric': 0.06228615863141524, 'fraction_numerical': 0.0029548989113530326, 'mean_word_length': 1.7236340533672172, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '968978', 'n_tokens_mistral': 3143, 'n_tokens_neox': 2927, 'n_words': 848}