5
\$\begingroup\$

My app saves data to a file - I obviously don't want the app to write to the file from multiple locations on disk. What I want is a generic "utility" class that will know how to run a piece of code (let's call it SingletonTask) with the following rules:

  1. Only one SingletonTask will run at any given moment in the system - EDIT: note that this singleton task may be asynchronous in nature and may require execution on the UI thread.
  2. When multiple calls are made to run the SingletonTask, the first one will immediately run the SingletonTask and every subsequent call that happens before SingletonTask runs will "queue up".
  3. All calls that are "queued up" during the run of SingletonTask will cause only a single new SingletonTask to execute (and all of them will complete when that one SingletonTask completes)
  4. Obviously, if during the SingletonTask executed in (3), more calls come in, they will follow rule (2) and so on.

I could not find anything like this in .NET (looking for something that works on Win8) - but maybe I don't know what the correct keywords are.

In any case - here's my attempt at doing this.. I have a utility class called SigletonTask you inherit from. That class has a protected abstract method called RunProcessAsync(). The method users call is RunAsync() and it behaves as described in the above rules.

I tested this and it seems to work, however, there is an issue with it: It's currently limiting usage to from within the UI thread. This makes the solution simpler, but means you cannot use it from arbitrary threads. I looked into locking inside the method, but could not come up with something elegant that does not deadlock.

My questions are:

  1. Can you see any issues with the code?
  2. Can you make the code better?

Here's the SingletonTask class:

(EDIT: This is a new version. Old version at the bottom.)

(Also - it's based on cold tasks)

public abstract class SingletonTask
{
    private Task m_runningTask = null;
    private Task m_nextTask = null;
    private object m_lock = new object();

    public SingletonTask()
    {
    }

    public Task RunAsync()
    {
        return RunAsync(() => RunProcessAsync());
    }

    public async Task RunAsync(Func<Task> taskGetter)
    {
        Task task1 = null;
        Task task2 = null;
        ColdTask start1 = null;
        ColdTask start2 = null;
        lock (m_lock)
        {
            if (m_runningTask == null)
            {
                start1 = new ColdTask(taskGetter);
                Task innerTask = start1.GetTask();
                m_runningTask = innerTask.ContinueWith(x => { lock (m_lock) { m_runningTask = null; } });
                task1 = m_runningTask;
            }
            else
            {
                task1 = m_runningTask;

                if (m_nextTask == null)
                {
                    start2 = new ColdTask(taskGetter);
                    Task innerTask = start2.GetTask();
                    m_nextTask = innerTask.ContinueWith(x => { lock (m_lock) { m_nextTask = null; } });
                }

                task2 = m_nextTask;
            }
        }

        if (start1 != null)
        {
            start1.Start().RunWithoutWarning();
        }

        if (task1 != null)
        {
            await task1;
        }

        if (start2 != null)
        {
            start2.Start().RunWithoutWarning();
        }

        if (task2 != null)
        {
            await task2;
        }

    }


    protected virtual Task RunProcessAsync()
    {
        return null;
    }
}

And here's how you inherit from it:

class MySing : SingletonTask
{
    private int m_id = 1000;
    Action<string> log;
    public MySing(Action<string> log, CoreDispatcher dispatcher) : base(dispatcher)
    {
        this.log = log;
    }
    protected override async Task RunProcessAsync()
    {
        int id = m_id++;
        log("Before delay for " + id.ToString());
        await Task.Delay(4000);
        log("After delay for " + id.ToString());
    }
}

My test app is fairly simple - there's a button that calls into an instance of MySing and awaits the result, logging before awaiting and after awaiting:

private void Button_Click_1(object sender, RoutedEventArgs e)
{
    int id = m_id++;
    Log("Running with id=" + id.ToString());
    await m_sing.RunAsync();
    Log("Run returned with id=" + id.ToString());
}

Tap the button 3 times quickly, for example, and you should see:

Running with id=0

Before delay for 1000

runnjng with id=1

Running with id=2

After delay for 1000

Run returned with id=0

Before delay for 1001

After delay for 1001

Run returned with id=1

Run returned with id=2

(I hope asking such a question is kosher here)

------------------------- OLD -------------------------------

(This is the code for my original question, before the edit.)

public abstract class SingletonTask
{
    private Task m_runningTask = null;
    private Task m_nextTask = null;

    private object m_lock = new object();
    private CoreDispatcher m_dispatcher = null;

    public SingletonTask(CoreDispatcher dispatcher)
    {
        m_dispatcher = dispatcher;
    }

    public async Task RunAsync()
    {
        if (!m_dispatcher.HasThreadAccess)
        {
            throw new UnauthorizedAccessException("Unauthorized thread access");
        }

        Task firstRun = null;
        if (m_runningTask == null)
        {
            Task innerRun = RunProcessAsync();
            m_runningTask = innerRun.ContinueWith(x => m_runningTask = null);
            firstRun = innerRun;
        }
        else if (m_nextTask == null)
        {
            await m_runningTask;
            if (m_runningTask == null)
            {
                Task innerRun = RunProcessAsync();
                m_runningTask = innerRun.ContinueWith(x => m_runningTask = null);
            }

            firstRun = m_runningTask;
        }

        if (firstRun != null)
        {
            await firstRun;
        }
    }

    protected abstract Task RunProcessAsync();
}
\$\endgroup\$

migrated from stackoverflow.com Oct 3 '12 at 12:08

This question came from our site for professional and enthusiast programmers.

  • \$\begingroup\$ Found anything neater? \$\endgroup\$ – Şafak Gür Oct 18 '12 at 13:58
  • \$\begingroup\$ Updated. I think I found something I am happy with. Had to implement Cold tasks for it (link just above the new code). This seems to work great and works from any thread. \$\endgroup\$ – Shahar Prish Oct 18 '12 at 14:13
7
\$\begingroup\$

I made something like this before but I don't claim that it's better.
Although it's much simpler:

public sealed class SingleTask
{
    private Task _Task;
    private readonly object _Lock = new object();

    // You can overload this with TaskCreationOptions, TaskScheduler etc.
    public SingleTask() { _Task = Task<object>.FromResult(null); }

    // You can overload Queue methods with TaskContinuationOptions,
    // TaskScheduler, CancellationToken etc.
    public Task Queue(Action action)
    {
        lock (_Lock)
            return _Task = _Task.ContinueWith(t => action.Invoke());
    }
    public Task<T> Queue<T>(Func<T> function)
    {
        lock (_Lock)
        {
            _Task = _Task.ContinueWith<T>(t => function.Invoke());
            return _Task as Task<T>;
        }
    }
}

You can use ContinueWith methods for queuing actions so I don't think you need a custom approach here. Here the Queue methods also return task objects so you can wait them individually.

You can make this class static if you want it to behave like a singleton.
But I can't see a reason for doing so.

As for your code, I don't see any issues with it.

Edit (Usage)

Let's say you have these methods you want to run asynchronously:

private void LongInitialization() { /* Some operation */ }
private int LongCalculation() { /* Some operation */ }

You can:

// Creates a completed task.
var single = new SingleTask();

// Starts initialization and returns its task.
Task initializeTask = single.Queue(LongInitialization);

// Queues LongCalculation to run after LongInitialization is completed.
Task<int> calculateTask = task.Queue(LongCalculation);

In these samples however, you can't interact with UI controls because you can't access a control from a thread other than the one the control was created on. So if you have a method like this:

private void UpdateText() { txtStatus.Text = GetStatusFromDB(); }

...which interacts with a TextBox, and if you call single.Queue(UpdateText), you'll see an exception is thrown with the message: "Cross-thread operation not valid".

What you can do here to prevent this, is providing an overload to the SingleTask's Queue(Action) method that takes a TaskScheduler as a parameter:

public Task Queue(Action action, TaskScheduler scheduler)
{
    lock (_Lock)
        return _Task = _Task.ContinueWith(t => action.Invoke(), scheduler);
}

Then you can call it like this:

single.Queue(UpdateText, TaskScheduler.FromCurrentSynchronizationContext());

By calling Queue like this on a UI thread you say "Hey! Here is the UI's synchronization context, call its Post method so it won't break my program".

You also can provide overloads to Queue so it can take TaskContinuationOptions (if you want to specify LongRunning and make it run on a different thread, etc.) and/or CancellationToken parameters (if you want to provide cancellation). For example:

public Task Queue(Action<CancellationToken> action,
    TaskScheduler scheduler,
    TaskContinuationOptions options,
    CancellationToken token)
{
    lock (_Lock)
        return _Task = _Task
            .ContinueWith(t => action.Invoke(token), token, options, scheduler);
}

You can share the same SingleTask insance anywhere you want to do things asynchronously but in an order so, I hope this will satisfy your goals. Again, if you want to have only one instance of this, you can always make this class static.

\$\endgroup\$
  • \$\begingroup\$ So in this solution dead locking could occur if somebody were to await on the returned Task forever, though a bug, still a realistic scenario right? \$\endgroup\$ – Mike Perrenoud Oct 3 '12 at 12:07
  • \$\begingroup\$ Hey! Thanks for the reply. So - the issue I am seeing here is that your functions are not async themselves - you are going to run them on a separate thread but if, say, they need the UI thread, or if your task is simply of an asynchronous nature you need to wrap it in a Func<> or Action and do .Wait() on it right? Updated my original question. \$\endgroup\$ – Shahar Prish Oct 3 '12 at 19:09
  • \$\begingroup\$ @ŞafakGür That's great - but again - what I meant is that the action you are scheduling is not async by itself. You could, for example, in the action, type the following: () => DoSomethingAsync.Wait(), however, that means you have a whole thread that's "taken" waiting for the operation to complete. I want the action you execute itself to be Asynchronous - not the method that executes the action. Note how in my implementation, RunProcessAsync() is itself an async operation). \$\endgroup\$ – Shahar Prish Oct 4 '12 at 7:43
  • \$\begingroup\$ @Shahar: So you want to queue operations that are already asynchronous I guess? Alright then, let me look that again :) \$\endgroup\$ – Şafak Gür Oct 4 '12 at 7:49
  • 1
    \$\begingroup\$ @Shahar: I see your point; async/await thingy already does it by sending the rest of the method as a delegate to the awaitable to invoke when its job is done and provide synchronization to asynchronous operations. I don't think I can add much to your wrapper in this case. You may want to change m_id++ to Interlocked.Increment(ref m_id) but other than that, your approach seems solid enough to me. I may need something like this, too so I'll look more into it when I can. Hope you'll end up with something neater. \$\endgroup\$ – Şafak Gür Oct 4 '12 at 8:33
3
\$\begingroup\$

Okay, after much updating I effectively have two approaches. Both of them use a Queue, but one uses the new ConcurrentQueue.

Old Fashioned Queue

public class Scheduler
{
    private BackgroundWorker _processor = new BackgroundWorker();

    private Queue<Action> ScheduledTasks { get; set; }

    public Scheduler()
    {
        _processor.WorkerSupportsCancellation = true;
        _processor.DoWork += (s, args) =>
        {
            while (!_processor.CancellationPending)
            {
                Queue.Synchronized(this.ScheduledTasks).Dequeue()();
            }
        }

        _processor.RunWorkerAsync();
    }

    public void ScheduleTask(Action task)
    {
        Queue.Synchronized(this.ScheduledTasks).Enqueue(task);
    }
}

New Concurrent Queue

public class Scheduler
{
    private BackgroundWorker _processor = new BackgroundWorker();

    private ConcurrentQueue<Action> ScheduledTasks { get; set; }

    public Scheduler()
    {
        _processor.WorkerSupportsCancellation = true;
        _processor.DoWork += (s, args) =>
        {
            while (!_processor.CancellationPending)
            {
                Action task = null;
                if (this.ScheduledTasks.TryDequeue(out task))
                {
                    task();
                }
            }
        }

        _processor.RunWorkerAsync();
    }

    public void ScheduleTask(Action task)
    {
        this.ScheduledTasks.Enqueue(task);
    }
}
\$\endgroup\$
  • \$\begingroup\$ Is there a particular reason why you don't use a ConcurrentQueue? \$\endgroup\$ – Şafak Gür Oct 3 '12 at 11:48
  • \$\begingroup\$ @ŞafakGür, nope. I've just never used that class before and didn't even know it existed. \$\endgroup\$ – Mike Perrenoud Oct 3 '12 at 11:51
  • \$\begingroup\$ @ŞafakGür, I have updated my answer to show use with the ConcurrentQueue(T). \$\endgroup\$ – Mike Perrenoud Oct 3 '12 at 11:59
  • \$\begingroup\$ It's part of the thread-safe collections that came with the .NET 4 (They're in System.Collections.Concurrent namespace). They are lock-free and highly suited for concurrent operations like this. Might perform better (I'm not sure though). Edit: Just saw your edit. \$\endgroup\$ – Şafak Gür Oct 3 '12 at 11:59
  • 1
    \$\begingroup\$ @ŞafakGür, and this is why I use Stack Overflow. I don't know everything, and so it's great when I learn new things like this! \$\endgroup\$ – Mike Perrenoud Oct 3 '12 at 12:04
0
\$\begingroup\$

Wow, there's a lot of complex answers here.

First, if you accept the restriction of UI-thread only, this is quite simple:

public abstract class SingletonTask
{
  private Task _runningTask = null;

  // Must be called from UI thread.
  public Task RunAsync()
  {
    if (_runningTask != null)
      return _runningTask;
    _runningTask = StartNewRunAsync();
    return _runningTask;
  }

  private async Task StartNewRunAsync()
  {
    try { await RunProcessAsync(); }
    finally { _runningTask = null; }
  }

  protected abstract Task RunProcessAsync();
}

The code above uses the UI thread as a synchronization mechanism. So, _runningTask is always written to from the UI thread (and read from there as well). The coalescing of tasks is handled by just keeping a single "current" instance.

Extending this to be multithreaded is a bit trickier:

public abstract class SingletonTask
{
  private Task _runningTask = null;
  private object _mutex = new object();

  public Task RunAsync()
  {
    Task result;
    TaskCompletionSource<object> tcs = null;
    lock (_mutex)
    {
      if (_runningTask != null)
        return _runningTask;
      tcs = new TaskCompletionSource<object>();
      _runningTask = result = StartNewRunAsync(tcs.Task);
    }
    tcs.SetResult(null);
    return result;
  }

  private async Task StartNewRunAsync(Task start)
  {
    await start;
    try { await RunProcessAsync(); }
    finally { lock (_mutex) _runningTask = null; }
  }

  protected abstract Task RunProcessAsync();
}

In this case, we need a _mutex for mutual exclusion. The other tricky part is that we don't want to call RunProcessAsync while holding a lock (as a general rule, you should never invoke arbitrary code while holding a lock). So, I use a "signal" (tcs) to get the Task from StartNewRunAsync before it actually invokes RunProcessAsync. That way I can update _runningTask still within the lock and then tell it to really start running at the tcs.SetResult(null) line. When it completes, it just sets _runningTask to null, so the next time RunAsync takes the lock, it'll start a new one.

\$\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.