Simple asynchronous operation implementation

Question

Could you please see if they are no bugs in this implementation? I am especially interested in knowing if there could be here any multithreading problems (such as race conditions).

It is written in C# for .NET 3.5.

public interface IAsyncOperation
{
 OperationToken Token { get; }
 bool IsCompleted { get; }
 Exception Exception { get; }
 bool Wait(int timeout);
}
public interface IAsyncOperation<out TResult> : IAsyncOperation
{
 TResult Result { get; }
}
/// <summary>
/// Asynchronous operation that may be executed on a separate thread.
/// IsCompleted and Exception would be synchronized after invoking Wait.
/// </summary>
public abstract class AsyncOperation : IAsyncOperation
{
 protected readonly object CompletionLock = new object();
 private volatile bool isCompleted; // volatile in order not to change the order of writing Exception and Result
 protected AsyncOperation(bool isCompleted = false)
 {
 IsCompleted = isCompleted;
 Token = OperationToken.New();
 }
 public OperationToken Token { get; protected set; }
 public bool IsCompleted
 {
 get { return isCompleted; }
 protected set { isCompleted = value; }
 }
 public Exception Exception { get; protected set; }
 public bool Wait(int timeout)
 {
 if (!IsCompleted)
 {
 lock (CompletionLock)
 {
 if (!IsCompleted)
 {
 // when migrated to .NET 4.5 then implement with ManualResetEventSlim
 Monitor.Wait(CompletionLock, timeout);
 }
 }
 }
 return IsCompleted;
 }
 protected void Complete()
 {
 if (IsCompleted)
 {
 throw new InvalidOperationException("The operation was already completed");
 }
 lock (CompletionLock)
 {
 if (IsCompleted)
 {
 throw new InvalidOperationException("The operation was already completed");
 }
 IsCompleted = true;
 Monitor.PulseAll(CompletionLock);
 }
 }
 protected void Complete(Exception exception)
 {
 if (!IsCompleted)
 {
 lock (CompletionLock)
 {
 if (!IsCompleted)
 {
 Exception = exception;
 IsCompleted = true;
 Monitor.PulseAll(CompletionLock);
 }
 }
 }
 }
}
public abstract class AsyncOperation<TResult> : AsyncOperation, IAsyncOperation<TResult>
{
 protected AsyncOperation(bool isCompleted = false)
 : base(isCompleted)
 {
 }
 public TResult Result { get; private set; }
 protected void Complete(TResult result)
 {
 if (!IsCompleted)
 {
 lock (CompletionLock)
 {
 if (!IsCompleted)
 {
 Result = result;
 IsCompleted = true;
 Monitor.PulseAll(CompletionLock);
 }
 }
 }
 }
}
/// <summary>
/// Invokes the action on the ThreadPool.
/// </summary>
public class PooledOperation : AsyncOperation
{
 private readonly Action<OperationToken> action;
 private bool isStarted;
 public PooledOperation(Action<OperationToken> action)
 {
 this.action = action;
 ThreadPool.QueueUserWorkItem(Execute);
 }
 public static IAsyncOperation<TResult> Run<TResult>(Func<OperationToken, TResult> function)
 {
 var result = new PooledOperation<TResult>(function);
 result.Start();
 return result;
 }
 public static IAsyncOperation Run(Action<OperationToken> action)
 {
 var result = new PooledOperation(action);
 result.Start();
 return result;
 }
 public void Start()
 {
 if (!isStarted)
 {
 isStarted = ThreadPool.QueueUserWorkItem(Execute);
 }
 }
 private void Execute(object state)
 {
 try
 {
 action(Token);
 Complete();
 }
 catch (Exception ex)
 {
 Complete(ex);
 }
 }
}
public class PooledOperation<TResult> : AsyncOperation<TResult>
{
 private readonly Func<OperationToken, TResult> function;
 public PooledOperation(Func<OperationToken, TResult> function)
 {
 this.function = function;
 }
 public void Start()
 {
 if (!IsCompleted)
 {
 ThreadPool.QueueUserWorkItem(Execute);
 }
 }
 private void Execute(object state)
 {
 try
 {
 TResult result = function(Token);
 Complete(result);
 }
 catch (Exception ex)
 {
 Complete(ex);
 }
 }
}
/// <summary>
/// Used in own ThreadPool-like implementation.
/// </summary>
internal class AsyncCommand : AsyncOperation
{
 private readonly Action<OperationToken> command;
 public AsyncCommand(Action<OperationToken> command)
 {
 command.ThrowIfNull("command");
 this.command = command;
 }
 public void Execute()
 {
 if (IsCompleted)
 {
 throw new InvalidOperationException("The operation was already completed");
 }
 try
 {
 command(Token);
 Complete();
 }
 catch (Exception ex)
 {
 Complete(ex);
 }
 }
}

• Mathieu Guindon: As far as you can tell, does the code seem to work as expected?
• Pellared: Yes, it looks like it is working fine.
• Mathieu Guindon: Good! I was just making sure, since the site's scope isn't to debug non-working code (see help center for more info).

Answer 1

1. For this:

   protected void Complete()
   {
    if (!IsCompleted)
    {
    lock (CompletionLock)
    {
    if (!IsCompleted)
    {
    IsCompleted = true;
    Monitor.PulseAll(CompletionLock);
    }
    }
    }
   }
   

   I guess if a client/subclass calls Complete() twice it's an error in the client code. You may want to fail early and throw an exception. (The Pragmatic Programmer: From Journeyman to Master by Andrew Hunt and David Thomas: Dead Programs Tell No Lies.)

   Inverting the first condition and using a guard clause also would make the code more flatten:

   protected void Complete()
   {
    if (IsCompleted)
    {
    throw new ...;
    }
    lock (CompletionLock)
    {
    if (IsCompleted)
    {
    throw new ...;
    }
    IsCompleted = true;
    Monitor.PulseAll(CompletionLock);
    }
   }
   

   You could extract a CheckNotCompleted method here to remove some duplicateion:

   protected void Complete()
   {
    CheckNotCompleted();
    lock (CompletionLock)
    {
    CheckNotCompleted();
    IsCompleted = true;
    Monitor.PulseAll(CompletionLock);
    }
   }
   private void CheckNotCompleted() 
   {
    if (IsCompleted)
    {
    throw new ...;
    }
   }
   

2. I'd consider using composition instead of inheritance. Effective Java, Second Edition, Item 16: Favor composition over inheritance is a good resource about that.

3. The code reminds me Futures and Executors from Java.

   In Java, you can create an ExecutorService with Executors.newFixedThreadPool() (and with many other methods).

   int threadNumber = 1;
   ExecutorService executorService = Executors.newFixedThreadPool(threadNumber);
   

   Then you can submit a task instance which implements Runnable or Callable which will be run on another thread and you get a Future instance immediately:

   Future future = executorService.submit(task);
   

   Future has isDone() and get(timeout, timeUnit) methods which are very similar to your IsCompleted and Wait methods.

   I feel that I've lost a little bit in the C# classes (I'm not too familiar with C#) but I hope the design of Future could give you some ideas.

• Pellared: Thanks for good comments! 1. I was thinking of it by my own and now I am sure that it is a good idea. 2. I agree with the concept but I do not know how I could implement it using composition with having good encapsulation (I do not want the Complete method to be public). I have subclasses that for example has an additional generic TResult Result property or implementation using Thread, ThreadPool, custom own ThreadPool. Do you want me to post more code for the review?
• palacsint: @Pellared: Feel free to share more code. public does not hurt, you can have an instance in a private field which is not known by any other class, so noone could call its public methods (so it's safe).
• Pellared: I have added the code. Thanks for involvement.
• palacsint: @Pellared: Answer updated, I hope it helps.