9
\$\begingroup\$

Beeing on .NET 3.5 i don't have access to the TPL. Yet i have become fed up of having to manage manually the logic behind delegate.BeginInvoke type of scenarios each time and i set up to implement my own Task class.

Functionality it should support:

  • starting a parallel execution (using a thread pool thread)
  • capable of Waiting on the result with a specific timeout
  • capable of aborting said execution
  • should provide event for callbacks to attach to

This is my implementation. Can you guys please review it and specify if there are any issues i might not have considered or maybe more effective ways to implement some things?

public class Task<T>
{
    #region Members
    private IAsyncResult _async;
    private Func<T> _action;
    private Func<T> _innerAction;
    private WaitHandle _waitHandle;
    private Thread _thread;
    private object _completedLock = new object();
    private object _abortLock = new object();
    private T _result;
    private bool _endCalled = false;
    #endregion

    #region Properties
    public object Tag { get; private set; }
    public bool IsCompleted { get; private set; }
    public bool IsRunning { get; private set; }

    public T Result
    {
        get
        {
            if (!_endCalled)
            {
                lock (_completedLock)
                {
                    if (!_endCalled)
                    {
                        try
                        {
                            if (_async != null)
                            {
                                _result = _innerAction.EndInvoke(_async);
                                IsCompleted = true;
                            }
                        }
                        finally
                        {
                            _endCalled = true;
                        }
                    }
                }
            }
            return _result;
        }
    }
    #endregion

    #region Events
    public event EventHandler Completed;
    #endregion

    public Task(Func<T> action, object tag)
        : this(action)
    {
        Tag = tag;
    }

    public Task(Func<T> action)
    {
        _action = action;
        _innerAction = () =>
        {
            try
            {
                _thread = Thread.CurrentThread;
                var result = _action();
                return result;
            }
            finally
            {
                lock (_abortLock)
                {
                    IsRunning = false;
                }
            }
        };
    }

    #region Public Methods
    public void Run()
    {
        if (!IsRunning || IsCompleted)
        {
            lock (_completedLock)
            {
                if (!IsRunning || IsCompleted)
                {
                    ResetState();
                    _async = _innerAction.BeginInvoke(obj => OnCompleted(), null);
                    IsRunning = true;
                    return;
                }
            }
        }
        throw new InvalidOperationException("Task is already running");
    }

    public bool WaitForCompletion(TimeSpan timeout)
    {
        if (IsRunning && !IsCompleted)
        {
            lock (_completedLock)
            {
                if (!IsCompleted)
                {
                    _waitHandle = _async.AsyncWaitHandle;
                    return _waitHandle.WaitOne(timeout);
                }
            }
        }
        return IsCompleted;
    }

    public bool WaitForCompletion(int timeoutMilliseconds)
    {
        return WaitForCompletion(TimeSpan.FromMilliseconds(timeoutMilliseconds));
    }

    public bool Abort()
    {
        bool result = false;
        if (!IsCompleted)
        {
            lock (_abortLock)
            {
                if (!IsCompleted && IsRunning)
                {
                    if (_thread != null)
                    {
                        _thread.Abort();
                    }
                    result = true;
                }
                ResetState();
            }
        }
        return result;
    }
    #endregion

    private void ResetState()
    {
        _async = null;
        _endCalled = false;
        _result = default(T);
        _thread = null;
        IsCompleted = false;
        IsRunning = false;
    }

    private void OnCompleted()
    {
        lock (_completedLock)
        {
            IsCompleted = true;
            if (_waitHandle != null)
            {
                _waitHandle.Close();
                _waitHandle = null;
            }
        }

        if (Completed != null)
        {
            try
            {
                Completed(this, EventArgs.Empty);
            }
            catch
            {
                //We swallow this as there is no way to catch it at an upper level
                //on the execution thread and do something about it.
                //Callbacks should not throw anyway.
            }
        }
    }
}
\$\endgroup\$
5
\$\begingroup\$
  • You may want to consider replacing the flags (IsRunning/IsCompleted) with a single state enum.

  • Some state checks use _abortLock while others use _completedLock.

    1. If Abort() is called on a Task that has not started,
    2. a context switch happens just before ResetState() is called,
    3. a second thread calls Run() and the new inner thread starts,
    4. another context switch happens and the first thread resumes to call ResetState()

    Yeah, a few moons would need to be in alignment for this to happen ... but still.

  • If Run() is called followed by Abort() before the inner thread actually starts, then _thread could be null resulting the thread not actually being aborted and inconsistent state. (again with the moons.)

  • Abort will not terminate the thread immediately, you should wait for it to complete before resetting the state.

  • I'm not sure aborting a thread pool thread is a good idea.

  • You set Tag in one of the constructors, but I don't see it being used anywhere.

\$\endgroup\$
  • \$\begingroup\$ thank for the suggestions. Some answers to your questions: \$\endgroup\$ – AZ01 May 13 '11 at 10:22
  • \$\begingroup\$ I am very well aware of the non deterministic nature of Abort(). It should not be used anyway too often. Aborting a threadPool thread is ok in my tests as the thread gets recycled and started again for a new execution just fine. The only difficulty is making sure you abort it while it's running your task and not some other thing scheduled by the pool. I am trying to be sure of that with the locks on _abortLock in the finally clause of the delegate and the Abort method \$\endgroup\$ – AZ01 May 13 '11 at 10:32
  • \$\begingroup\$ on your specific case i think i need to add an lock(_abortLock) to run in addition to the present _completedLock. That should take care of this scenarion \$\endgroup\$ – AZ01 May 13 '11 at 10:37
  • \$\begingroup\$ the Tag property is there so i can mark a task with some identifier and be able to differentiate it from other similar tasks in the same collection. \$\endgroup\$ – AZ01 May 13 '11 at 10:40
  • \$\begingroup\$ @AZ01, It might be better to just use the one lock object for both "_completedLock" and "_abortLock" (and just give it a more generic name). Regarding the Tag, I had thought it was intended for use by the inner thread but saw no easy way for the inner thread to access it. If its for the organization of the tasks then it should be fine. \$\endgroup\$ – Brian Reichle May 14 '11 at 12:07
3
\$\begingroup\$

Maybe this link helps you:

If you download Reactive Extensions for .NET 3.5SP1 from http://msdn.microsoft.com/en-us/devlabs/ee794896.aspx, you'll find a System.Threading.dll assembly... that contains TPL for .NET 3.5.

PS: The Rx page has moved to: http://msdn.microsoft.com/en-us/data/gg577609

\$\endgroup\$
  • 2
    \$\begingroup\$ -1 thanks for the links but my request was for review of the existing code \$\endgroup\$ – AZ01 May 13 '11 at 10:06

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.