// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the Apache 2.0 License.
// See the LICENSE file in the project root for more information.
#if !WINDOWS && !NO_THREAD
using System.Collections.Generic;
using System.Reactive.Disposables;
using System.Threading;
namespace System.Reactive.Concurrency
{
///
/// Represents an object that schedules units of work on the CLR thread pool.
///
/// Singleton instance of this type exposed through this static property.
public sealed class ThreadPoolScheduler : LocalScheduler, ISchedulerLongRunning, ISchedulerPeriodic
{
private static readonly Lazy s_instance = new Lazy(() => new ThreadPoolScheduler());
private static readonly Lazy s_newBackgroundThread = new Lazy(() => new NewThreadScheduler(action => new Thread(action) { IsBackground = true }));
///
/// Gets the singleton instance of the CLR thread pool scheduler.
///
public static ThreadPoolScheduler Instance => s_instance.Value;
private ThreadPoolScheduler()
{
}
///
/// Schedules an action to be executed.
///
/// The type of the state passed to the scheduled action.
/// State passed to the action to be executed.
/// Action to be executed.
/// The disposable object used to cancel the scheduled action (best effort).
/// is null.
public override IDisposable Schedule(TState state, Func action)
{
if (action == null)
throw new ArgumentNullException(nameof(action));
var workItem = new UserWorkItem(this, state, action);
ThreadPool.QueueUserWorkItem(
closureWorkItem => ((UserWorkItem)closureWorkItem).Run(),
workItem);
return workItem;
}
///
/// Schedules an action to be executed after dueTime, using a System.Threading.Timer object.
///
/// The type of the state passed to the scheduled action.
/// State passed to the action to be executed.
/// Action to be executed.
/// Relative time after which to execute the action.
/// The disposable object used to cancel the scheduled action (best effort).
/// is null.
public override IDisposable Schedule(TState state, TimeSpan dueTime, Func action)
{
if (action == null)
throw new ArgumentNullException(nameof(action));
var dt = Scheduler.Normalize(dueTime);
if (dt.Ticks == 0)
{
return Schedule(state, action);
}
return new Timer(this, state, dt, action);
}
///
/// Schedules a long-running task by creating a new thread. Cancellation happens through polling.
///
/// The type of the state passed to the scheduled action.
/// State passed to the action to be executed.
/// Action to be executed.
/// The disposable object used to cancel the scheduled action (best effort).
/// is null.
public IDisposable ScheduleLongRunning(TState state, Action action)
{
if (action == null)
throw new ArgumentNullException(nameof(action));
return s_newBackgroundThread.Value.ScheduleLongRunning(state, action);
}
///
/// Starts a new stopwatch object.
///
/// New stopwatch object; started at the time of the request.
public override IStopwatch StartStopwatch()
{
//
// Strictly speaking, this explicit override is not necessary because the base implementation calls into
// the enlightenment module to obtain the CAL, which would circle back to System.Reactive.PlatformServices
// where we're currently running. This is merely a short-circuit to avoid the additional roundtrip.
//
return new StopwatchImpl();
}
///
/// Schedules a periodic piece of work, using a System.Threading.Timer object.
///
/// The type of the state passed to the scheduled action.
/// Initial state passed to the action upon the first iteration.
/// Period for running the work periodically.
/// Action to be executed, potentially updating the state.
/// The disposable object used to cancel the scheduled recurring action (best effort).
/// is null.
/// is less than zero.
public IDisposable SchedulePeriodic(TState state, TimeSpan period, Func action)
{
if (period < TimeSpan.Zero)
throw new ArgumentOutOfRangeException(nameof(period));
if (action == null)
throw new ArgumentNullException(nameof(action));
if (period == TimeSpan.Zero)
{
return new FastPeriodicTimer(state, action);
}
else
{
return new PeriodicTimer(state, period, action);
}
}
private sealed class FastPeriodicTimer : IDisposable
{
private TState _state;
private Func _action;
private volatile bool _disposed;
public FastPeriodicTimer(TState state, Func action)
{
_state = state;
_action = action;
ThreadPool.QueueUserWorkItem(_ => Tick(_), this); // Replace with method group as soon as Roslyn will cache the delegate then.
}
private static void Tick(object state)
{
var timer = (FastPeriodicTimer)state;
if (!timer._disposed)
{
timer._state = timer._action(timer._state);
ThreadPool.QueueUserWorkItem(_ => Tick(_), timer);
}
}
public void Dispose()
{
_disposed = true;
_action = Stubs.I;
}
}
//
// See ConcurrencyAbstractionLayerImpl.cs for more information about the code
// below and its timer rooting behavior.
//
private sealed class Timer : IDisposable
{
private readonly MultipleAssignmentDisposable _disposable;
private readonly IScheduler _parent;
private readonly TState _state;
private Func _action;
private volatile System.Threading.Timer _timer;
public Timer(IScheduler parent, TState state, TimeSpan dueTime, Func action)
{
_parent = parent;
_state = state;
_action = action;
_disposable = new MultipleAssignmentDisposable();
_disposable.Disposable = Disposable.Create(Stop);
// Don't want the spin wait in Tick to get stuck if this thread gets aborted.
try { }
finally
{
//
// Rooting of the timer happens through the passed state,
// which is the current instance and has a field to store the Timer instance.
//
_timer = new System.Threading.Timer(_ => Tick(_) /* Don't convert to method group until Roslyn catches up */, this, dueTime, TimeSpan.FromMilliseconds(System.Threading.Timeout.Infinite));
}
}
private static void Tick(object state)
{
var timer = (Timer)state;
try
{
timer._disposable.Disposable = timer._action(timer._parent, timer._state);
}
finally
{
SpinWait.SpinUntil(timer.IsTimerAssigned);
timer.Stop();
}
}
private bool IsTimerAssigned() => _timer != null;
public void Dispose() => _disposable.Dispose();
private void Stop()
{
var timer = _timer;
if (timer != TimerStubs.Never)
{
_action = Nop;
_timer = TimerStubs.Never;
timer.Dispose();
}
}
private IDisposable Nop(IScheduler scheduler, TState state) => Disposable.Empty;
}
private sealed class PeriodicTimer : IDisposable
{
private TState _state;
private Func _action;
private readonly AsyncLock _gate;
private volatile System.Threading.Timer _timer;
public PeriodicTimer(TState state, TimeSpan period, Func action)
{
_state = state;
_action = action;
_gate = new AsyncLock();
//
// Rooting of the timer happens through the this.Tick delegate's target object,
// which is the current instance and has a field to store the Timer instance.
//
_timer = new System.Threading.Timer(@this => ((PeriodicTimer)@this).Tick(), this, period, period);
}
private void Tick()
{
_gate.Wait(
this,
@this =>
{
@this._state = @this._action(@this._state);
});
}
public void Dispose()
{
var timer = _timer;
if (timer != null)
{
_action = Stubs.I;
_timer = null;
timer.Dispose();
_gate.Dispose();
}
}
}
}
}
#endif