implab/ImplabNet Commit - r186:75103928da09

working on cancelation and error handling

cin -

r186:75103928da09 ref20160224

parent child

Context file:

r186:75103928da09

Collapse all files

Implab/AbstractEvent.cs +5 -10

             using System;
             using Implab.Parallels;
             using System.Threading;
             using System.Reflection;
             namespace Implab {
                 public abstract class AbstractEvent<THandler> : ICancellationToken, ICancellable {
                     const int UNRESOLVED_SATE = 0;
                     const int TRANSITIONAL_STATE = 1;
                     protected const int SUCCEEDED_STATE = 2;
                     protected const int REJECTED_STATE = 3;
                     protected const int CANCELLED_STATE = 4;
                     const int CANCEL_NOT_REQUESTED = 0;
                     const int CANCEL_REQUESTING = 1;
                     const int CANCEL_REQUESTED = 2;
                     const int RESERVED_HANDLERS_COUNT = 4;
                     int m_state;
                     Exception m_error;
                     int m_handlersCount;
                     //readonly THandler[] m_handlers = new THandler[RESERVED_HANDLERS_COUNT];
                     THandler[] m_handlers;
                     MTQueue<THandler> m_extraHandlers;
                     int m_handlerPointer = -1;
                     int m_handlersCommited;
                     int m_cancelRequest;
                     Exception m_cancelationReason;
                     MTQueue<Action<Exception>> m_cancelationHandlers;
                     #region state managment
                     bool BeginTransit() {
                         return UNRESOLVED_SATE == Interlocked.CompareExchange(ref m_state, TRANSITIONAL_STATE, UNRESOLVED_SATE);
                     }
                     void CompleteTransit(int state) {
                         if (TRANSITIONAL_STATE != Interlocked.CompareExchange(ref m_state, state, TRANSITIONAL_STATE))
                             throw new InvalidOperationException("Can't complete transition when the object isn't in the transitional state");
                     }
                     void WaitTransition() {
                         while (m_state == TRANSITIONAL_STATE) {
                             Thread.MemoryBarrier();
                         }
                     }
                     protected bool BeginSetResult() {
                         if (!BeginTransit()) {
                             WaitTransition();
                             if (m_state != CANCELLED_STATE)
                                 throw new InvalidOperationException("The promise is already resolved");
                             return false;
                         }
                         return true;
                     }
                     protected void EndSetResult() {
                         CompleteTransit(SUCCEEDED_STATE);
                         Signal();
                     }
                     /// <summary>
                     /// Выполняет обещание, сообщая об ошибке
                     /// </summary>
                     /// <remarks>
                     /// Поскольку обещание должно работать в многопточной среде, при его выполнении сразу несколько потоков
                     /// могу вернуть ошибку, при этом только первая будет использована в качестве результата, остальные
                     /// будут проигнорированы.
                     /// </remarks>
                     /// <param name="error">Исключение возникшее при выполнении операции</param>
                     /// <exception cref="InvalidOperationException">Данное обещание уже выполнено</exception>
                     protected void SetError(Exception error) {
-                        while (error is PromiseTransientException)
-                            error = error.InnerException;
-                        var isCancel = error is OperationCanceledException;
                         if (BeginTransit()) {
-                            m_error = isCancel ? error.InnerException : error;
+                            m_error = error;
-                            CompleteTransit(isCancel ? CANCELLED_STATE : REJECTED_STATE);
+                            CompleteTransit(REJECTED_STATE);
                             Signal();
                         } else {
                             WaitTransition();
-                            if (!isCancel || m_state == SUCCEEDED_STATE)
+                            if (m_state == SUCCEEDED_STATE)
                                 throw new InvalidOperationException("The promise is already resolved");
                         }
                     }
                     /// <summary>
                     /// Отменяет операцию, если это возможно.
                     /// </summary>
                     /// <remarks>Для определения была ли операция отменена следует использовать свойство <see cref="IsCancelled"/>.</remarks>
                     protected void SetCancelled(Exception reason) {
                         if (BeginTransit()) {
                             m_error = reason;
                             CompleteTransit(CANCELLED_STATE);
                             Signal();
                         }
                     }
                     protected abstract void SignalHandler(THandler handler, int signal);
                     void Signal() {
                         var hp = m_handlerPointer;
                         var slot = hp +1 ;
                         while (slot < m_handlersCommited) {
                             if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) == hp) {
                                 SignalHandler(m_handlers[slot], m_state);
                             }
                             hp = m_handlerPointer;
                             slot = hp +1 ;
                         }
                         if (m_extraHandlers != null) {
                             THandler handler;
                             while (m_extraHandlers.TryDequeue(out handler))
                                 SignalHandler(handler, m_state);
                         }
                     }
                     #endregion
                     protected abstract Signal GetResolveSignal();
                     #region synchronization traits
                     protected void WaitResult(int timeout) {
                         if (!(IsResolved || GetResolveSignal().Wait(timeout)))
                             throw new TimeoutException();
                         switch (m_state) {
                             case SUCCEEDED_STATE:
                                 return;
                             case CANCELLED_STATE:
-                                throw new OperationCanceledException();
+                                throw new OperationCanceledException("The operation has been cancelled", m_error);
                             case REJECTED_STATE:
                                 throw new TargetInvocationException(m_error);
                             default:
-                                throw new ApplicationException(String.Format("Invalid promise state {0}", m_state));
+                                throw new ApplicationException(String.Format("The promise state {0} is invalid", m_state));
                         }
                     }
                     #endregion
                     #region handlers managment
                     protected void AddHandler(THandler handler) {
                         if (m_state > 1) {
                             // the promise is in the resolved state, just invoke the handler
                             SignalHandler(handler, m_state);
                         } else {
                             var slot = Interlocked.Increment(ref m_handlersCount) - 1;
                             if (slot < RESERVED_HANDLERS_COUNT) {
                                 if (slot == 0) {
                                     m_handlers = new THandler[RESERVED_HANDLERS_COUNT];
                                 } else {
                                     while (m_handlers == null)
                                         Thread.MemoryBarrier();
                                 }
                                 m_handlers[slot] = handler;
                                 while (slot != Interlocked.CompareExchange(ref m_handlersCommited, slot + 1, slot)) {
                                 }
                                 if (m_state > 1) {
                                     do {
                                         var hp = m_handlerPointer;
                                         slot = hp + 1;
                                         if (slot < m_handlersCommited) {
                                             if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) != hp)
                                                 continue;
                                             SignalHandler(m_handlers[slot], m_state);
                                         }
                                         break;
                                     } while(true);
                                 }
                             } else {
                                 if (slot == RESERVED_HANDLERS_COUNT) {
                                     m_extraHandlers = new MTQueue<THandler>();
                                 } else {
                                     while (m_extraHandlers == null)
                                         Thread.MemoryBarrier();
                                 }
                                 m_extraHandlers.Enqueue(handler);
                                 if (m_state > 1 && m_extraHandlers.TryDequeue(out handler))
                                 // if the promise have been resolved while we was adding the handler to the queue
                                 // we can't guarantee that someone is still processing it
                                 // therefore we need to fetch a handler from the queue and execute it
                                 // note that fetched handler may be not the one that we have added
                                 // even we can fetch no handlers at all :)
                                     SignalHandler(handler, m_state);
                             }
                         }
                     }
                     #endregion
                     #region IPromise implementation
                     public bool IsResolved {
                         get {
                             Thread.MemoryBarrier();
                             return m_state > 1;
                         }
                     }
                     public bool IsCancelled {
                         get {
                             Thread.MemoryBarrier();
                             return m_state == CANCELLED_STATE;
                         }
                     }
                     #endregion
                     public Exception Error {
                         get {
                             return m_error;
                         }
                     }
                     public bool CancelOperationIfRequested() {
                         if (IsCancellationRequested) {
                             CancelOperation(CancellationReason);
                             return true;
                         }
                         return false;
                     }
                     public virtual void CancelOperation(Exception reason) {
                         SetCancelled(reason);
                     }
                     public void CancellationRequested(Action<Exception> handler) {
                         Safe.ArgumentNotNull(handler, "handler");
                         if (IsCancellationRequested)
                             handler(CancellationReason);
                         if (m_cancelationHandlers == null)
                             Interlocked.CompareExchange(ref m_cancelationHandlers, new MTQueue<Action<Exception>>(), null);
                         m_cancelationHandlers.Enqueue(handler);
                         if (IsCancellationRequested && m_cancelationHandlers.TryDequeue(out handler))
                             // TryDeque implies MemoryBarrier()
                             handler(m_cancelationReason);
                     }
                     public bool IsCancellationRequested {
                         get {
                             do {
                                 if (m_cancelRequest == CANCEL_NOT_REQUESTED)
                                     return false;
                                 if (m_cancelRequest == CANCEL_REQUESTED)
                                     return true;
                                 Thread.MemoryBarrier();
                             } while(true);
                         }
                     }
                     public Exception CancellationReason {
                         get {
                             do {
                                 Thread.MemoryBarrier();
                             } while(m_cancelRequest == CANCEL_REQUESTING);
                             return m_cancelationReason;
                         }
                     }
                     #region ICancellable implementation
                     public void Cancel() {
                         Cancel(null);
                     }
                     public void Cancel(Exception reason) {
                         if (CANCEL_NOT_REQUESTED == Interlocked.CompareExchange(ref m_cancelRequest, CANCEL_REQUESTING, CANCEL_NOT_REQUESTED)) {
                             m_cancelationReason = reason;
                             m_cancelRequest = CANCEL_REQUESTED;
                             if (m_cancelationHandlers != null) {
                                 Action<Exception> handler;
                                 while (m_cancelationHandlers.TryDequeue(out handler))
                                     handler(m_cancelationReason);
                             }
                         }
                     }
                     #endregion
                 }
             }

Implab/AbstractPromise.cs +1 -1

             using System;
             using Implab.Parallels;
             namespace Implab {
                 public abstract class AbstractPromise : AbstractEvent<AbstractPromise.HandlerDescriptor>, IPromise {
                     public struct HandlerDescriptor {
                         readonly Action m_handler;
                         readonly Action<Exception> m_error;
                         readonly Action<Exception> m_cancel;
                         readonly PromiseEventType m_mask;
                         public HandlerDescriptor(Action success, Action<Exception> error, Action<Exception> cancel) {
                             m_handler = success;
                             m_error = error;
                             m_cancel = cancel;
                             m_mask = PromiseEventType.Success;
                         }
                         public HandlerDescriptor(Action handler, PromiseEventType mask) {
                             m_handler = handler;
                             m_error = null;
                             m_cancel = null;
                             m_mask = mask;
                         }
                         public void SignalSuccess() {
                             if ((m_mask & PromiseEventType.Success) != 0 && m_handler != null) {
                                 try {
                                     m_handler();
                                 } catch (Exception err) {
                                     // avoid calling handler twice in case of error
                                     if (m_error != null)
                                         SignalError(err);
                                 }
                             }
                         }
                         public void SignalError(Exception err) {
                             if (m_error != null) {
                                 try {
                                     m_error(err);
                                     // Analysis disable once EmptyGeneralCatchClause
                                 } catch {
                                 }
                             } else if ((m_mask & PromiseEventType.Error ) != 0 && m_handler != null) {
                                 try {
                                     m_handler();
                                     // Analysis disable once EmptyGeneralCatchClause
                                 } catch {
                                 }
                             }
                         }
                         public void SignalCancel(Exception reason) {
                             if (m_cancel != null) {
                                 try {
                                     m_cancel(reason);
                                 } catch (Exception err) {
                                     SignalError(err);
                                 }
                             } else if ( (m_mask & PromiseEventType.Cancelled) != 0 && m_handler != null) {
                                 try {
                                     m_handler();
                                     // Analysis disable once EmptyGeneralCatchClause
                                 } catch {
                                 }
                             }
                         }
                     }
                     #region implemented abstract members of AbstractPromise
                     protected override void SignalHandler(HandlerDescriptor handler, int signal) {
                         switch (signal) {
                             case SUCCEEDED_STATE:
                                 handler.SignalSuccess();
                                 break;
                             case REJECTED_STATE:
                                 handler.SignalError(Error);
                                 break;
                             case CANCELLED_STATE:
                                 handler.SignalCancel(CancellationReason);
                                 break;
                             default:
                                 throw new InvalidOperationException(String.Format("Invalid promise signal: {0}", signal));
                         }
                     }
                     protected override Signal GetResolveSignal() {
                         var signal = new Signal();
                         On(signal.Set, PromiseEventType.All);
                         return signal;
                     }
                     #endregion
                     public Type PromiseType {
                         get {
                             return typeof(void);
                         }
                     }
                     public IPromise On(Action success, Action<Exception> error, Action<Exception> cancel) {
                         AddHandler(new HandlerDescriptor(success, error, cancel));
                         return this;
                     }
                     public IPromise On(Action success, Action<Exception> error) {
                         AddHandler(new HandlerDescriptor(success, error, null));
                         return this;
                     }
                     public IPromise On(Action success) {
                         AddHandler(new HandlerDescriptor(success, null, null));
                         return this;
                     }
                     public IPromise On(Action handler, PromiseEventType events) {
                         AddHandler(new HandlerDescriptor(handler,events));
                         return this;
                     }
                     public IPromise<T> Cast<T>() {
                         throw new InvalidCastException();
                     }
                     public void Join() {
                         WaitResult(-1);
                     }
                     public void Join(int timeout) {
                         WaitResult(timeout);
                     }
                     protected void SetResult() {
-                        BeginSetResult();
+                        if(BeginSetResult())
                             EndSetResult();
                     }
                 }
             }

Implab/ActionChainTaskBase.cs +24 -7

             using System;
             using System.Threading;
             namespace Implab {
                 public class ActionChainTaskBase : AbstractPromise {
                     readonly Func<Exception, IPromise> m_error;
                     readonly Func<Exception, IPromise> m_cancel;
                     int m_cancelationLock;
                     protected ActionChainTaskBase(Func<Exception, IPromise> error, Func<Exception, IPromise> cancel, bool autoCancellable) {
                         m_error = error;
                         m_cancel = cancel;
                         if (autoCancellable)
                             CancellationRequested(CancelOperation);
                     }
                     public void Reject(Exception error) {
                         if (LockCancelation())
                             HandleErrorInternal(error);
                     }
                     public override void CancelOperation(Exception reason) {
                         if (LockCancelation()) {
+                            if (!(reason is OperationCanceledException))
+                                reason = reason != null ? new OperationCanceledException(null, reason) : new OperationCanceledException();
                             if (m_cancel != null) {
                                 try {
-                                    m_cancel(reason).On(SetResult, SetError, SetCancelled);
+                                    m_cancel(reason).On(SetResult, HandleErrorInternal, HandleCancelInternal);
                                 } catch (Exception err) {
                                     HandleErrorInternal(err);
                                 }
                             } else {
-                                SetCancelled(reason);
+                                HandleErrorInternal(reason);
                             }
                         }
                     }
-                    protected void HandleErrorInternal(Exception error) {
+                    void HandleCancelInternal(Exception reason) {
+                        if (!(reason is OperationCanceledException))
+                            reason = reason != null ? new OperationCanceledException(null, reason) : new OperationCanceledException();
+                        HandleErrorInternal(reason);
+                    }
+                    void HandleErrorInternal(Exception error) {
                         if (m_error != null) {
                             try {
                                 var p = m_error(error);
                                 p.On(SetResult, SetError, SetCancelled);
                                 CancellationRequested(p.Cancel);
                             } catch (Exception err) {
-                                SetError(err);
+                                error = err;
                             }
                         } else {
+                            SetErrorInternal(error);
+                        }
+                    }
+                    void SetErrorInternal(Exception error) {
+                        while (error is PromiseTransientException)
+                            error = error.InnerException;
+                        if (error is OperationCanceledException)
+                            SetCancelled(error);
+                        else
                             SetError(error);
                     }
                     protected bool LockCancelation() {
                         return 0 == Interlocked.CompareExchange(ref m_cancelationLock, 1, 0);
                     }
                 }
             }

Implab/Components/RunnableComponent.cs +2 -9

             using System;
             namespace Implab.Components {
                 public abstract class RunnableComponent : IDisposable, IRunnable, IInitializable {
                     enum Commands {
                         Ok = 0,
                         Fail,
                         Init,
                         Start,
                         Stop,
                         Dispose,
                         Last = Dispose
                     }
                     class StateMachine {
                         static readonly ExecutionState[,] _transitions;
                         static StateMachine() {
                             _transitions = new ExecutionState[(int)ExecutionState.Last + 1, (int)Commands.Last + 1];
                             Edge(ExecutionState.Created, ExecutionState.Initializing, Commands.Init);
                             Edge(ExecutionState.Created, ExecutionState.Disposed, Commands.Dispose);
                             Edge(ExecutionState.Initializing, ExecutionState.Ready, Commands.Ok);
                             Edge(ExecutionState.Initializing, ExecutionState.Failed, Commands.Fail);
                             Edge(ExecutionState.Ready, ExecutionState.Starting, Commands.Start);
                             Edge(ExecutionState.Ready, ExecutionState.Disposed, Commands.Dispose);
                             Edge(ExecutionState.Starting, ExecutionState.Running, Commands.Ok);
                             Edge(ExecutionState.Starting, ExecutionState.Failed, Commands.Fail);
                             Edge(ExecutionState.Starting, ExecutionState.Stopping, Commands.Stop);
                             Edge(ExecutionState.Starting, ExecutionState.Disposed, Commands.Dispose);
                             Edge(ExecutionState.Running, ExecutionState.Failed, Commands.Fail);
                             Edge(ExecutionState.Running, ExecutionState.Stopping, Commands.Stop);
                             Edge(ExecutionState.Running, ExecutionState.Disposed, Commands.Dispose);
                             Edge(ExecutionState.Stopping, ExecutionState.Failed, Commands.Fail);
                             Edge(ExecutionState.Stopping, ExecutionState.Disposed, Commands.Ok);
                             Edge(ExecutionState.Failed, ExecutionState.Disposed, Commands.Dispose);
                         }
                         static void Edge(ExecutionState s1, ExecutionState s2, Commands cmd) {
                             _transitions[(int)s1, (int)cmd] = s2;
                         }
                         public ExecutionState State {
                             get;
                             private set;
                         }
                         public StateMachine(ExecutionState initial) {
                             State = initial;
                         }
                         public bool Move(Commands cmd) {
                             var next = _transitions[(int)State, (int)cmd];
                             if (next == ExecutionState.Undefined)
                                 return false;
                             State = next;
                             return true;
                         }
                     }
                     IPromise m_pending;
                     Exception m_lastError;
                     readonly StateMachine m_stateMachine;
                     protected RunnableComponent(bool initialized) {
                         m_stateMachine = new StateMachine(initialized ? ExecutionState.Ready : ExecutionState.Created);
                     }
                     protected virtual int DisposeTimeout {
                         get {
                             return 10000;
                         }
                     }
                     void ThrowInvalidCommand(Commands cmd) {
                         if (m_stateMachine.State == ExecutionState.Disposed)
                             throw new ObjectDisposedException(ToString());
                         throw new InvalidOperationException(String.Format("Commnd {0} is not allowed in the state {1}", cmd, m_stateMachine.State));
                     }
                     void Move(Commands cmd) {
                         if (!m_stateMachine.Move(cmd))
                             ThrowInvalidCommand(cmd);
                     }
                     void Invoke(Commands cmd, Action action) {
                         lock (m_stateMachine)
                             Move(cmd);
                         try {
                             action();
                             lock(m_stateMachine)
                                 Move(Commands.Ok);
                         } catch (Exception err) {
                             lock (m_stateMachine) {
                                 Move(Commands.Fail);
                                 m_lastError = err;
                             }
                             throw;
                         }
                     }
                     IPromise InvokeAsync(Commands cmd, Func<IPromise> action, Action<IPromise, IDeferred> chain) {
                         IPromise promise = null;
                         IPromise prev;
                         var task = new ActionChainTask(action, null, null, true);
                         lock (m_stateMachine) {
                             Move(cmd);
                             prev = m_pending;
                             promise = task.Then(
                                 () => {
                                     lock(m_stateMachine) {
                                         if (m_pending == promise) {
                                             Move(Commands.Ok);
                                             m_pending = null;
                                         }
                                     }
                                 }, e => {
                                     lock(m_stateMachine) {
                                         if (m_pending == promise) {
                                             Move(Commands.Fail);
                                             m_pending = null;
                                             m_lastError = e;
                                         }
                                     }
                                     throw new PromiseTransientException(e);
                                 },
                                 r => {
-                                    lock(m_stateMachine) {
+                                    // handle cancellation as exception
-                                        if (m_pending == promise) {
-                                            Move(Commands.Fail);
-                                            m_pending = null;
-                                            m_lastError = new OperationCanceledException("The operation has been cancelled", r);
                                     throw new OperationCanceledException("The operation has been cancelled", r);
                                 }
                             );
                             m_pending = promise;
                         }
                         if (prev == null)
                             task.Resolve();
                         else
                             chain(prev, task);
                         return promise;
                     }
                     #region IInitializable implementation
                     public void Init() {
                         Invoke(Commands.Init, OnInitialize);
                     }
                     protected virtual void OnInitialize() {
                     }
                     #endregion
                     #region IRunnable implementation
                     public IPromise Start() {
                         return InvokeAsync(Commands.Start, OnStart, null);
                     }
                     protected virtual IPromise OnStart() {
                         return Promise.SUCCESS;
                     }
                     public IPromise Stop() {
                         return InvokeAsync(Commands.Stop, OnStop, StopPending).Then(Dispose);
                     }
                     protected virtual IPromise OnStop() {
                         return Promise.SUCCESS;
                     }
                     /// <summary>
                     /// Stops the current operation if one exists.
                     /// </summary>
                     /// <param name="current">Current.</param>
                     /// <param name="stop">Stop.</param>
                     protected virtual void StopPending(IPromise current, IDeferred stop) {
                         if (current == null) {
                             stop.Resolve();
                         } else {
-                            current.On(stop.Resolve, stop.Reject, stop.CancelOperation);
+                            current.On(stop.Resolve, stop.Reject, e => stop.Resolve());
                             current.Cancel();
                         }
                     }
                     public ExecutionState State {
                         get {
                             return m_stateMachine.State;
                         }
                     }
                     public Exception LastError {
                         get {
                             return m_lastError;
                         }
                     }
                     #endregion
                     #region IDisposable implementation
                     public void Dispose() {
                         IPromise pending;
                         lock (m_stateMachine) {
                             if (m_stateMachine.State == ExecutionState.Disposed)
                                 return;
                             Move(Commands.Dispose);
                             GC.SuppressFinalize(this);
                             pending = m_pending;
                             m_pending = null;
                         }
                         if (pending != null) {
                             pending.Cancel();
                             pending.Timeout(DisposeTimeout).On(
                                 () => Dispose(true, null),
                                 err => Dispose(true, err),
                                 reason => Dispose(true, new OperationCanceledException("The operation is cancelled", reason))
                             );
                         } else {
                             Dispose(true, m_lastError);
                         }
                     }
                     ~RunnableComponent() {
                         Dispose(false, null);
                     }
                     #endregion
                     protected virtual void Dispose(bool disposing, Exception lastError) {
                     }
                 }
             }

General Comments 0

Write
Preview

You need to be logged in to leave comments. Login now

No TODOs yet

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages