Mercurial > pub > ImplabNet

comparison Implab/AbstractPromise.cs @ 125:f803565868a4 v2

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
improved performance of promises
author cin
date Thu, 15 Jan 2015 12:09:20 +0300
parents 2573b562e328
children 671f60cd0250
comparison
equal deleted inserted replaced
124:a336cb13c6a9 125:f803565868a4
10 const int TRANSITIONAL_STATE = 1; 10 const int TRANSITIONAL_STATE = 1;
11 const int SUCCEEDED_STATE = 2; 11 const int SUCCEEDED_STATE = 2;
12 const int REJECTED_STATE = 3; 12 const int REJECTED_STATE = 3;
13 const int CANCELLED_STATE = 4; 13 const int CANCELLED_STATE = 4;
14 14
15 const int RESERVED_HANDLERS_COUNT = 4;
16
15 int m_state; 17 int m_state;
16 Exception m_error; 18 Exception m_error;
17 19 int m_handlersCount;
18 readonly AsyncQueue<THandler> m_handlers = new AsyncQueue<THandler>(); 20
21 readonly THandler[] m_handlers = new THandler[RESERVED_HANDLERS_COUNT];
22 MTQueue<THandler> m_extraHandlers;
23 int m_handlerPointer = -1;
24 int m_handlersCommited;
19 25
20 #region state managment 26 #region state managment
21 bool BeginTransit() { 27 bool BeginTransit() {
22 return UNRESOLVED_SATE == Interlocked.CompareExchange(ref m_state, TRANSITIONAL_STATE, UNRESOLVED_SATE); 28 return UNRESOLVED_SATE == Interlocked.CompareExchange(ref m_state, TRANSITIONAL_STATE, UNRESOLVED_SATE);
23 } 29 }
86 protected abstract void SignalError(THandler handler, Exception error); 92 protected abstract void SignalError(THandler handler, Exception error);
87 93
88 protected abstract void SignalCancelled(THandler handler); 94 protected abstract void SignalCancelled(THandler handler);
89 95
90 void OnSuccess() { 96 void OnSuccess() {
91 THandler handler; 97 var hp = m_handlerPointer;
92 while (m_handlers.TryDequeue(out handler)) 98 var slot = hp +1 ;
93 SignalSuccess(handler); 99 while (slot < m_handlersCommited) {
100 if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) == hp) {
101 SignalSuccess(m_handlers[slot]);
102 }
103 hp = m_handlerPointer;
104 slot = hp +1 ;
105 }
106
107
108 if (m_extraHandlers != null) {
109 THandler handler;
110 while (m_extraHandlers.TryDequeue(out handler))
111 SignalSuccess(handler);
112 }
94 } 113 }
95 114
96 void OnError() { 115 void OnError() {
97 THandler handler; 116 var hp = m_handlerPointer;
98 while (m_handlers.TryDequeue(out handler)) 117 var slot = hp +1 ;
99 SignalError(handler,m_error); 118 while (slot < m_handlersCommited) {
119 if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) == hp) {
120 SignalError(m_handlers[slot],m_error);
121 }
122 hp = m_handlerPointer;
123 slot = hp +1 ;
124 }
125
126 if (m_extraHandlers != null) {
127 THandler handler;
128 while (m_extraHandlers.TryDequeue(out handler))
129 SignalError(handler, m_error);
130 }
100 } 131 }
101 132
102 void OnCancelled() { 133 void OnCancelled() {
103 THandler handler; 134 var hp = m_handlerPointer;
104 while (m_handlers.TryDequeue(out handler)) 135 var slot = hp +1 ;
105 SignalCancelled(handler); 136 while (slot < m_handlersCommited) {
137 if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) == hp) {
138 SignalCancelled(m_handlers[slot]);
139 }
140 hp = m_handlerPointer;
141 slot = hp +1 ;
142 }
143
144 if (m_extraHandlers != null) {
145 THandler handler;
146 while (m_extraHandlers.TryDequeue(out handler))
147 SignalCancelled(handler);
148 }
106 } 149 }
107 150
108 #endregion 151 #endregion
109 152
110 protected abstract void Listen(PromiseEventType events, Action handler); 153 protected abstract void Listen(PromiseEventType events, Action handler);
143 186
144 #region handlers managment 187 #region handlers managment
145 188
146 protected void AddHandler(THandler handler) { 189 protected void AddHandler(THandler handler) {
147 190
148 if (IsResolved) { 191 if (m_state > 1) {
192 // the promise is in the resolved state, just invoke the handler
149 InvokeHandler(handler); 193 InvokeHandler(handler);
150
151 } else { 194 } else {
152 // the promise is in the resolved state, just invoke the handler 195 var slot = Interlocked.Increment(ref m_handlersCount) - 1;
153 m_handlers.Enqueue(handler); 196
154 197 if (slot < RESERVED_HANDLERS_COUNT) {
155 198 m_handlers[slot] = handler;
156 if (IsResolved && m_handlers.TryDequeue(out handler)) 199
200 while (slot != Interlocked.CompareExchange(ref m_handlersCommited, slot + 1, slot)) {
201 }
202
203 if (m_state > 1) {
204 do {
205 var hp = m_handlerPointer;
206 slot = hp + 1;
207 if (slot < m_handlersCommited) {
208 if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) != hp)
209 continue;
210 InvokeHandler(m_handlers[slot]);
211 }
212 break;
213 } while(true);
214 }
215 } else {
216 if (slot == RESERVED_HANDLERS_COUNT) {
217 m_extraHandlers = new MTQueue<THandler>();
218 } else {
219 while (m_extraHandlers == null)
220 Thread.MemoryBarrier();
221 }
222
223 m_extraHandlers.Enqueue(handler);
224
225 if (m_state > 1 && m_extraHandlers.TryDequeue(out handler))
157 // if the promise have been resolved while we was adding the handler to the queue 226 // if the promise have been resolved while we was adding the handler to the queue
158 // we can't guarantee that someone is still processing it 227 // we can't guarantee that someone is still processing it
159 // therefore we need to fetch a handler from the queue and execute it 228 // therefore we need to fetch a handler from the queue and execute it
160 // note that fetched handler may be not the one that we have added 229 // note that fetched handler may be not the one that we have added
161 // even we can fetch no handlers at all :) 230 // even we can fetch no handlers at all :)
162 InvokeHandler(handler); 231 InvokeHandler(handler);
232 }
163 } 233 }
164 } 234 }
165 235
166 protected void InvokeHandler(THandler handler) { 236 protected void InvokeHandler(THandler handler) {
167 switch (m_state) { 237 switch (m_state) {