using System.Collections.Concurrent;
namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Scripts;
///
/// A dedicated, bounded for running script
/// and alarm on-trigger bodies.
///
/// Script bodies may perform synchronous blocking I/O (a database connection, a
/// synchronous external-system call). Running them on the shared .NET
/// lets a burst of blocking scripts starve the pool and stall
/// unrelated Akka dispatchers and HTTP request handling. This scheduler owns a fixed set
/// of dedicated threads, so script blocking is contained to those threads and cannot
/// exhaust the global pool.
///
/// The scheduler is process-wide (one set of threads for all instances) and is sized
/// from the first time it is configured.
///
public sealed class ScriptExecutionScheduler : TaskScheduler, IDisposable
{
private readonly BlockingCollection _queue = new();
private readonly List _threads;
private int _disposed;
// Per-worker "busy since" timestamp (Environment.TickCount64 ms) while a task
// is executing on that worker, 0 when idle. Written by the worker thread,
// read (lock-free) by the observability gauges below. S2/UA5: makes a
// saturated or stuck script-execution pool visible on the site health report.
private readonly long[] _busySinceTicks;
private static volatile ScriptExecutionScheduler? _shared;
private static readonly object SharedLock = new();
///
/// The process-wide script-execution scheduler. Lazily created on first use with the
/// thread count from ; the
/// first caller wins, subsequent calls reuse the existing instance.
///
/// Site runtime options supplying the thread count for the scheduler.
/// The process-wide instance, creating it on first call.
public static ScriptExecutionScheduler Shared(SiteRuntimeOptions options)
{
if (_shared != null)
return _shared;
lock (SharedLock)
{
return _shared ??= new ScriptExecutionScheduler(options.ScriptExecutionThreadCount);
}
}
///
/// Creates a scheduler backed by dedicated threads.
///
/// Number of dedicated worker threads to create.
public ScriptExecutionScheduler(int threadCount)
{
if (threadCount < 1)
threadCount = 1;
_busySinceTicks = new long[threadCount];
_threads = new List(threadCount);
for (var i = 0; i < threadCount; i++)
{
var index = i; // capture per-worker slot index
var thread = new Thread(() => WorkerLoop(index))
{
IsBackground = true,
Name = $"script-execution-{i}"
};
_threads.Add(thread);
thread.Start();
}
}
///
public override int MaximumConcurrencyLevel => _threads.Count;
/// Number of tasks waiting in the queue (not counting those currently executing).
public int QueueDepth => _queue.Count;
/// Number of worker threads currently executing a task.
public int BusyThreadCount
{
get
{
var count = 0;
for (var i = 0; i < _busySinceTicks.Length; i++)
if (Volatile.Read(ref _busySinceTicks[i]) != 0) count++;
return count;
}
}
///
/// Age of the longest-running in-flight task across all workers, or null when
/// the pool is idle. A large value with a full
/// indicates a stuck/blocked script holding a dedicated thread.
///
public TimeSpan? OldestBusyAge
{
get
{
var now = Environment.TickCount64;
long oldestSince = 0;
for (var i = 0; i < _busySinceTicks.Length; i++)
{
var since = Volatile.Read(ref _busySinceTicks[i]);
if (since != 0 && (oldestSince == 0 || since < oldestSince))
oldestSince = since;
}
return oldestSince == 0 ? null : TimeSpan.FromMilliseconds(Math.Max(0, now - oldestSince));
}
}
private void WorkerLoop(int index)
{
try
{
foreach (var task in _queue.GetConsumingEnumerable())
{
Volatile.Write(ref _busySinceTicks[index], Environment.TickCount64);
try
{
TryExecuteTask(task);
}
finally
{
Volatile.Write(ref _busySinceTicks[index], 0);
}
}
}
catch (ObjectDisposedException)
{
// Scheduler disposed — worker exits.
}
}
///
protected override void QueueTask(Task task) => _queue.Add(task);
///
protected override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
{
// Only inline if we are already on one of this scheduler's worker threads,
// so script work never escapes onto a thread-pool thread.
if (Thread.CurrentThread.Name?.StartsWith("script-execution-", StringComparison.Ordinal) != true)
return false;
return TryExecuteTask(task);
}
///
protected override IEnumerable GetScheduledTasks() => _queue.ToArray();
/// Signals the worker threads to stop and waits for them to drain, then disposes the queue.
public void Dispose()
{
if (Interlocked.Exchange(ref _disposed, 1) != 0)
return;
_queue.CompleteAdding();
foreach (var thread in _threads)
thread.Join(TimeSpan.FromSeconds(5));
_queue.Dispose();
}
}