3
\$\begingroup\$

The AsyncWorkerQueue class uses TPL Data Flow and Rx to async process and cancel work in a queue. It takes in a worker factory (Func) and executes it when its turn (or not if its scheduled to be canceled).

public class AsyncWorkerQueue : IDisposable
{
    private TransformBlock<TaskFactory, TaskFactory> Worker { get; set; }
    private TransformBlock<TaskFactory, TaskFactory> DirtyWorkers { get; set; }
    private IObservable<TaskFactory> StartedTasks { get; set; }
    private IObservable<TaskFactory> CompletedTasks { get; set; }
    private TaskFactory CurrentWorker { get; set; }
    private SemaphoreSlim SuspendSemaphore { get; set; }
    private Task SuspendWorker { get; set; }
    public bool IsSuspended { get; private set; }
    private CancellationTokenSource QueueWorkersToken { get; set; }
    private CancellationTokenSource SuspendToken { get; set; }
    public AsyncWorkerQueue(Action<Exception> handleException = null)
    {
        DirtyWorkers = new TransformBlock<TaskFactory, TaskFactory>(tf => tf);
        Worker = new TransformBlock<TaskFactory, TaskFactory>(tf =>
        {
            try
            {
                if (SuspendToken.IsCancellationRequested)
                {
                    tf.CancelImmediately();
                    DirtyWorkers.Post(tf);
                    return tf;
                }
                CurrentWorker = tf;
                tf.Execute();
                DirtyWorkers.Post(tf);
                tf.Worker.Wait();
            }
            catch (OperationCanceledException) { }
            catch (Exception ex) { if (handleException != null) handleException(ex); }
            return tf;
        });
        CompletedTasks = Worker.AsObservable();
        QueueWorkersToken = new CancellationTokenSource();
        StartedTasks = DirtyWorkers.AsObservable();
        SuspendSemaphore = new SemaphoreSlim(0);
        SuspendWorker = Task.FromResult(false);
        SuspendToken = new CancellationTokenSource();
    }
    public Task DoWork(Func<Task> workerFactory, CancellationTokenSource source = null)
    {
        return DoWorker(workerFactory, source);
    }
    private async Task DoWorker(Func<Task> workerFactory, CancellationTokenSource source = null)
    {
        try
        {
            await SuspendWorker;
        }
        catch(OperationCanceledException) {}
        QueueWorkersToken.Token.ThrowIfCancellationRequested();
        TaskFactory tf = new TaskFactory(workerFactory, source);
        SemaphoreSlim s = new SemaphoreSlim(0);
        bool isCanceled = false;
        Task sWait = s.WaitAsync(QueueWorkersToken.Token);
        ActionBlock<TaskFactory> taskWaiter = new ActionBlock<TaskFactory>(finishedTask =>
        {
            if (finishedTask.TaskId == tf.TaskId)
            {
                isCanceled = finishedTask.Worker.IsCanceled;
                s.Release();
            }
        });
        var taskWaiterObserver = taskWaiter.AsObserver();
        using (CompletedTasks.Subscribe(taskWaiterObserver))
        {
            Worker.Post(tf);
            try
            {
                await sWait;
            }
            catch (OperationCanceledException) { throw; }
            if (isCanceled)
                throw new OperationCanceledException();
        }
    }
    public async Task WaitCurrentWorkers(Action doWait = null, Action<Exception> handleException = null, bool cancelQueue = false)
    {
        await Task.Yield();
        this.Suspend(cancelQueue);
        SemaphoreSlim s = new SemaphoreSlim(0);
        Task sWait = s.WaitAsync();
        ActionBlock<TaskFactory> taskDisposer = new ActionBlock<TaskFactory>(tf =>
        {
            tf.Cancel();
            try
            {
                tf.Worker.Wait();
            }
            catch (OperationCanceledException) { }
            catch (Exception ex) { if (handleException != null)handleException(ex); }
            if (Worker.InputCount == 0)
                s.Release();
        });
        var observer = taskDisposer.AsObserver();
        IDisposable subscriber = null;
        try
        {
            subscriber = StartedTasks.Subscribe(observer);
            if (CurrentWorker != null)
            {
                CurrentWorker.Cancel();
                try
                {
                    await CurrentWorker.Worker;
                }
                catch (OperationCanceledException) { }
                catch (Exception ex) { if (handleException != null)handleException(ex); }
            }
            if (Worker.InputCount > 0 || Worker.OutputCount > 0)
                await sWait;
        }
        catch { }
        finally
        {
            if(subscriber != null)
                subscriber.Dispose();
        }
    }
    public void Start()
    {
        if (IsSuspended)
        {
            SuspendToken = new CancellationTokenSource();
            SuspendSemaphore.Release();
            IsSuspended = false;
        }
    }
    public void Suspend(bool cancelQueueTasks = false)
    {
        if (cancelQueueTasks)
            SuspendToken.Cancel(true);
        if (!IsSuspended)
        {
            SuspendWorker = DoSuspendWorker();
            IsSuspended = true;
        }
    }
    private async Task DoSuspendWorker()
    {
        try
        {
            await SuspendSemaphore.WaitAsync();
        }
        catch (OperationCanceledException) { }
    }
    private class TaskFactory
    {
        public Guid TaskId { get; private set; }
        private Func<Task> Factory { get; set; }
        public Task Worker { get; private set; }
        private CancellationTokenSource CancellationTokenSource { get; set; }
        public TaskFactory(Func<Task> taskFactory, CancellationTokenSource source = null)
        {
            TaskId = Guid.NewGuid();
            Factory = taskFactory;
            CancellationTokenSource = source;
        }
        public void Cancel()
        {
            if(this.CancellationTokenSource != null)
                this.CancellationTokenSource.Cancel(true);
        }
        public void CancelImmediately()
        {
            this.Cancel();
            Worker = Task.FromResult(false);
        }
        public Task Execute()
        {
            Worker = Factory();
            return Worker;
        }
    }

    public async void Dispose()
    {
        if (QueueWorkersToken != null)
            QueueWorkersToken.Cancel(true);
        await WaitCurrentWorkers(cancelQueue: true);
        if (Worker != null)
            Worker.Complete();
        if (DirtyWorkers != null)
            DirtyWorkers.Complete();
    }
}

I also added an example class that uses AsyncWorkerQueue, this is a view model base for async loading/unloading of content.

public abstract class NavigationItem : BindableBase, IDisposable
{
    public NavigationData Data { get; private set; }
    public string DisplayName
    {
        get { return Data.DisplayName; }
    }
    public bool IsCloseable
    {
        get { return Data.IsCloseable; }
    }
    public NavigationItem(NavigationData data)
    {
        Data = data;
        State = NavigationItemState.Unloaded;
        Close = new DelegateCommand(() =>
        {
            Task t = Unload();
            EventHandler closed = Closed;
            if (closed != null)
                closed(this, EventArgs.Empty);
        });
        LoadWorkers = new AsyncWorkerQueue(ex => this.Unload());
        UnloadWorkers = new AsyncWorkerQueue();
    }
    private AsyncWorkerQueue LoadWorkers { get; set; }
    private AsyncWorkerQueue UnloadWorkers { get; set; }
    private NavigationItemState state;
    public NavigationItemState State
    {
        get { return state; }
        set
        {
            if (SetProperty(ref state, value))
            {
                OnPropertyChanged(() => IsUnloading);
                OnPropertyChanged(() => IsUnloaded);
                OnPropertyChanged(() => IsLoading);
                OnPropertyChanged(() => IsLoaded);
            }
        }
    }
    public bool IsUnloading { get { return State == NavigationItemState.Unloading; } }
    public bool IsUnloaded { get { return State == NavigationItemState.Unloaded; } }
    public bool IsLoading { get { return State == NavigationItemState.Loading; } }
    public bool IsLoaded { get { return State == NavigationItemState.Loaded; } }
    public event EventHandler Unloading;
    public event EventHandler Unloaded;
    public event EventHandler Loading;
    public event EventHandler Loaded;
    public event EventHandler<NavigationItemAddedEventArgs> NavigationItemAdded;
    public ICommand Close { get; private set; }
    public event EventHandler Closed;
    private void RaiseUnloading()
    {
        State = NavigationItemState.Unloading;
        var evt = Unloading;
        if (evt != null)
            evt(this, EventArgs.Empty);
    }
    private void RaiseUnloaded()
    {
        State = NavigationItemState.Unloaded;
        var evt = Unloaded;
        if (evt != null)
            evt(this, EventArgs.Empty);
    }
    private void RaiseLoading()
    {
        State = NavigationItemState.Loading;
        var evt = Loading;
        if (evt != null)
            evt(this, EventArgs.Empty);
    }
    private void RaiseLoaded()
    {
        State = NavigationItemState.Loaded;
        var evt = Loaded;
        if (evt != null)
            evt(this, EventArgs.Empty);
    }
    protected void AddNavigationItem(NavigationItem item)
    {
        var evt = NavigationItemAdded;
        if (evt != null)
            evt(this, new NavigationItemAddedEventArgs(item));
    }
    protected abstract Task DoLoad(CancellationToken token);
    public async Task Load()
    {
        if (State == NavigationItemState.Loading || State == NavigationItemState.Loaded)
            await Unload();
        var cancelationTokenSource = new CancellationTokenSource();
        await LoadWorkers.DoWork(() => DoLoadWorker(cancelationTokenSource.Token), cancelationTokenSource);
    }
    private async Task DoLoadWorker(CancellationToken token)
    {
        await UnloadWorkers.WaitCurrentWorkers();
        UnloadWorkers.Start();
        token.ThrowIfCancellationRequested();
        RaiseLoading();
        token.ThrowIfCancellationRequested();
        await DoLoad(token);
        token.ThrowIfCancellationRequested();
        RaiseLoaded();
    }
    protected abstract Task DoUnload();
    public Task Unload()
    {
        return UnloadWorkers.DoWork(() => DoUnloadWorker());
    }
    private async Task DoUnloadWorker()
    {
        await Task.Yield();
        RaiseUnloading();
        await LoadWorkers.WaitCurrentWorkers(cancelQueue: true);
        try
        {
            await DoUnload();
            RaiseUnloaded();
        }
        catch { throw; }
        finally
        {
            LoadWorkers.Start();
        }
    }
    ~NavigationItem()
    {
        this.Dispose();
    }

    public void Dispose()
    {
        if (LoadWorkers != null)
            LoadWorkers.Dispose();
        if (UnloadWorkers != null)
            UnloadWorkers.Dispose();
    }
}
public abstract class NavigationItem<T> : NavigationItem
{
    public NavigationItem(NavigationData data)
        : base(data)
    {
        Items = new ObservableCollection<T>();
    }
    public ObservableCollection<T> Items { get; private set; }
    protected abstract Task<IEnumerable<T>> GetItems(CancellationToken token);
    protected override async Task DoLoad(CancellationToken token)
    {
        var items = await GetItems(token);
        token.ThrowIfCancellationRequested();
        if (items != null)
        {
            DispatcherLocator.Dispatcher.Invoke(() =>
            {
                foreach (var item in items)
                {
                    Items.Add(item);
                    if (token.IsCancellationRequested)
                        break;
                }
            });
        }
    }
    protected override Task DoUnload()
    {
        DispatcherLocator.Dispatcher.Invoke(() => Items.Clear());
        return Task.FromResult(false);
    }
}

Lastly I have a lighter weight implementation of this queuing concept which is less generic and relies on two Tasks that block each-other to prevent critical logic from being executed simultaneously. (We don't want to Load and Unload at the same time). This uses a pure async/await pattern.

public abstract class NavigationItem : BindableBase
{
    public NavigationData Data { get; private set; }
    public string DisplayName
    {
        get { return Data.DisplayName; }
    }
    public bool IsCloseable
    {
        get { return Data.IsCloseable; }
    }
    public NavigationItem(NavigationData data)
    {
        Data = data;
        State = NavigationItemState.Unloaded;
        Close = new DelegateCommand(() =>
        {
            Task t = Unload();
            EventHandler closed = Closed;
            if (closed != null)
                closed(this, EventArgs.Empty);
        });
    }
    private Task LoadWorker { get; set; }
    private Task UnloadWorker { get; set; }
    private NavigationItemState state;
    public NavigationItemState State
    {
        get { return state; }
        set
        {
            if (SetProperty(ref state, value))
            {
                OnPropertyChanged(() => IsUnloading);
                OnPropertyChanged(() => IsUnloaded);
                OnPropertyChanged(() => IsLoading);
                OnPropertyChanged(() => IsLoaded);
            }
        }
    }
    public bool IsUnloading { get { return State == NavigationItemState.Unloading; } }
    public bool IsUnloaded { get { return State == NavigationItemState.Unloaded; } }
    public bool IsLoading { get { return State == NavigationItemState.Loading; } }
    public bool IsLoaded { get { return State == NavigationItemState.Loaded; } }
    public event EventHandler Unloading;
    public event EventHandler Unloaded;
    public event EventHandler Loading;
    public event EventHandler Loaded;
    public event EventHandler<NavigationItemAddedEventArgs> NavigationItemAdded;
    public ICommand Close { get; private set; }
    public event EventHandler Closed;
    private void RaiseUnloading()
    {
        State = NavigationItemState.Unloading;
        var evt = Unloading;
        if (evt != null)
            evt(this, EventArgs.Empty);
    }
    private void RaiseUnloaded()
    {
        State = NavigationItemState.Unloaded;
        var evt = Unloaded;
        if (evt != null)
            evt(this, EventArgs.Empty);
    }
    private void RaiseLoading()
    {
        State = NavigationItemState.Loading;
        var evt = Loading;
        if (evt != null)
            evt(this, EventArgs.Empty);
    }
    private void RaiseLoaded()
    {
        State = NavigationItemState.Loaded;
        var evt = Loaded;
        if (evt != null)
            evt(this, EventArgs.Empty);
    }
    protected void AddNavigationItem(NavigationItem item)
    {
        var evt = NavigationItemAdded;
        if (evt != null)
            evt(this, new NavigationItemAddedEventArgs(item));
    }
    private async Task WaitLoad()
    {
        if (LoadWorker != null)
        {
            try
            {
                await LoadWorker;
            }
            catch (OperationCanceledException) { } //task canceled
            catch { Task unloader = Unload(); }
        }
    }
    private async Task WaitUnload()
    {
        if (UnloadWorker != null)
            await UnloadWorker;
    }
    protected abstract Task DoLoad(CancellationToken token);
    public async Task Load()
    {
        await Task.Yield();
        if (State == NavigationItemState.Loading || State == NavigationItemState.Loaded)
            await Unload();
        else if(State == NavigationItemState.Unloading)
            await WaitUnload();
        RaiseLoading();
        LoadCancelationTokenSource = new CancellationTokenSource();
        LoadWorker = DoLoadWorker();
        await WaitLoad();
    }
    private CancellationTokenSource LoadCancelationTokenSource { get; set; }
    private CancellationToken LoadCancelationToken { get { return LoadCancelationTokenSource.Token; } }
    private async Task DoLoadWorker()
    {
        LoadCancelationToken.ThrowIfCancellationRequested();
        await DoLoad(LoadCancelationToken);
        LoadCancelationToken.ThrowIfCancellationRequested();
        RaiseLoaded();
    }
    protected abstract Task DoUnload();
    public async Task Unload()
    {
        await WaitUnload();
        RaiseUnloading();
        UnloadWorker = DoUnloadWorker();
        await WaitUnload();
    }
    private async Task DoUnloadWorker()
    {
        await Task.Yield();
        if (LoadWorker != null)
            LoadCancelationTokenSource.Cancel(true);
        await WaitLoad();
        await DoUnload();
        RaiseUnloaded();
    }
}
public abstract class NavigationItem<T> : NavigationItem
{
    public NavigationItem(NavigationData data)
        : base(data)
    {
        Items = new ObservableCollection<T>();
    }
    public ObservableCollection<T> Items { get; private set; }
    protected abstract Task<IEnumerable<T>> GetItems(CancellationToken token);
    protected override async Task DoLoad(CancellationToken token)
    {
        var items = await GetItems(token);
        token.ThrowIfCancellationRequested();
        if (items != null)
        {
            DispatcherLocator.Dispatcher.Invoke(() =>
            {
                foreach (var item in items)
                {
                    Items.Add(item);
                    if (token.IsCancellationRequested)
                        break;
                }
            });
        }
    }
    protected override Task DoUnload()
    {
        DispatcherLocator.Dispatcher.Invoke(() => Items.Clear());
        return Task.FromResult(false);
    }
}
\$\endgroup\$
2
\$\begingroup\$
TaskFactory

That's a pretty confusing name, since there already is a TaskFactory class in TPL, with different meaning.


new TransformBlock<TaskFactory, TaskFactory>(tf => tf)

You don't need a TransformBlock for this, you can use BufferBlock instead.


Action<Exception> handleException

I would expect handleException to be a bool. A better name for this would be exceptionHandler.


StartedTasks = DirtyWorkers.AsObservable();

Doesn't DirtyWorkers also contain workers that were canceled before they could be started?


try
{
    await sWait;
}
catch (OperationCanceledException) { throw; }

I you wrote just await sWait;, it would work the same.


I find the whole system of transformBlock.AsObservable().Subscribe(actionBlock.AsObserver()), where actionBlock checks for id, overcomplicated.

One way to simplify this is to have TaskFactory expose a Completed Task. You can then await it, or use ContinueWith() on it.


Instead of SemaphoreSlim that starts with 0, is awaited once and Released once, you could use TaskCompletionSource.


await Task.Yield();

what is the purpose of this? I don't see any reason for it here.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.