Mercurial > pub > ImplabNet

comparison Implab/Parallels/WorkerPool.cs @ 149:eb793fbbe4ea v2

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
fixed promises cancellation
author cin
date Wed, 06 May 2015 17:11:27 +0300
parents 471f596b2603
children
comparison
equal deleted inserted replaced
148:e6d4b41f0101 149:eb793fbbe4ea
28 28
29 public WorkerPool() { 29 public WorkerPool() {
30 InitPool(); 30 InitPool();
31 } 31 }
32 32
33 public Promise<T> Invoke<T>(Func<T> task) { 33 public IPromise<T> Invoke<T>(Func<T> task) {
34 if (task == null)
35 throw new ArgumentNullException("task");
36 if (IsDisposed)
37 throw new ObjectDisposedException(ToString());
38
39 var promise = new FuncTask<T>(task, null, null, true);
40
41 var lop = TraceContext.Instance.CurrentOperation;
42
43 EnqueueTask(delegate {
44 TraceContext.Instance.EnterLogicalOperation(lop, false);
45
46 promise.Resolve();
47
48 TraceContext.Instance.Leave();
49 });
50
51 return promise;
52 }
53
54 public IPromise Invoke(Action task) {
55 if (task == null)
56 throw new ArgumentNullException("task");
57 if (IsDisposed)
58 throw new ObjectDisposedException(ToString());
59
60 var promise = new ActionTask(task, null, null, true);
61
62 var lop = TraceContext.Instance.CurrentOperation;
63
64 EnqueueTask(delegate {
65 TraceContext.Instance.EnterLogicalOperation(lop, false);
66
67 promise.Resolve();
68
69 TraceContext.Instance.Leave();
70 });
71
72 return promise;
73 }
74
75 public IPromise<T> Invoke<T>(Func<ICancellationToken, T> task) {
34 if (task == null) 76 if (task == null)
35 throw new ArgumentNullException("task"); 77 throw new ArgumentNullException("task");
36 if (IsDisposed) 78 if (IsDisposed)
37 throw new ObjectDisposedException(ToString()); 79 throw new ObjectDisposedException(ToString());
38 80
41 var lop = TraceContext.Instance.CurrentOperation; 83 var lop = TraceContext.Instance.CurrentOperation;
42 84
43 EnqueueTask(delegate { 85 EnqueueTask(delegate {
44 TraceContext.Instance.EnterLogicalOperation(lop, false); 86 TraceContext.Instance.EnterLogicalOperation(lop, false);
45 try { 87 try {
46 promise.Resolve(task()); 88 if (!promise.CancelOperationIfRequested())
89 promise.Resolve(task(promise));
90 } catch (Exception e) {
91 promise.Reject(e);
92 } finally {
93 TraceContext.Instance.Leave();
94 }
95 });
96
97 return promise;
98 }
99
100 public IPromise Invoke<T>(Action<ICancellationToken> task) {
101 if (task == null)
102 throw new ArgumentNullException("task");
103 if (IsDisposed)
104 throw new ObjectDisposedException(ToString());
105
106 var promise = new Promise();
107
108 var lop = TraceContext.Instance.CurrentOperation;
109
110 EnqueueTask(delegate {
111 TraceContext.Instance.EnterLogicalOperation(lop, false);
112 try {
113 if (!promise.CancelOperationIfRequested()) {
114 task(promise);
115 promise.Resolve();
116 }
47 } catch (Exception e) { 117 } catch (Exception e) {
48 promise.Reject(e); 118 promise.Reject(e);
49 } finally { 119 } finally {
50 TraceContext.Instance.Leave(); 120 TraceContext.Instance.Leave();
51 } 121 }