Merge origin/dev-v7-deploy into dev-v8-zbwip (builds)
This commit is contained in:
Stephan
@@ -1,15 +1,19 @@
|
||||
using System;
|
||||
using System.Collections.Concurrent;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using System.Threading.Tasks.Dataflow;
|
||||
using System.Web.Hosting;
|
||||
using Umbraco.Core;
|
||||
using Umbraco.Core.Events;
|
||||
using Umbraco.Core.Logging;
|
||||
|
||||
namespace Umbraco.Web.Scheduling
|
||||
{
|
||||
// exists for logging purposes
|
||||
internal class BackgroundTaskRunner
|
||||
/// <summary>
|
||||
/// Manages a queue of tasks and runs them in the background.
|
||||
/// </summary>
|
||||
/// <remarks>This class exists for logging purposes - the one you want to use is BackgroundTaskRunner{T}.</remarks>
|
||||
public abstract class BackgroundTaskRunner
|
||||
{ }
|
||||
|
||||
/// <summary>
|
||||
@@ -18,80 +22,114 @@ namespace Umbraco.Web.Scheduling
|
||||
/// <typeparam name="T">The type of the managed tasks.</typeparam>
|
||||
/// <remarks>The task runner is web-aware and will ensure that it shuts down correctly when the AppDomain
|
||||
/// shuts down (ie is unloaded).</remarks>
|
||||
internal class BackgroundTaskRunner<T> : BackgroundTaskRunner, IBackgroundTaskRunner<T>
|
||||
public class BackgroundTaskRunner<T> : BackgroundTaskRunner, IBackgroundTaskRunner<T>
|
||||
where T : class, IBackgroundTask
|
||||
{
|
||||
// do not remove this comment!
|
||||
//
|
||||
// if you plan to do anything on this class, first go and read
|
||||
// http://blog.stephencleary.com/2012/12/dont-block-in-asynchronous-code.html
|
||||
// http://stackoverflow.com/questions/19481964/calling-taskcompletionsource-setresult-in-a-non-blocking-manner
|
||||
// http://stackoverflow.com/questions/21225361/is-there-anything-like-asynchronous-blockingcollectiont
|
||||
// and more, and more, and more
|
||||
// and remember: async is hard
|
||||
|
||||
private readonly string _logPrefix;
|
||||
private readonly BackgroundTaskRunnerOptions _options;
|
||||
private readonly ILogger _logger;
|
||||
private readonly BlockingCollection<T> _tasks = new BlockingCollection<T>();
|
||||
private readonly object _locker = new object();
|
||||
|
||||
// that event is used to stop the pump when it is alive and waiting
|
||||
// on a latched task - so it waits on the latch, the cancellation token,
|
||||
// and the completed event
|
||||
private readonly ManualResetEventSlim _completedEvent = new ManualResetEventSlim(false);
|
||||
private readonly BufferBlock<T> _tasks = new BufferBlock<T>(new DataflowBlockOptions());
|
||||
|
||||
// in various places we are testing these vars outside a lock, so make them volatile
|
||||
private volatile bool _isRunning; // is running
|
||||
private volatile bool _isCompleted; // does not accept tasks anymore, may still be running
|
||||
private volatile bool _completed; // does not accept tasks anymore, may still be running
|
||||
|
||||
private Task _runningTask;
|
||||
private CancellationTokenSource _tokenSource;
|
||||
private Task _runningTask; // the threading task that is currently executing background tasks
|
||||
private CancellationTokenSource _shutdownTokenSource; // used to cancel everything and shutdown
|
||||
private CancellationTokenSource _cancelTokenSource; // used to cancel the current task
|
||||
private CancellationToken _shutdownToken;
|
||||
|
||||
private bool _terminating; // ensures we raise that event only once
|
||||
private bool _terminated; // remember we've terminated
|
||||
private TaskCompletionSource<int> _terminatedSource; // awaitable source
|
||||
private readonly TaskCompletionSource<int> _terminatedSource = new TaskCompletionSource<int>(); // enable awaiting termination
|
||||
|
||||
internal event TypedEventHandler<BackgroundTaskRunner<T>, TaskEventArgs<T>> TaskError;
|
||||
internal event TypedEventHandler<BackgroundTaskRunner<T>, TaskEventArgs<T>> TaskStarting;
|
||||
internal event TypedEventHandler<BackgroundTaskRunner<T>, TaskEventArgs<T>> TaskCompleted;
|
||||
internal event TypedEventHandler<BackgroundTaskRunner<T>, TaskEventArgs<T>> TaskCancelled;
|
||||
|
||||
// triggers when the runner stops (but could start again if a task is added to it)
|
||||
internal event TypedEventHandler<BackgroundTaskRunner<T>, EventArgs> Stopped;
|
||||
|
||||
// triggers when the hosting environment requests that the runner terminates
|
||||
internal event TypedEventHandler<BackgroundTaskRunner<T>, EventArgs> Terminating;
|
||||
|
||||
// triggers when the runner terminates (no task can be added, no task is running)
|
||||
internal event TypedEventHandler<BackgroundTaskRunner<T>, EventArgs> Terminated;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="BackgroundTaskRunner{T}"/> class.
|
||||
/// </summary>
|
||||
public BackgroundTaskRunner(ILogger logger)
|
||||
: this(typeof (T).FullName, new BackgroundTaskRunnerOptions(), logger)
|
||||
{ }
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="BackgroundTaskRunner{T}"/> class.
|
||||
/// </summary>
|
||||
/// <param name="name">The name of the runner.</param>
|
||||
/// <param name="logger"></param>
|
||||
public BackgroundTaskRunner(string name, ILogger logger)
|
||||
: this(name, new BackgroundTaskRunnerOptions(), logger)
|
||||
{ }
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="BackgroundTaskRunner{T}"/> class with a set of options.
|
||||
/// </summary>
|
||||
/// <param name="options">The set of options.</param>
|
||||
/// <param name="logger"></param>
|
||||
public BackgroundTaskRunner(BackgroundTaskRunnerOptions options, ILogger logger)
|
||||
: this(typeof (T).FullName, options, logger)
|
||||
{ }
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="BackgroundTaskRunner{T}"/> class with a set of options.
|
||||
/// </summary>
|
||||
/// <param name="name">The name of the runner.</param>
|
||||
/// <param name="options">The set of options.</param>
|
||||
/// <param name="logger"></param>
|
||||
public BackgroundTaskRunner(string name, BackgroundTaskRunnerOptions options, ILogger logger)
|
||||
// fixme - this is temp
|
||||
// at the moment MainDom is internal so we have to find a way to hook into it - temp
|
||||
public class MainDomHook
|
||||
{
|
||||
if (options == null) throw new ArgumentNullException("options");
|
||||
if (logger == null) throw new ArgumentNullException("logger");
|
||||
private MainDomHook(MainDom mainDom, Action install, Action release)
|
||||
{
|
||||
MainDom = mainDom;
|
||||
Install = install;
|
||||
Release = release;
|
||||
}
|
||||
|
||||
internal MainDom MainDom { get; }
|
||||
public Action Install { get; }
|
||||
public Action Release { get; }
|
||||
|
||||
public static MainDomHook Create(Action install, Action release)
|
||||
{
|
||||
return new MainDomHook(Core.DI.Current.Container.GetInstance<MainDom>(), install, release);
|
||||
}
|
||||
|
||||
public static MainDomHook CreateForTest(Action install, Action release)
|
||||
{
|
||||
return new MainDomHook(null, install, release);
|
||||
}
|
||||
|
||||
public bool Register()
|
||||
{
|
||||
if (MainDom != null)
|
||||
return MainDom.Register(Install, Release);
|
||||
|
||||
// tests
|
||||
Install?.Invoke();
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="BackgroundTaskRunner{T}"/> class.
|
||||
/// </summary>
|
||||
/// <param name="logger">A logger.</param>
|
||||
/// <param name="hook">An optional main domain hook.</param>
|
||||
public BackgroundTaskRunner(ILogger logger, MainDomHook hook = null)
|
||||
: this(typeof(T).FullName, new BackgroundTaskRunnerOptions(), logger, hook)
|
||||
{ }
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="BackgroundTaskRunner{T}"/> class.
|
||||
/// </summary>
|
||||
/// <param name="name">The name of the runner.</param>
|
||||
/// <param name="logger">A logger.</param>
|
||||
/// <param name="hook">An optional main domain hook.</param>
|
||||
public BackgroundTaskRunner(string name, ILogger logger, MainDomHook hook = null)
|
||||
: this(name, new BackgroundTaskRunnerOptions(), logger, hook)
|
||||
{ }
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="BackgroundTaskRunner{T}"/> class with a set of options.
|
||||
/// </summary>
|
||||
/// <param name="options">The set of options.</param>
|
||||
/// <param name="logger">A logger.</param>
|
||||
/// <param name="hook">An optional main domain hook.</param>
|
||||
public BackgroundTaskRunner(BackgroundTaskRunnerOptions options, ILogger logger, MainDomHook hook = null)
|
||||
: this(typeof(T).FullName, options, logger, hook)
|
||||
{ }
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="BackgroundTaskRunner{T}"/> class with a set of options.
|
||||
/// </summary>
|
||||
/// <param name="name">The name of the runner.</param>
|
||||
/// <param name="options">The set of options.</param>
|
||||
/// <param name="logger">A logger.</param>
|
||||
/// <param name="hook">An optional main domain hook.</param>
|
||||
public BackgroundTaskRunner(string name, BackgroundTaskRunnerOptions options, ILogger logger, MainDomHook hook = null)
|
||||
{
|
||||
if (options == null) throw new ArgumentNullException(nameof(options));
|
||||
if (logger == null) throw new ArgumentNullException(nameof(logger));
|
||||
_options = options;
|
||||
_logPrefix = "[" + name + "] ";
|
||||
_logger = logger;
|
||||
@@ -99,56 +137,52 @@ namespace Umbraco.Web.Scheduling
|
||||
if (options.Hosted)
|
||||
HostingEnvironment.RegisterObject(this);
|
||||
|
||||
if (options.AutoStart)
|
||||
if (hook != null)
|
||||
_completed = _terminated = hook.Register() == false;
|
||||
|
||||
if (options.AutoStart && _terminated == false)
|
||||
StartUp();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets the number of tasks in the queue.
|
||||
/// </summary>
|
||||
public int TaskCount
|
||||
{
|
||||
get { return _tasks.Count; }
|
||||
}
|
||||
public int TaskCount => _tasks.Count;
|
||||
|
||||
/// <summary>
|
||||
/// Gets a value indicating whether a task is currently running.
|
||||
/// Gets a value indicating whether a threading task is currently running.
|
||||
/// </summary>
|
||||
public bool IsRunning
|
||||
{
|
||||
get { return _isRunning; }
|
||||
}
|
||||
public bool IsRunning => _isRunning;
|
||||
|
||||
/// <summary>
|
||||
/// Gets a value indicating whether the runner has completed and cannot accept tasks anymore.
|
||||
/// </summary>
|
||||
public bool IsCompleted
|
||||
{
|
||||
get { return _isCompleted; }
|
||||
}
|
||||
public bool IsCompleted => _completed;
|
||||
|
||||
/// <summary>
|
||||
/// Gets the running task as an immutable object.
|
||||
/// Gets the running threading task as an immutable awaitable.
|
||||
/// </summary>
|
||||
/// <exception cref="InvalidOperationException">There is no running task.</exception>
|
||||
/// <remarks>
|
||||
/// Unless the AutoStart option is true, there will be no running task until
|
||||
/// a background task is added to the queue. Unless the KeepAlive option is true, there
|
||||
/// will be no running task when the queue is empty.
|
||||
/// <para>Unless the AutoStart option is true, there will be no current threading task until
|
||||
/// a background task is added to the queue, and there will be no current threading task
|
||||
/// when the queue is empty. In which case this method returns null.</para>
|
||||
/// <para>The returned value can be awaited and that is all (eg no continuation).</para>
|
||||
/// </remarks>
|
||||
public ThreadingTaskImmutable CurrentThreadingTask
|
||||
internal ThreadingTaskImmutable CurrentThreadingTask
|
||||
{
|
||||
get
|
||||
{
|
||||
lock (_locker)
|
||||
{
|
||||
if (_runningTask == null)
|
||||
throw new InvalidOperationException("There is no current Threading.Task.");
|
||||
return new ThreadingTaskImmutable(_runningTask);
|
||||
return _runningTask == null ? null : new ThreadingTaskImmutable(_runningTask);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// fixme should the above throw, return null, a completed task?
|
||||
|
||||
// fixme what's the diff?!
|
||||
/// <summary>
|
||||
/// Gets an awaitable used to await the runner running operation.
|
||||
/// </summary>
|
||||
@@ -168,20 +202,21 @@ namespace Umbraco.Web.Scheduling
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets an awaitable used to await the runner.
|
||||
/// Gets an awaitable object that can be used to await for the runner to terminate.
|
||||
/// </summary>
|
||||
/// <returns>An awaitable instance.</returns>
|
||||
/// <remarks>Used to wait until the runner is terminated.</remarks>
|
||||
public ThreadingTaskImmutable TerminatedAwaitable
|
||||
/// <returns>An awaitable object.</returns>
|
||||
/// <remarks>
|
||||
/// <para>Used to wait until the runner has terminated.</para>
|
||||
/// <para>This is for unit tests and should not be used otherwise. In most cases when the runner
|
||||
/// has terminated, the application domain is going down and it is not the right time to do things.</para>
|
||||
/// </remarks>
|
||||
internal ThreadingTaskImmutable TerminatedAwaitable
|
||||
{
|
||||
get
|
||||
{
|
||||
lock (_locker)
|
||||
{
|
||||
if (_terminatedSource == null && _terminated == false)
|
||||
_terminatedSource = new TaskCompletionSource<int>();
|
||||
var task = _terminatedSource == null ? Task.FromResult(0) : _terminatedSource.Task;
|
||||
return new ThreadingTaskImmutable(task);
|
||||
return new ThreadingTaskImmutable(_terminatedSource.Task);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -195,12 +230,12 @@ namespace Umbraco.Web.Scheduling
|
||||
{
|
||||
lock (_locker)
|
||||
{
|
||||
if (_isCompleted)
|
||||
if (_completed)
|
||||
throw new InvalidOperationException("The task runner has completed.");
|
||||
|
||||
// add task
|
||||
_logger.Debug<BackgroundTaskRunner>(_logPrefix + "Task added {0}", task.GetType);
|
||||
_tasks.Add(task);
|
||||
_logger.Debug<BackgroundTaskRunner>(_logPrefix + "Task added {0}", () => task.GetType().FullName);
|
||||
_tasks.Post(task);
|
||||
|
||||
// start
|
||||
StartUpLocked();
|
||||
@@ -217,15 +252,15 @@ namespace Umbraco.Web.Scheduling
|
||||
{
|
||||
lock (_locker)
|
||||
{
|
||||
if (_isCompleted)
|
||||
if (_completed)
|
||||
{
|
||||
_logger.Debug<BackgroundTaskRunner>(_logPrefix + "Task cannot be added {0}, the task runner is already shutdown", task.GetType);
|
||||
_logger.Debug<BackgroundTaskRunner>(_logPrefix + "Task cannot be added {0}, the task runner has already shutdown", () => task.GetType().FullName);
|
||||
return false;
|
||||
}
|
||||
|
||||
// add task
|
||||
_logger.Debug<BackgroundTaskRunner>(_logPrefix + "Task added {0}", task.GetType);
|
||||
_tasks.Add(task);
|
||||
_logger.Debug<BackgroundTaskRunner>(_logPrefix + "Task added {0}", () => task.GetType().FullName);
|
||||
_tasks.Post(task);
|
||||
|
||||
// start
|
||||
StartUpLocked();
|
||||
@@ -234,18 +269,32 @@ namespace Umbraco.Web.Scheduling
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Cancels to current task, if any.
|
||||
/// </summary>
|
||||
/// <remarks>Has no effect if the task runs synchronously, or does not want to cancel.</remarks>
|
||||
public void CancelCurrentBackgroundTask()
|
||||
{
|
||||
lock (_locker)
|
||||
{
|
||||
if (_completed)
|
||||
throw new InvalidOperationException("The task runner has completed.");
|
||||
_cancelTokenSource?.Cancel();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts the tasks runner, if not already running.
|
||||
/// </summary>
|
||||
/// <remarks>Is invoked each time a task is added, to ensure it is going to be processed.</remarks>
|
||||
/// <exception cref="InvalidOperationException">The task runner has completed.</exception>
|
||||
public void StartUp()
|
||||
internal void StartUp()
|
||||
{
|
||||
if (_isRunning) return;
|
||||
|
||||
lock (_locker)
|
||||
{
|
||||
if (_isCompleted)
|
||||
if (_completed)
|
||||
throw new InvalidOperationException("The task runner has completed.");
|
||||
|
||||
StartUpLocked();
|
||||
@@ -263,8 +312,10 @@ namespace Umbraco.Web.Scheduling
|
||||
_isRunning = true;
|
||||
|
||||
// create a new token source since this is a new process
|
||||
_tokenSource = new CancellationTokenSource();
|
||||
_runningTask = PumpIBackgroundTasks(Task.Factory, _tokenSource.Token);
|
||||
_shutdownTokenSource = new CancellationTokenSource();
|
||||
_shutdownToken = _shutdownTokenSource.Token;
|
||||
_runningTask = Task.Run(async () => await Pump().ConfigureAwait(false), _shutdownToken);
|
||||
|
||||
_logger.Debug<BackgroundTaskRunner>(_logPrefix + "Starting");
|
||||
}
|
||||
|
||||
@@ -279,176 +330,181 @@ namespace Umbraco.Web.Scheduling
|
||||
{
|
||||
lock (_locker)
|
||||
{
|
||||
_isCompleted = true; // do not accept new tasks
|
||||
_completed = true; // do not accept new tasks
|
||||
if (_isRunning == false) return; // done already
|
||||
}
|
||||
|
||||
// try to be nice
|
||||
// assuming multiple threads can do these without problems
|
||||
_completedEvent.Set();
|
||||
_tasks.CompleteAdding();
|
||||
// complete the queue
|
||||
// will stop waiting on the queue or on a latch
|
||||
_tasks.Complete();
|
||||
|
||||
if (force)
|
||||
{
|
||||
// we must bring everything down, now
|
||||
Thread.Sleep(100); // give time to CompleteAdding()
|
||||
Thread.Sleep(100); // give time to Complete()
|
||||
lock (_locker)
|
||||
{
|
||||
// was CompleteAdding() enough?
|
||||
// was Complete() enough?
|
||||
if (_isRunning == false) return;
|
||||
}
|
||||
// try to cancel running async tasks (cannot do much about sync tasks)
|
||||
// break delayed tasks delay
|
||||
// truncate running queues
|
||||
_tokenSource.Cancel(false); // false is the default
|
||||
// break latched tasks
|
||||
// stop processing the queue
|
||||
_shutdownTokenSource.Cancel(false); // false is the default
|
||||
}
|
||||
|
||||
// tasks in the queue will be executed...
|
||||
if (wait == false) return;
|
||||
|
||||
if (_runningTask != null)
|
||||
_runningTask.Wait(); // wait for whatever is running to end...
|
||||
_runningTask?.Wait(); // wait for whatever is running to end...
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Runs background tasks for as long as there are background tasks in the queue, with an asynchronous operation.
|
||||
/// </summary>
|
||||
/// <param name="factory">The supporting <see cref="TaskFactory"/>.</param>
|
||||
/// <param name="token">A cancellation token.</param>
|
||||
/// <returns>The asynchronous operation.</returns>
|
||||
private Task PumpIBackgroundTasks(TaskFactory factory, CancellationToken token)
|
||||
private async Task Pump()
|
||||
{
|
||||
var taskSource = new TaskCompletionSource<object>(factory.CreationOptions);
|
||||
var enumerator = _options.KeepAlive ? _tasks.GetConsumingEnumerable(token).GetEnumerator() : null;
|
||||
|
||||
// ReSharper disable once MethodSupportsCancellation // always run
|
||||
var taskSourceContinuing = taskSource.Task.ContinueWith(t =>
|
||||
while (true)
|
||||
{
|
||||
// because the pump does not lock, there's a race condition,
|
||||
// the pump may stop and then we still have tasks to process,
|
||||
// and then we must restart the pump - lock to avoid race cond
|
||||
var onStopped = false;
|
||||
// get the next task
|
||||
// if it returns null the runner is going down, stop
|
||||
var bgTask = await GetNextBackgroundTask(_shutdownToken);
|
||||
if (bgTask == null) return;
|
||||
|
||||
// set a cancellation source so that the current task can be cancelled
|
||||
// link from _shutdownToken so that we can use _cancelTokenSource for both
|
||||
lock (_locker)
|
||||
{
|
||||
if (token.IsCancellationRequested || _tasks.Count == 0)
|
||||
{
|
||||
_logger.Debug<BackgroundTaskRunner>(_logPrefix + "Stopping");
|
||||
|
||||
if (_options.PreserveRunningTask == false)
|
||||
_runningTask = null;
|
||||
|
||||
// stopped
|
||||
_isRunning = false;
|
||||
onStopped = true;
|
||||
}
|
||||
_cancelTokenSource = CancellationTokenSource.CreateLinkedTokenSource(_shutdownToken);
|
||||
}
|
||||
|
||||
if (onStopped)
|
||||
// wait for latch should return the task
|
||||
// if it returns null it's either that the task has been cancelled
|
||||
// or the whole runner is going down - in both cases, continue,
|
||||
// and GetNextBackgroundTask will take care of shutdowns
|
||||
bgTask = await WaitForLatch(bgTask, _cancelTokenSource.Token);
|
||||
if (bgTask == null) continue;
|
||||
|
||||
// executes & be safe - RunAsync should NOT throw but only raise an event,
|
||||
// but... just make sure we never ever take everything down
|
||||
try
|
||||
{
|
||||
OnEvent(Stopped, "Stopped");
|
||||
return;
|
||||
await RunAsync(bgTask, _cancelTokenSource.Token).ConfigureAwait(false);
|
||||
}
|
||||
|
||||
// if _runningTask is taskSource.Task then we must keep continuing it,
|
||||
// not starting a new taskSource, else _runningTask would complete and
|
||||
// something may be waiting on it
|
||||
//PumpIBackgroundTasks(factory, token); // restart
|
||||
// ReSharper disable MethodSupportsCancellation // always run
|
||||
t.ContinueWithTask(_ => PumpIBackgroundTasks(factory, token)); // restart
|
||||
// ReSharper restore MethodSupportsCancellation
|
||||
});
|
||||
|
||||
Action<Task> pump = null;
|
||||
pump = task =>
|
||||
{
|
||||
// RunIBackgroundTaskAsync does NOT throw exceptions, just raises event
|
||||
// so if we have an exception here, really, wtf? - must read the exception
|
||||
// anyways so it does not bubble up and kill everything
|
||||
if (task != null && task.IsFaulted)
|
||||
catch (Exception e)
|
||||
{
|
||||
var exception = task.Exception;
|
||||
_logger.Error<BackgroundTaskRunner>(_logPrefix + "Task runner exception.", exception);
|
||||
_logger.Error<BackgroundTaskRunner>(_logPrefix + "Task runner exception.", e);
|
||||
}
|
||||
|
||||
// is it ok to run?
|
||||
if (TaskSourceCanceled(taskSource, token)) return;
|
||||
|
||||
// try to get a task
|
||||
// the blocking MoveNext will end if token is cancelled or collection is completed
|
||||
T bgTask;
|
||||
var hasBgTask = _options.KeepAlive
|
||||
// ReSharper disable once PossibleNullReferenceException
|
||||
? (bgTask = enumerator.MoveNext() ? enumerator.Current : null) != null // blocking
|
||||
: _tasks.TryTake(out bgTask); // non-blocking
|
||||
|
||||
// no task, signal the runner we're done
|
||||
if (hasBgTask == false)
|
||||
// done
|
||||
lock (_locker)
|
||||
{
|
||||
TaskSourceCompleted(taskSource, token);
|
||||
return;
|
||||
_cancelTokenSource = null;
|
||||
}
|
||||
|
||||
// wait for latched task, supporting cancellation
|
||||
var dbgTask = bgTask as ILatchedBackgroundTask;
|
||||
if (dbgTask != null && dbgTask.IsLatched)
|
||||
{
|
||||
WaitHandle.WaitAny(new[] { dbgTask.Latch, token.WaitHandle, _completedEvent.WaitHandle });
|
||||
if (TaskSourceCanceled(taskSource, token)) return;
|
||||
// else run now, either because latch ok or runner is completed
|
||||
// still latched & not running on shutdown = stop here
|
||||
if (dbgTask.IsLatched && dbgTask.RunsOnShutdown == false)
|
||||
{
|
||||
dbgTask.Dispose(); // will not run
|
||||
TaskSourceCompleted(taskSource, token);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
// run the task as first task, or a continuation
|
||||
task = task == null
|
||||
? RunIBackgroundTaskAsync(bgTask, token)
|
||||
// ReSharper disable once MethodSupportsCancellation // always run
|
||||
: task.ContinueWithTask(_ => RunIBackgroundTaskAsync(bgTask, token));
|
||||
|
||||
// and pump
|
||||
// ReSharper disable once MethodSupportsCancellation // always run
|
||||
task.ContinueWith(t => pump(t));
|
||||
};
|
||||
|
||||
// start it all
|
||||
factory.StartNew(() => pump(null),
|
||||
token,
|
||||
_options.LongRunning ? TaskCreationOptions.LongRunning : TaskCreationOptions.None,
|
||||
TaskScheduler.Default);
|
||||
|
||||
return taskSourceContinuing;
|
||||
}
|
||||
|
||||
private static bool TaskSourceCanceled(TaskCompletionSource<object> taskSource, CancellationToken token)
|
||||
{
|
||||
if (token.IsCancellationRequested)
|
||||
{
|
||||
taskSource.SetCanceled();
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
private static void TaskSourceCompleted(TaskCompletionSource<object> taskSource, CancellationToken token)
|
||||
// gets the next background task from the buffer
|
||||
private async Task<T> GetNextBackgroundTask(CancellationToken token)
|
||||
{
|
||||
if (token.IsCancellationRequested)
|
||||
taskSource.SetCanceled();
|
||||
else
|
||||
taskSource.SetResult(null);
|
||||
while (true)
|
||||
{
|
||||
var task = await GetNextBackgroundTask2(token);
|
||||
if (task != null) return task;
|
||||
|
||||
lock (_locker)
|
||||
{
|
||||
// deal with race condition
|
||||
if (_shutdownToken.IsCancellationRequested == false && _tasks.Count > 0) continue;
|
||||
|
||||
// if we really have nothing to do, stop
|
||||
_logger.Debug<BackgroundTaskRunner>(_logPrefix + "Stopping");
|
||||
|
||||
if (_options.PreserveRunningTask == false)
|
||||
_runningTask = null;
|
||||
_isRunning = false;
|
||||
_shutdownToken = CancellationToken.None;
|
||||
}
|
||||
|
||||
OnEvent(Stopped, "Stopped");
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Runs a background task asynchronously.
|
||||
/// </summary>
|
||||
/// <param name="bgTask">The background task.</param>
|
||||
/// <param name="token">A cancellation token.</param>
|
||||
/// <returns>The asynchronous operation.</returns>
|
||||
internal async Task RunIBackgroundTaskAsync(T bgTask, CancellationToken token)
|
||||
private async Task<T> GetNextBackgroundTask2(CancellationToken shutdownToken)
|
||||
{
|
||||
// exit if cancelling
|
||||
if (shutdownToken.IsCancellationRequested)
|
||||
return null;
|
||||
|
||||
// if keepalive is false then don't block, exit if there is
|
||||
// no task in the buffer - yes, there is a race cond, which
|
||||
// we'll take care of
|
||||
if (_options.KeepAlive == false && _tasks.Count == 0)
|
||||
return null;
|
||||
|
||||
try
|
||||
{
|
||||
// A Task<TResult> that informs of whether and when more output is available. If, when the
|
||||
// task completes, its Result is true, more output is available in the source (though another
|
||||
// consumer of the source may retrieve the data). If it returns false, more output is not
|
||||
// and will never be available, due to the source completing prior to output being available.
|
||||
|
||||
var output = await _tasks.OutputAvailableAsync(shutdownToken); // block until output or cancelled
|
||||
if (output == false) return null;
|
||||
}
|
||||
catch (TaskCanceledException)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
// A task that represents the asynchronous receive operation. When an item value is successfully
|
||||
// received from the source, the returned task is completed and its Result returns the received
|
||||
// value. If an item value cannot be retrieved because the source is empty and completed, an
|
||||
// InvalidOperationException exception is thrown in the returned task.
|
||||
|
||||
// the source cannot be empty *and* completed here - we know we have output
|
||||
return await _tasks.ReceiveAsync(shutdownToken);
|
||||
}
|
||||
catch (TaskCanceledException)
|
||||
{
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
// if bgTask is not a latched background task, or if it is not latched, returns immediately
|
||||
// else waits for the latch, taking care of completion and shutdown and whatnot
|
||||
private async Task<T> WaitForLatch(T bgTask, CancellationToken token)
|
||||
{
|
||||
var latched = bgTask as ILatchedBackgroundTask;
|
||||
if (latched == null || latched.IsLatched == false) return bgTask;
|
||||
|
||||
// support cancelling awaiting
|
||||
// read https://github.com/dotnet/corefx/issues/2704
|
||||
// read http://stackoverflow.com/questions/27238232/how-can-i-cancel-task-whenall
|
||||
var tokenTaskSource = new TaskCompletionSource<bool>();
|
||||
token.Register(s => ((TaskCompletionSource<bool>)s).SetResult(true), tokenTaskSource);
|
||||
|
||||
// returns the task that completed
|
||||
// - latched.Latch completes when the latch releases
|
||||
// - _tasks.Completion completes when the runner completes
|
||||
// - tokenTaskSource.Task completes when this task, or the whole runner, is cancelled
|
||||
var task = await Task.WhenAny(latched.Latch, _tasks.Completion, tokenTaskSource.Task);
|
||||
|
||||
// ok to run now
|
||||
if (task == latched.Latch)
|
||||
return bgTask;
|
||||
|
||||
// if shutting down, return the task only if it runs on shutdown
|
||||
if (_shutdownToken.IsCancellationRequested == false && latched.RunsOnShutdown) return bgTask;
|
||||
|
||||
// else, either it does not run on shutdown or it's been cancelled, dispose
|
||||
latched.Dispose();
|
||||
return null;
|
||||
}
|
||||
|
||||
// runs the background task, taking care of shutdown (as far as possible - cannot abort
|
||||
// a non-async Run for example, so we'll do our best)
|
||||
private async Task RunAsync(T bgTask, CancellationToken token)
|
||||
{
|
||||
try
|
||||
{
|
||||
@@ -464,7 +520,7 @@ namespace Umbraco.Web.Scheduling
|
||||
else
|
||||
bgTask.Run();
|
||||
}
|
||||
finally // ensure we disposed - unless latched (again)
|
||||
finally // ensure we disposed - unless latched again ie wants to re-run
|
||||
{
|
||||
var lbgTask = bgTask as ILatchedBackgroundTask;
|
||||
if (lbgTask == null || lbgTask.IsLatched == false)
|
||||
@@ -482,11 +538,32 @@ namespace Umbraco.Web.Scheduling
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.Error<BackgroundTaskRunner>(_logPrefix + "Task has failed", ex);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#region Events
|
||||
|
||||
// triggers when a background task starts
|
||||
public event TypedEventHandler<BackgroundTaskRunner<T>, TaskEventArgs<T>> TaskStarting;
|
||||
|
||||
// triggers when a background task has completed
|
||||
public event TypedEventHandler<BackgroundTaskRunner<T>, TaskEventArgs<T>> TaskCompleted;
|
||||
|
||||
// triggers when a background task throws
|
||||
public event TypedEventHandler<BackgroundTaskRunner<T>, TaskEventArgs<T>> TaskError;
|
||||
|
||||
// triggers when a background task is cancelled
|
||||
public event TypedEventHandler<BackgroundTaskRunner<T>, TaskEventArgs<T>> TaskCancelled;
|
||||
|
||||
// triggers when the runner stops (but could start again if a task is added to it)
|
||||
internal event TypedEventHandler<BackgroundTaskRunner<T>, EventArgs> Stopped;
|
||||
|
||||
// triggers when the hosting environment requests that the runner terminates
|
||||
internal event TypedEventHandler<BackgroundTaskRunner<T>, EventArgs> Terminating;
|
||||
|
||||
// triggers when the runner has terminated (no task can be added, no task is running)
|
||||
internal event TypedEventHandler<BackgroundTaskRunner<T>, EventArgs> Terminated;
|
||||
|
||||
private void OnEvent(TypedEventHandler<BackgroundTaskRunner<T>, EventArgs> handler, string name)
|
||||
{
|
||||
if (handler == null) return;
|
||||
@@ -526,7 +603,7 @@ namespace Umbraco.Web.Scheduling
|
||||
{
|
||||
OnEvent(TaskCancelled, "TaskCancelled", e);
|
||||
|
||||
//dispose it
|
||||
// dispose it
|
||||
e.Task.Dispose();
|
||||
}
|
||||
|
||||
@@ -608,9 +685,7 @@ namespace Umbraco.Web.Scheduling
|
||||
_logger.Info<BackgroundTaskRunner>(_logPrefix + "Waiting for tasks to complete");
|
||||
Shutdown(false, false); // do not accept any more tasks, flush the queue, do not wait
|
||||
|
||||
// raise the completed event only after the running task has completed
|
||||
// and there's no more task running
|
||||
|
||||
// raise the completed event only after the running threading task has completed
|
||||
lock (_locker)
|
||||
{
|
||||
if (_runningTask != null)
|
||||
@@ -632,6 +707,7 @@ namespace Umbraco.Web.Scheduling
|
||||
}
|
||||
}
|
||||
|
||||
// called by Stop either immediately or eventually
|
||||
private void Terminate(bool immediate)
|
||||
{
|
||||
// signal the environment we have terminated
|
||||
@@ -640,8 +716,6 @@ namespace Umbraco.Web.Scheduling
|
||||
// complete the awaitable completion source, if any
|
||||
|
||||
HostingEnvironment.UnregisterObject(this);
|
||||
_logger.Info<BackgroundTaskRunner>(_logPrefix + "Tasks " + (immediate ? "cancelled" : "completed") + ", terminated");
|
||||
OnEvent(Terminated, "Terminated");
|
||||
|
||||
TaskCompletionSource<int> terminatedSource;
|
||||
lock (_locker)
|
||||
@@ -649,8 +723,12 @@ namespace Umbraco.Web.Scheduling
|
||||
_terminated = true;
|
||||
terminatedSource = _terminatedSource;
|
||||
}
|
||||
if (terminatedSource != null)
|
||||
terminatedSource.SetResult(0);
|
||||
|
||||
_logger.Info<BackgroundTaskRunner>(_logPrefix + "Tasks " + (immediate ? "cancelled" : "completed") + ", terminated");
|
||||
|
||||
OnEvent(Terminated, "Terminated");
|
||||
|
||||
terminatedSource.SetResult(0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user