implab/ImplabNet Commit - r10:aa33d0bb8c0c

implemeted new cancellable promises concept

cin -

r10:aa33d0bb8c0c promises

parent child

Context file:

r10:aa33d0bb8c0c

Collapse all files

Implab.Test/PromiseHelper.cs created 644 +16 0

@@ -0,0 +1,16
	1	using System;
	2	using System.Collections.Generic;
	3	using System.Linq;
	4	using System.Text;
	5	using System.Threading;
	6
	7	namespace Implab.Test {
	8	class PromiseHelper {
	9	public static Promise<T> Sleep<T>(int timeout, T retVal) {
	10	return AsyncPool.Invoke(() => {
	11	Thread.Sleep(timeout);
	12	return retVal;
	13	});
	14	}
	15	}
	16	}

.hgignore +1 0

             Implab.Fx/bin/
             Implab.Fx.Test/bin/
             Implab.Fx.Test/obj/
+            _ReSharper.Implab/

Implab.Test/AsyncTests.cs +31 -5

@@ -1,10 +1,9
1	using System;	1	using System;
2	using Microsoft.VisualStudio.TestTools.UnitTesting;	2	using Microsoft.VisualStudio.TestTools.UnitTesting;
3	using Implab;
4	using System.Reflection;	3	using System.Reflection;
5	using System.Threading;	4	using System.Threading;
6		5
7	namespace Implab.Tests	6	namespace Implab.Test
8	{	7	{
9	[TestClass]	8	[TestClass]
10	public class AsyncTests	9	public class AsyncTests
@@ -90,12 +89,39 namespace Implab.Tests
90	public void PoolTest ()	89	public void PoolTest ()
91	{	90	{
92	var pid = Thread.CurrentThread.ManagedThreadId;	91	var pid = Thread.CurrentThread.ManagedThreadId;
93	var p = AsyncPool.Invoke (() => {	92	var p = AsyncPool.Invoke (() => Thread.CurrentThread.ManagedThreadId);
94	return Thread.CurrentThread.ManagedThreadId;
95	});
96		93
97	Assert.AreNotEqual (pid, p.Join ());	94	Assert.AreNotEqual (pid, p.Join ());
98	}	95	}
		96
		97	[TestMethod]
		98	public void ComplexCase1Test() {
		99	var flags = new bool[3];
		100
		101	// op1 (aync 200ms) => op2 (async 200ms) => op3 (sync map)
		102
		103	var p = PromiseHelper
		104	.Sleep(200, "Alan")
		105	.Cancelled(() => flags[0] = true)
		106	.Chain(x =>
		107	PromiseHelper
		108	.Sleep(200, "Hi, " + x)
		109	.Map( y => y )
		110	.Cancelled(() => flags[1] = true)
		111	)
		112	.Cancelled(() => flags[2] = true);
		113	Thread.Sleep(300);
		114	p.Cancel();
		115	try {
		116	Assert.AreEqual(p.Join(), "Hi, Alan");
		117	Assert.Fail("Shouldn't get here");
		118	} catch(OperationCanceledException) {
		119	}
		120
		121	Assert.IsFalse(flags[0]);
		122	Assert.IsTrue(flags[1]);
		123	Assert.IsTrue(flags[2]);
		124	}
99	}	125	}
100	}	126	}
101		127

Implab.Test/Implab.Test.csproj +1 0

               </ItemGroup>
               <ItemGroup>
                 <Compile Include="AsyncTests.cs" />
+                <Compile Include="PromiseHelper.cs" />
                 <Compile Include="Properties\AssemblyInfo.cs" />
               </ItemGroup>
               <ItemGroup>

Implab.suo binary modified 0 0

NO CONTENT: modified file, binary diff hidden

Implab/IPromise.cs +7 0

@@ -29,5 +29,12 namespace Implab
29	/// <param name="dependencies">Try to cancel the whole promise chain, the parent promise will be cancelled only if it has only one promise</param>	29	/// <param name="dependencies">Try to cancel the whole promise chain, the parent promise will be cancelled only if it has only one promise</param>
30	/// <returns></returns>	30	/// <returns></returns>
31	bool Cancel(bool dependencies);	31	bool Cancel(bool dependencies);
		32
		33	/// <summary>
		34	/// Registers handler for the case when the promise is cencelled. If the promise already cancelled the
		35	/// handler will be invoked immediatelly.
		36	/// </summary>
		37	/// <param name="handler">The handler</param>
		38	void HandleCancelled(Action handler);
32	}	39	}
33	}	40	}

Implab/Promise.cs +114 -98

             using System;
             using System.Collections.Generic;
-            using System.Linq;
             using System.Reflection;
-            using System.Text;
             using System.Diagnostics;
             using System.Threading;
                 public delegate void ErrorHandler(Exception e);
-                public delegate void ResultHandler<T>(T result);
+                public delegate void ResultHandler<in T>(T result);
-                public delegate TNew ResultMapper<TSrc, TNew>(TSrc result);
+                public delegate TNew ResultMapper<in TSrc, out TNew>(TSrc result);
-                public delegate Promise<TNew> ChainedOperation<TSrc, TNew>(TSrc result);
+                public delegate Promise<TNew> ChainedOperation<in TSrc, TNew>(TSrc result);
                 /// <summary>
                 /// Класс для асинхронного получения результатов. Так называемое "обещание".
                         public ErrorHandler errorHandler;
                     }
-                    IPromise m_parent;
+                    readonly IPromise m_parent;
                     LinkedList<ResultHandlerInfo> m_resultHandlers = new LinkedList<ResultHandlerInfo>();
                     LinkedList<Action> m_cancelHandlers = new LinkedList<Action>();
-                    object m_lock = new Object();
+                    readonly object m_lock = new Object();
-                    bool m_cancellable;
+                    readonly bool m_cancellable;
+                    int m_childrenCount = 0;
                     PromiseState m_state;
                     T m_result;
                     Exception m_error;
-                    int m_childrenCount;
                     public Promise() {
                         m_cancellable = true;
                     }
                     public Promise(IPromise parent, bool cancellable) {
                         m_cancellable = cancellable;
                         m_parent = parent;
+                        if (parent != null)
+                            parent.HandleCancelled(InternalCancel);
                     }
-                    /// <summary>
+                    void InternalCancel() {
-                    /// Событие, возникающее при отмене асинхронной операции.
+                        // don't try to cancel parent :)
-                    /// </summary>
+                        Cancel(false);
-                    /// <description>
-                    /// Как правило используется для оповещения объекта, выполняющего асинхронную операцию, о том, что ее следует отменить.
-                    /// </description>
-                    public event EventHandler Cancelled;
                     /// <summary>
                     /// Выполняет обещание, сообщая об успешном выполнении.
                             m_state = PromiseState.Resolved;
                         }
-                        // state has been changed to rejected new handlers can't be added
+                        OnStateChanged();
-                        foreach (var handler in m_resultHandlers)
-                            InvokeHandler(handler);
-                        /* ResultHandlerInfo handler;
-                        while (FetchNextHandler(out handler))
-                            InvokeHandler(handler); */
                     }
                     /// <summary>
                             m_state = PromiseState.Rejected;
                         }
-                        // state has been changed to rejected new handlers can't be added
+                        OnStateChanged();
-                        foreach (var handler in m_resultHandlers)
-                            InvokeHandler(handler);
-                        /*ResultHandlerInfo handler;
-                        while (FetchNextHandler(out handler))
-                            InvokeHandler(handler);*/
                     }
                     /// <summary>
                         return Cancel(true);
                     }
+                    protected virtual void OnStateChanged() {
+                        switch (m_state) {
+                            case PromiseState.Resolved:
+                                foreach (var resultHandlerInfo in m_resultHandlers)
+                                    try {
+                                        if (resultHandlerInfo.resultHandler != null)
+                                            resultHandlerInfo.resultHandler(m_result);
+                                    } catch (Exception e) {
+                                        try {
+                                            if (resultHandlerInfo.errorHandler != null)
+                                                resultHandlerInfo.errorHandler(e);
+                                        } catch { }
+                                    }
+                                break;
+                            case PromiseState.Cancelled:
+                                foreach (var cancelHandler in m_cancelHandlers)
+                                    cancelHandler();
+                                break;
+                            case PromiseState.Rejected:
+                                foreach (var resultHandlerInfo in m_resultHandlers)
+                                    try {
+                                        if (resultHandlerInfo.errorHandler != null)
+                                            resultHandlerInfo.errorHandler(m_error);
+                                    } catch { }
+                                break;
+                            default:
+                                throw new InvalidOperationException(String.Format("Promise entered an invalid state {0}", m_state));
+                        }
+                        m_resultHandlers = null;
+                        m_cancelHandlers = null;
+                    }
                     /// <summary>
                     /// Добавляет обработчики событий выполнения обещания.
                     /// </summary>
                         if (success != null)
                             handlerInfo.resultHandler = x => {
-                                try {
                                 success(x);
                                 medium.Resolve(x);
-                                } catch (Exception e) {
-                                    medium.Reject(e);
                             };
                         else
-                            handlerInfo.resultHandler = x => medium.Resolve(x);
+                            handlerInfo.resultHandler = medium.Resolve;
                         if (error != null)
                             handlerInfo.errorHandler = x => {
                                 medium.Reject(x);
                             };
                         else
-                            handlerInfo.errorHandler = x => medium.Reject(x);
+                            handlerInfo.errorHandler = medium.Reject;
                         AddHandler(handlerInfo);
                         AddHandler(new ResultHandlerInfo {
                             resultHandler = x => {
+                                // to avoid handler being called multiple times we handle exception by ourselfs
                                 try {
                                     handler();
                                     medium.Resolve(x);
                             throw new ArgumentNullException("mapper");
                         // создаем прицепленное обещание
-                        Promise<TNew> chained = new Promise<TNew>();
+                        var chained = new Promise<TNew>();
                         AddHandler(new ResultHandlerInfo() {
-                            resultHandler = delegate(T result) {
+                            resultHandler = result => chained.Resolve(mapper(result)),
-                                try {
-                                    // если преобразование выдаст исключение, то сработает reject сцепленного deferred
-                                    chained.Resolve(mapper(result));
-                                } catch (Exception e) {
-                                    chained.Reject(e);
-                            },
                             errorHandler = delegate(Exception e) {
                                 if (error != null)
+                                    try {
                                         error(e);
+                                    } catch { }
                                 // в случае ошибки нужно передать исключение дальше по цепочке
                                 chained.Reject(e);
                             }
                         // создать посредника, к которому будут подвызяваться следующие обработчики.
                         // когда будет выполнена реальная асинхронная операция, она обратиться к посреднику, чтобы
                         // передать через него результаты работы.
-                        Promise<TNew> medium = new Promise<TNew>();
+                        var medium = new Promise<TNew>(this, true);
-                        AddHandler(new ResultHandlerInfo() {
+                        AddHandler(new ResultHandlerInfo {
                             resultHandler = delegate(T result) {
-                                try {
+                                if (medium.State == PromiseState.Cancelled)
-                                    chained(result).Then(
+                                    return;
-                                        x => medium.Resolve(x),
-                                        e => medium.Reject(e)
+                                var promise = chained(result);
+                                // notify chained operation that it's not needed
+                                medium.Cancelled(() => promise.Cancel());
+                                promise.Then(
+                                    medium.Resolve,
+                                    medium.Reject
                                     );
-                                } catch (Exception e) {
-                                    // если сцепленное действие выдало исключение вместо обещания, то передаем ошибку по цепочке
-                                    medium.Reject(e);
                             },
                             errorHandler = delegate(Exception e) {
                                 if (error != null)
                         return Chain(chained, null);
                     }
+                    public Promise<T> Cancelled(Action handler) {
+                        if (handler == null)
+                            return this;
+                        lock (m_lock) {
+                            if (m_state == PromiseState.Unresolved)
+                                m_cancelHandlers.AddLast(handler);
+                            else if (m_state == PromiseState.Cancelled)
+                                handler();
+                        }
+                        return this;
+                    }
+                    public void HandleCancelled(Action handler) {
+                        Cancelled(handler);
+                    }
                     /// <summary>
                     /// Дожидается отложенного обещания и в случае успеха, возвращает
                     /// его, результат, в противном случае бросает исключение.
                     /// <param name="timeout">Время ожидания</param>
                     /// <returns>Результат выполнения обещания</returns>
                     public T Join(int timeout) {
-                        ManualResetEvent evt = new ManualResetEvent(false);
+                        var evt = new ManualResetEvent(false);
                         Anyway(() => evt.Set());
+                        Cancelled(() => evt.Set());
                         if (!evt.WaitOne(timeout, true))
                             throw new TimeoutException();
-                        if (m_error != null)
+                        switch (State) {
+                            case PromiseState.Resolved:
+                                return m_result;
+                            case PromiseState.Cancelled:
+                                throw new OperationCanceledException();
+                            case PromiseState.Rejected:
                                 throw new TargetInvocationException(m_error);
-                        else
+                            default:
-                            return m_result;
+                                throw new ApplicationException(String.Format("Invalid promise state {0}", State));
+                        }
                     }
                     public T Join() {
                         return Join(Timeout.Infinite);
                     }
-                    /// <summary>
-                    /// Данный метод последовательно извлекает обработчики обещания и когда
-                    /// их больше не осталось - ставит состояние "разрешено".
-                    /// </summary>
-                    /// <param name="handler">Информация об обработчике</param>
-                    /// <returns>Признак того, что еще остались обработчики в очереди</returns>
-                    bool FetchNextHandler(out ResultHandlerInfo handler) {
-                        handler = default(ResultHandlerInfo);
-                        lock (this) {
-                            Debug.Assert(m_state != PromiseState.Unresolved);
-                            if (m_resultHandlers.Count > 0) {
-                                handler = m_resultHandlers.First.Value;
-                                m_resultHandlers.RemoveFirst();
-                                return true;
-                            } else {
-                                return false;
                     void AddHandler(ResultHandlerInfo handler) {
                         bool invokeRequired = false;
-                        lock (this) {
+                        lock (m_lock) {
-                            if (m_state == PromiseState.Unresolved)
+                            m_childrenCount++;
+                            if (m_state == PromiseState.Unresolved) {
                                 m_resultHandlers.AddLast(handler);
-                            else
+                            } else
                                 invokeRequired = true;
                         }
                     }
                     void InvokeHandler(ResultHandlerInfo handler) {
-                        if (m_error == null) {
+                        switch (m_state) {
+                            case PromiseState.Resolved:
                                 try {
                                     if (handler.resultHandler != null)
                                         handler.resultHandler(m_result);
+                                } catch (Exception e) {
+                                    try {
+                                        if (handler.errorHandler != null)
+                                            handler.errorHandler(e);
                                     } catch { }
                                 }
+                                break;
-                        if (m_error != null) {
+                            case PromiseState.Rejected:
                                 try {
                                     if (handler.errorHandler != null)
                                         handler.errorHandler(m_error);
                                 } catch { }
+                                break;
+                            default:
+                                // do nothing
+                                return;
                         }
                     }
                             }
                         }
-                        if (dependencies && m_parent != null && m_parent.IsExclusive) {
+                        if (result)
-                            // TODO syncronize IsExclusive, AddHandler, Cancel (maybe CancelExclusive)
+                            OnStateChanged();
-                            m_parent.Cancel(true);
-                        if (result) {
+                        if (dependencies && m_parent != null && m_parent.IsExclusive) {
-                            // state has been changed to cancelled, new handlers can't be added
+                            m_parent.Cancel(true);
-                            foreach (var handler in m_cancelHandlers)
-                                handler();
                         }
                         return result;

Implab/TaskController.cs +1 -2

                 /// </remarks>
                 class TaskController
                 {
-                    object m_lock;
+                    readonly object m_lock;
                     string m_message;
-                    bool m_cancelled;
                     float m_current;
                     float m_max;

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