implab/ImplabNet Commit - r130:671f60cd0250

fixed Resove method bug when calling it on already cancelled promise

cin -

r130:671f60cd0250 v2

parent child

Context file:

r130:671f60cd0250

Collapse all files

Implab.Diagnostics.Interactive/InteractiveListener.cs +3 -3

             using Implab.Parallels;
             using System;
             using System.Collections.Generic;
             using System.Linq;
             using System.Text;
             using System.Threading;
             using System.Threading.Tasks;
             using System.Windows.Forms;
             namespace Implab.Diagnostics.Interactive
             {
                 public class InteractiveListener: TextListenerBase
                 {
                     TraceForm m_form;
                     SynchronizationContext m_syncGuiThread;
-                    readonly Promise<object> m_guiStarted = new Promise<object>();
+                    readonly Promise m_guiStarted = new Promise();
                     readonly IPromise m_guiFinished;
                     //        readonly IPromise m_workerFinished = new Promise<object>();
                     readonly MTQueue<TraceViewItem> m_queue = new MTQueue<TraceViewItem>();
                     readonly AutoResetEvent m_queueEvent = new AutoResetEvent(false);
                     int m_queueLength;
                     bool m_exitPending;
                     readonly object m_pauseLock = new object();
                     bool m_paused;
                     readonly ManualResetEvent m_pauseEvent = new ManualResetEvent(true);
                     public InteractiveListener(bool global) : base(global) {
-                        m_guiFinished = AsyncPool.InvokeNewThread(GuiThread);
+                        m_guiFinished = AsyncPool.RunThread(GuiThread);
-                        /*m_workerFinished = */AsyncPool.InvokeNewThread(QueueThread);
+                        /*m_workerFinished = */AsyncPool.RunThread(QueueThread);
                         m_guiStarted.Join();
                     }
                     void GuiThread() {
                         m_form = new TraceForm(); // will create SynchronizationContext
                         m_form.PauseEvents += (s,a) => Pause();
                         m_form.ResumeEvents += (s, a) => Resume();
                         m_syncGuiThread = SynchronizationContext.Current;
                         m_guiStarted.Resolve();
                         Application.Run();
                     }
                     void QueueThread() {
                         while (!m_exitPending) {
                             if (m_paused)
                                 m_pauseEvent.WaitOne();
                             TraceViewItem item;
                             if (m_queue.TryDequeue(out item)) {
                                 Interlocked.Decrement(ref m_queueLength);
                                 m_syncGuiThread.Send(x => m_form.AddTraceEvent(item),null);
                             } else {
                                 m_queueEvent.WaitOne();
                             }
                         }
                     }
                     public void Pause() {
                         // for consistency we need to set this properties atomically
                         lock (m_pauseLock) {
                             m_pauseEvent.Reset();
                             m_paused = true;
                         }
                     }
                     public void Resume() {
                         // for consistency we need to set this properties atomically
                         lock (m_pauseLock) {
                             m_paused = false;
                             m_pauseEvent.Set();
                         }
                     }
                     void Enqueue(TraceViewItem item) {
                         m_queue.Enqueue(item);
                         if (Interlocked.Increment(ref m_queueLength) == 1)
                             m_queueEvent.Set();
                     }
                     public void ShowForm() {
                         m_syncGuiThread.Post(x => m_form.Show(), null);
                     }
                     public void HideForm() {
                         m_syncGuiThread.Post(x => m_form.Hide(), null);
                     }
                     void Terminate() {
                         m_exitPending = true;
                         Resume();
                         m_syncGuiThread.Post(x => Application.ExitThread(), null);
                     }
                     protected override void Dispose(bool disposing) {
                         if (disposing) {
                             Terminate();
                             m_guiFinished.Join();
                         }
                         base.Dispose(disposing);
                     }
                     protected override void WriteEntry(LogEventArgs args, EventText text, string channel) {
                         var item = new TraceViewItem {
                             Indent = text.indent,
                             Message = text.content,
                             Thread = args.ThreadId,
                             Channel = channel,
                             Timestamp = Environment.TickCount
                         };
                         Enqueue(item);
                     }
                 }
             }

Implab/AbstractPromise.cs +3 -1

             using System;
             using Implab.Parallels;
             using System.Threading;
             using System.Reflection;
             namespace Implab {
                 public abstract class AbstractPromise<THandler> {
                     const int UNRESOLVED_SATE = 0;
                     const int TRANSITIONAL_STATE = 1;
                     const int SUCCEEDED_STATE = 2;
                     const int REJECTED_STATE = 3;
                     const int CANCELLED_STATE = 4;
                     const int RESERVED_HANDLERS_COUNT = 4;
                     int m_state;
                     Exception m_error;
                     int m_handlersCount;
                     readonly THandler[] m_handlers = new THandler[RESERVED_HANDLERS_COUNT];
                     MTQueue<THandler> m_extraHandlers;
                     int m_handlerPointer = -1;
                     int m_handlersCommited;
                     #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 void BeginSetResult() {
+                    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);
                         OnSuccess();
                     }
                     /// <summary>
                     /// Выполняет обещание, сообщая об ошибке
                     /// </summary>
                     /// <remarks>
                     /// Поскольку обещание должно работать в многопточной среде, при его выполнении сразу несколько потоков
                     /// могу вернуть ошибку, при этом только первая будет использована в качестве результата, остальные
                     /// будут проигнорированы.
                     /// </remarks>
                     /// <param name="error">Исключение возникшее при выполнении операции</param>
                     /// <exception cref="InvalidOperationException">Данное обещание уже выполнено</exception>
                     protected void SetError(Exception error) {
                         if (BeginTransit()) {
                             m_error = error is PromiseTransientException ? error.InnerException : error;
                             CompleteTransit(REJECTED_STATE);
                             OnError();
                         } else {
                             WaitTransition();
                             if (m_state == SUCCEEDED_STATE)
                                 throw new InvalidOperationException("The promise is already resolved");
                         }
                     }
                     /// <summary>
                     /// Отменяет операцию, если это возможно.
                     /// </summary>
                     /// <remarks>Для определения была ли операция отменена следует использовать свойство <see cref="IsCancelled"/>.</remarks>
                     protected void SetCancelled() {
                         if (BeginTransit()) {
                             CompleteTransit(CANCELLED_STATE);
                             OnCancelled();
                         }
                     }
                     protected abstract void SignalSuccess(THandler handler);
                     protected abstract void SignalError(THandler handler, Exception error);
                     protected abstract void SignalCancelled(THandler handler);
                     void OnSuccess() {
                         var hp = m_handlerPointer;
                         var slot = hp +1 ;
                         while (slot < m_handlersCommited) {
                             if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) == hp) {
                                 SignalSuccess(m_handlers[slot]);
                             }
                             hp = m_handlerPointer;
                             slot = hp +1 ;
                         }
                         if (m_extraHandlers != null) {
                             THandler handler;
                             while (m_extraHandlers.TryDequeue(out handler))
                                 SignalSuccess(handler);
                         }
                     }
                     void OnError() {
                         var hp = m_handlerPointer;
                         var slot = hp +1 ;
                         while (slot < m_handlersCommited) {
                             if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) == hp) {
                                 SignalError(m_handlers[slot],m_error);
                             }
                             hp = m_handlerPointer;
                             slot = hp +1 ;
                         }
                         if (m_extraHandlers != null) {
                             THandler handler;
                             while (m_extraHandlers.TryDequeue(out handler))
                                 SignalError(handler, m_error);
                         }
                     }
                     void OnCancelled() {
                         var hp = m_handlerPointer;
                         var slot = hp +1 ;
                         while (slot < m_handlersCommited) {
                             if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) == hp) {
                                 SignalCancelled(m_handlers[slot]);
                             }
                             hp = m_handlerPointer;
                             slot = hp +1 ;
                         }
                         if (m_extraHandlers != null) {
                             THandler handler;
                             while (m_extraHandlers.TryDequeue(out handler))
                                 SignalCancelled(handler);
                         }
                     }
                     #endregion
                     protected abstract void Listen(PromiseEventType events, Action handler);
                     #region synchronization traits
                     protected void WaitResult(int timeout) {
                         if (!IsResolved) {
                             var lk = new object();
                             Listen(PromiseEventType.All, () => {
                                 lock(lk) {
                                     Monitor.Pulse(lk);
                                 }
                             });
                             lock (lk) {
                                 while(!IsResolved) {
                                     if(!Monitor.Wait(lk,timeout))
                                         throw new TimeoutException();
                                 }
                             }
                         }
                         switch (m_state) {
                             case SUCCEEDED_STATE:
                                 return;
                             case CANCELLED_STATE:
                                 throw new OperationCanceledException();
                             case REJECTED_STATE:
                                 throw new TargetInvocationException(m_error);
                             default:
                                 throw new ApplicationException(String.Format("Invalid promise state {0}", 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
                             InvokeHandler(handler);
                         } else {
                             var slot = Interlocked.Increment(ref m_handlersCount) - 1;
                             if (slot < RESERVED_HANDLERS_COUNT) {
                                 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;
                                             InvokeHandler(m_handlers[slot]);
                                         }
                                         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 :)
                                 InvokeHandler(handler);
                             }
                         }
                     }
                     protected void InvokeHandler(THandler handler) {
                         switch (m_state) {
                             case SUCCEEDED_STATE:
                                 SignalSuccess(handler);
                                 break;
                             case CANCELLED_STATE:
                                 SignalCancelled(handler);
                                 break;
                             case REJECTED_STATE:
                                 SignalError(handler, m_error);
                                 break;
                             default:
                                 throw new Exception(String.Format("Invalid promise state {0}", m_state));
                         }
                     }
                     #endregion
                     #region IPromise implementation
                     public void Join(int timeout) {
                         WaitResult(timeout);
                     }
                     public void Join() {
                         WaitResult(-1);
                     }
                     public bool IsResolved {
                         get {
                             Thread.MemoryBarrier();
                             return m_state > 1;
                         }
                     }
                     public bool IsCancelled {
                         get {
                             Thread.MemoryBarrier();
                             return m_state == CANCELLED_STATE;
                         }
                     }
                     #endregion
                     #region ICancellable implementation
                     public void Cancel() {
                         SetCancelled();
                     }
                     #endregion
                 }
             }

Implab/Parallels/AsyncQueue.cs +2 0

             using System.Threading;
             using System.Collections.Generic;
             using System;
             using System.Collections;
             using System.Diagnostics;
             namespace Implab.Parallels {
                 public class AsyncQueue<T> : IEnumerable<T> {
                     class Chunk {
                         public Chunk next;
                         int m_low;
                         int m_hi;
                         int m_alloc;
                         readonly int m_size;
                         readonly T[] m_data;
                         public Chunk(int size) {
                             m_size = size;
                             m_data = new T[size];
                         }
                         public Chunk(int size, T value) {
                             m_size = size;
                             m_hi = 1;
                             m_alloc = 1;
                             m_data = new T[size];
                             m_data[0] = value;
                         }
                         public Chunk(int size, T[] data, int offset, int length, int alloc) {
                             m_size = size;
                             m_hi = length;
                             m_alloc = alloc;
                             m_data = new T[size];
                             Array.Copy(data, offset, m_data, 0, length);
                         }
                         public int Low {
                             get { return m_low; }
                         }
                         public int Hi {
                             get { return m_hi; }
                         }
                         public int Size {
                             get { return m_size; }
                         }
                         public bool TryEnqueue(T value, out bool extend) {
                             var alloc = Interlocked.Increment(ref m_alloc) - 1;
                             if (alloc >= m_size) {
                                 extend = alloc == m_size;
                                 return false;
                             }
                             extend = false;
                             m_data[alloc] = value;
                             while (alloc != Interlocked.CompareExchange(ref m_hi, alloc + 1, alloc)) {
                                 // spin wait for commit
                             }
                             return true;
                         }
                         /// <summary>
                         /// Prevents from allocating new space in the chunk and waits for all write operations to complete
                         /// </summary>
                         public void Commit() {
                             var actual = Math.Min(Interlocked.Exchange(ref m_alloc, m_size + 1), m_size);
                             while (m_hi != actual)
                                 Thread.MemoryBarrier();
                         }
                         public bool TryDequeue(out T value, out bool recycle) {
                             int low;
                             do {
                                 low = m_low;
                                 if (low >= m_hi) {
                                     value = default(T);
                                     recycle = (low == m_size);
                                     return false;
                                 }
                             } while(low != Interlocked.CompareExchange(ref m_low, low + 1, low));
                             recycle = (low == m_size - 1);
                             value = m_data[low];
                             return true;
                         }
                         public bool TryEnqueueBatch(T[] batch, int offset, int length, out int enqueued, out bool extend) {
                             //int alloc;
                             //int allocSize;
                             var alloc = Interlocked.Add(ref m_alloc, length) - length;
                             if (alloc > m_size) {
                                 // the chunk is full and someone already
                                 // creating the new one
                                 enqueued = 0; // nothing was added
                                 extend = false; // the caller shouldn't try to extend the queue
                                 return false; // nothing was added
                             }
                             enqueued = Math.Min(m_size - alloc, length);
                             extend = length > enqueued;
                             if (enqueued == 0)
                                 return false;
                             Array.Copy(batch, offset, m_data, alloc, enqueued);
                             while (alloc != Interlocked.CompareExchange(ref m_hi, alloc + enqueued, alloc)) {
                                 // spin wait for commit
                             }
                             return true;
                         }
                         public bool TryDequeueBatch(T[] buffer, int offset, int length,out int dequeued, out bool recycle) {
                             int low, hi, batchSize;
                             do {
                                 low = m_low;
                                 hi = m_hi;
                                 if (low >= hi) {
                                     dequeued = 0;
                                     recycle = (low == m_size); // recycling could be restarted and we need to signal again
                                     return false;
                                 }
                                 batchSize = Math.Min(hi - low, length);
                             } while(low != Interlocked.CompareExchange(ref m_low, low + batchSize, low));
                             recycle = (low == m_size - batchSize);
                             dequeued = batchSize;
                             Array.Copy(m_data, low, buffer, offset, batchSize);
                             return true;
                         }
                         public T GetAt(int pos) {
                             return m_data[pos];
                         }
                     }
                     public const int DEFAULT_CHUNK_SIZE = 32;
                     public const int MAX_CHUNK_SIZE = 262144;
                     Chunk m_first;
                     Chunk m_last;
                     /// <summary>
                     /// Adds the specified value to the queue.
                     /// </summary>
                     /// <param name="value">Tha value which will be added to the queue.</param>
                     public void Enqueue(T value) {
                         var last = m_last;
                         // spin wait to the new chunk
                         bool extend = true;
                         while (last == null || !last.TryEnqueue(value, out extend)) {
                             // try to extend queue
                             if (extend || last == null) {
                                 var chunk = new Chunk(DEFAULT_CHUNK_SIZE, value);
                                 if (EnqueueChunk(last, chunk))
                                     break; // success! exit!
                                 last = m_last;
                             } else {
                                 while (last == m_last) {
                                     Thread.MemoryBarrier();
                                 }
                                 last = m_last;
                             }
                         }
                     }
                     /// <summary>
                     /// Adds the specified data to the queue.
                     /// </summary>
                     /// <param name="data">The buffer which contains the data to be enqueued.</param>
                     /// <param name="offset">The offset of the data in the buffer.</param>
                     /// <param name="length">The size of the data to read from the buffer.</param>
                     public void EnqueueRange(T[] data, int offset, int length) {
                         if (data == null)
                             throw new ArgumentNullException("data");
+                        if (length == 0)
+                            return;
                         if (offset < 0)
                             throw new ArgumentOutOfRangeException("offset");
                         if (length < 1 || offset + length > data.Length)
                             throw new ArgumentOutOfRangeException("length");
                         var last = m_last;
                         bool extend;
                         int enqueued;
                         while (length > 0) {
                             extend = true;
                             if (last != null && last.TryEnqueueBatch(data, offset, length, out enqueued, out extend)) {
                                 length -= enqueued;
                                 offset += enqueued;
                             }
                             if (extend) {
                                 // there was no enough space in the chunk
                                 // or there was no chunks in the queue
                                 while (length > 0) {
                                     var size = Math.Min(length, MAX_CHUNK_SIZE);
                                     var chunk = new Chunk(
                                         Math.Max(size, DEFAULT_CHUNK_SIZE),
                                         data,
                                         offset,
                                         size,
                                         length // length >= size
                                     );
                                     if (!EnqueueChunk(last, chunk)) {
                                         // looks like the queue has been updated then proceed from the beginning
                                         last = m_last;
                                         break;
                                     }
                                     // we have successfully added the new chunk
                                     last = chunk;
                                     length -= size;
                                     offset += size;
                                 }
                             } else {
                                 // we don't need to extend the queue, if we successfully enqueued data
                                 if (length == 0)
                                     break;
                                 // if we need to wait while someone is extending the queue
                                 // spinwait
                                 while (last == m_last) {
                                     Thread.MemoryBarrier();
                                 }
                                 last = m_last;
                             }
                         }
                     }
                     /// <summary>
                     /// Tries to retrieve the first element from the queue.
                     /// </summary>
                     /// <returns><c>true</c>, if element is dequeued, <c>false</c> otherwise.</returns>
                     /// <param name="value">The value of the dequeued element.</param>
                     public bool TryDequeue(out T value) {
                         var chunk = m_first;
                         bool recycle;
                         while (chunk != null) {
                             var result = chunk.TryDequeue(out value, out recycle);
                             if (recycle) // this chunk is waste
                                 RecycleFirstChunk(chunk);
                             else
                                 return result; // this chunk is usable and returned actual result
                             if (result) // this chunk is waste but the true result is always actual
                                 return true;
                             // try again
                             chunk = m_first;
                         }
                         // the queue is empty
                         value = default(T);
                         return false;
                     }
                     /// <summary>
                     /// Tries to dequeue the specified amount of data from the queue.
                     /// </summary>
                     /// <returns><c>true</c>, if data was deuqueued, <c>false</c> otherwise.</returns>
                     /// <param name="buffer">The buffer to which the data will be written.</param>
                     /// <param name="offset">The offset in the buffer at which the data will be written.</param>
                     /// <param name="length">The maximum amount of data to be retrieved.</param>
                     /// <param name="dequeued">The actual amout of the retrieved data.</param>
                     public bool TryDequeueRange(T[] buffer, int offset, int length, out int dequeued) {
                         if (buffer == null)
                             throw new ArgumentNullException("buffer");
                         if (offset < 0)
                             throw new ArgumentOutOfRangeException("offset");
                         if (length < 1 || offset + length > buffer.Length)
                             throw new ArgumentOutOfRangeException("length");
                         var chunk = m_first;
                         bool recycle;
                         dequeued = 0;
                         while (chunk != null) {
                             int actual;
                             if (chunk.TryDequeueBatch(buffer, offset, length, out actual, out recycle)) {
                                 offset += actual;
                                 length -= actual;
                                 dequeued += actual;
                             }
                             if (recycle) // this chunk is waste
                                 RecycleFirstChunk(chunk);
                             else if (actual == 0)
                                 break; // the chunk is usable, but doesn't contain any data (it's the last chunk in the queue)
                             if (length == 0)
                                 return true;
                             // we still may dequeue something
                             // try again
                             chunk = m_first;
                         }
                         return dequeued != 0;
                     }
                     /// <summary>
                     /// Tries to dequeue all remaining data in the first chunk.
                     /// </summary>
                     /// <returns><c>true</c>, if data was dequeued, <c>false</c> otherwise.</returns>
                     /// <param name="buffer">The buffer to which the data will be written.</param>
                     /// <param name="offset">The offset in the buffer at which the data will be written.</param>
                     /// <param name="length">Tha maximum amount of the data to be dequeued.</param>
                     /// <param name="dequeued">The actual amount of the dequeued data.</param>
                     public bool TryDequeueChunk(T[] buffer, int offset, int length, out int dequeued) {
                         if (buffer == null)
                             throw new ArgumentNullException("buffer");
                         if (offset < 0)
                             throw new ArgumentOutOfRangeException("offset");
                         if (length < 1 || offset + length > buffer.Length)
                             throw new ArgumentOutOfRangeException("length");
                         var chunk = m_first;
                         bool recycle;
                         dequeued = 0;
                         while (chunk != null) {
                             int actual;
                             if (chunk.TryDequeueBatch(buffer, offset, length, out actual, out recycle)) {
                                 dequeued = actual;
                             }
                             if (recycle) // this chunk is waste
                                 RecycleFirstChunk(chunk);
                             // if we have dequeued any data, then return
                             if (dequeued != 0)
                                 return true;
                             // we still may dequeue something
                             // try again
                             chunk = m_first;
                         }
                         return false;
                     }
                     bool EnqueueChunk(Chunk last, Chunk chunk) {
                         if (Interlocked.CompareExchange(ref m_last, chunk, last) != last)
                             return false;
                         if (last != null)
                             last.next = chunk;
                         else {
                             m_first = chunk;
                         }
                         return true;
                     }
                     void RecycleFirstChunk(Chunk first) {
                         var next = first.next;
                         if (first != Interlocked.CompareExchange(ref m_first, next, first))
                             return;
                         if (next == null) {
                             if (first != Interlocked.CompareExchange(ref m_last, null, first)) {
                                 /*while (first.next == null)
                                     Thread.MemoryBarrier();*/
                                 // race
                                 // someone already updated the tail, restore the pointer to the queue head
                                 m_first = first;
                             }
                             // the tail is updated
                         }
                         // we need to update the head
                         //Interlocked.CompareExchange(ref m_first, next, first);
                         // if the head is already updated then give up
                         //return;
                     }
                     public void Clear() {
                         // start the new queue
                         var chunk = new Chunk(DEFAULT_CHUNK_SIZE);
                         do {
                             Thread.MemoryBarrier();
                             var first = m_first;
                             var last = m_last;
                             if (last == null) // nothing to clear
                                 return;
                             if (first == null || (first.next == null && first != last)) // inconcistency
                                 continue;
                             // here we will create inconsistency which will force others to spin
                             // and prevent from fetching. chunk.next = null
                             if (first != Interlocked.CompareExchange(ref m_first, chunk, first))
                                 continue;// inconsistent
                             m_last = chunk;
                             return;
                         } while(true);
                     }
                     public T[] Drain() {
                         // start the new queue
                         var chunk = new Chunk(DEFAULT_CHUNK_SIZE);
                         do {
                             Thread.MemoryBarrier();
                             var first = m_first;
                             var last = m_last;
                             if (last == null)
                                 return new T[0];
                             if (first == null || (first.next == null && first != last))
                                 continue;
                             // here we will create inconsistency which will force others to spin
                             // and prevent from fetching. chunk.next = null
                             if (first != Interlocked.CompareExchange(ref m_first, chunk, first))
                                 continue;// inconsistent
                             last = Interlocked.Exchange(ref m_last, chunk);
                             return ReadChunks(first, last);
                         } while(true);
                     }
                     static T[] ReadChunks(Chunk chunk, object last) {
                         var result = new List<T>();
                         var buffer = new T[DEFAULT_CHUNK_SIZE];
                         int actual;
                         bool recycle;
                         while (chunk != null) {
                             // ensure all write operations on the chunk are complete
                             chunk.Commit();
                             // we need to read the chunk using this way
                             // since some client still may completing the dequeue
                             // operation, such clients most likely won't get results
                             while (chunk.TryDequeueBatch(buffer, 0, buffer.Length, out actual, out recycle))
                                 result.AddRange(new ArraySegmentCollection(buffer, 0, actual));
                             if (chunk == last) {
                                 chunk = null;
                             } else {
                                 while (chunk.next == null)
                                     Thread.MemoryBarrier();
                                 chunk = chunk.next;
                             }
                         }
                         return result.ToArray();
                     }
                     struct ArraySegmentCollection : ICollection<T> {
                         readonly T[] m_data;
                         readonly int m_offset;
                         readonly int m_length;
                         public ArraySegmentCollection(T[] data, int offset, int length) {
                             m_data = data;
                             m_offset = offset;
                             m_length = length;
                         }
                         #region ICollection implementation
                         public void Add(T item) {
                             throw new NotSupportedException();
                         }
                         public void Clear() {
                             throw new NotSupportedException();
                         }
                         public bool Contains(T item) {
                             return false;
                         }
                         public void CopyTo(T[] array, int arrayIndex) {
                             Array.Copy(m_data,m_offset,array,arrayIndex, m_length);
                         }
                         public bool Remove(T item) {
                             throw new NotSupportedException();
                         }
                         public int Count {
                             get {
                                 return m_length;
                             }
                         }
                         public bool IsReadOnly {
                             get {
                                 return true;
                             }
                         }
                         #endregion
                         #region IEnumerable implementation
                         public IEnumerator<T> GetEnumerator() {
                             for (int i = m_offset; i < m_length + m_offset; i++)
                                 yield return m_data[i];
                         }
                         #endregion
                         #region IEnumerable implementation
                         IEnumerator IEnumerable.GetEnumerator() {
                             return GetEnumerator();
                         }
                         #endregion
                     }
                     #region IEnumerable implementation
                     class Enumerator : IEnumerator<T> {
                         Chunk m_current;
                         int m_pos = -1;
                         public Enumerator(Chunk fisrt) {
                             m_current = fisrt;
                         }
                         #region IEnumerator implementation
                         public bool MoveNext() {
                             if (m_current == null)
                                 return false;
                             if (m_pos == -1)
                                 m_pos = m_current.Low;
                             else
                                 m_pos++;
                             if (m_pos == m_current.Hi) {
                                 m_current = m_pos == m_current.Size ? m_current.next : null;
                                 m_pos = 0;
                                 if (m_current == null)
                                     return false;
                             }
                             return true;
                         }
                         public void Reset() {
                             throw new NotSupportedException();
                         }
                         object IEnumerator.Current {
                             get {
                                 return Current;
                             }
                         }
                         #endregion
                         #region IDisposable implementation
                         public void Dispose() {
                         }
                         #endregion
                         #region IEnumerator implementation
                         public T Current {
                             get {
                                 if (m_pos == -1 || m_current == null)
                                     throw new InvalidOperationException();
                                 return m_current.GetAt(m_pos);
                             }
                         }
                         #endregion
                     }
                     public IEnumerator<T> GetEnumerator() {
                         return new Enumerator(m_first);
                     }
                     #endregion
                     #region IEnumerable implementation
                     IEnumerator IEnumerable.GetEnumerator() {
                         return GetEnumerator();
                     }
                     #endregion
                 }
             }

Implab/Parallels/SharedLock.cs +2 -1

             using System;
             using System.Threading;
             using System.Diagnostics;
             namespace Implab.Parallels {
                 /// <summary>
                 /// Implements a lightweight mechanism to aquire a shared or an exclusive lock.
                 /// </summary>
                 public class SharedLock {
                     readonly object m_lock = new object();
                     int m_locks;
                     bool m_exclusive;
                     public bool LockExclusive(int timeout) {
                         lock (m_lock) {
                             if (m_locks > 0 && !Monitor.Wait(m_lock, timeout))
                                 return false;
                             m_exclusive = true;
                             m_locks = 1;
+                            return true;
                         }
                     }
                     public void LockExclusive() {
                         LockExclusive(-1);
                     }
                     public bool LockShared(int timeout) {
                         lock (m_lock) {
                             if (!m_exclusive) {
                                 m_locks++;
                                 return true;
                             }
-                            if (m_lock == 0) {
+                            if (m_locks == 0) {
                                 m_exclusive = false;
                                 m_locks = 1;
                                 return true;
                             }
                             if (Monitor.Wait(m_lock, timeout)) {
                                 Debug.Assert(m_locks == 0);
                                 m_locks = 1;
                                 m_exclusive = false;
                                 return true;
                             }
                             return false;
                         }
                     }
                     public void LockShared() {
                         LockShared(-1);
                     }
                     public void ReleaseShared() {
                         lock (m_lock) {
                             if (m_exclusive || m_locks <= 0)
                                 throw new InvalidOperationException();
                             m_locks--;
                             if (m_locks == 0)
                                 Monitor.PulseAll(m_lock);
                         }
                     }
                     public void ReleaseExclusive() {
                         lock (m_lock) {
                             if (!m_exclusive && m_locks != 1)
                                 throw new InvalidOperationException();
                             m_locks = 0;
                             Monitor.PulseAll(m_lock);
                         }
                     }
                 }
             }

Implab/PromiseT.cs +4 -3

             using System;
             using System.Diagnostics;
             namespace Implab {
                 /// <summary>
                 /// Класс для асинхронного получения результатов. Так называемое "обещание".
                 /// </summary>
                 /// <typeparam name="T">Тип получаемого результата</typeparam>
                 /// <remarks>
                 /// <para>Сервис при обращении к его методу дает обещаиние о выполнении операции,
                 /// клиент получив такое обещание может установить ряд обратных вызово для получения
                 /// событий выполнения обещания, тоесть завершения операции и предоставлении результатов.</para>
                 /// <para>
                 /// Обещение может быть как выполнено, так и выполнено с ошибкой. Для подписки на
                 /// данные события клиент должен использовать методы <c>Then</c>.
                 /// </para>
                 /// <para>
                 /// Сервис, в свою очередь, по окончанию выполнения операции (возможно с ошибкой),
                 /// использует методы <c>Resolve</c> либо <c>Reject</c> для оповещения клиетна о
                 /// выполнении обещания.
                 /// </para>
                 /// <para>
                 /// Если сервер успел выполнить обещание еще до того, как клиент на него подписался,
                 /// то в момент подписки клиента будут вызваны соответсвующие события в синхронном
                 /// режиме и клиент будет оповещен в любом случае. Иначе, обработчики добавляются в
                 /// список в порядке подписания и в этом же порядке они будут вызваны при выполнении
                 /// обещания.
                 /// </para>
                 /// <para>
                 /// Обрабатывая результаты обещания можно преобразовывать результаты либо инициировать
                 /// связанные асинхронные операции, которые также возвращают обещания. Для этого следует
                 /// использовать соответствующую форму методе <c>Then</c>.
                 /// </para>
                 /// <para>
                 /// Также хорошим правилом является то, что <c>Resolve</c> и <c>Reject</c> должен вызывать
                 /// только инициатор обещания иначе могут возникнуть противоречия.
                 /// </para>
                 /// </remarks>
                 public class Promise<T> : AbstractPromise<IDeferred<T>>, IPromise<T>, IDeferred<T> {
                     class StubDeferred : IDeferred<T> {
                         public static readonly StubDeferred instance = new StubDeferred();
                         StubDeferred() {
                         }
                         #region IDeferred implementation
                         public void Resolve(T value) {
                         }
                         public void Reject(Exception error) {
                         }
                         #endregion
                         #region ICancellable implementation
                         public void Cancel() {
                         }
                         #endregion
                     }
                     class RemapDescriptor<T2> : IDeferred<T> {
                         readonly Func<T,T2> m_remap;
                         readonly Func<Exception,T2> m_failed;
                         readonly Func<T2> m_cancel;
                         readonly IDeferred<T2> m_deferred;
                         public RemapDescriptor(Func<T,T2> remap, Func<Exception,T2> failed, Func<T2> cancel, IDeferred<T2> deferred ) {
                             Debug.Assert(deferred != null);
                             m_remap = remap;
                             m_failed = failed;
                             m_cancel = cancel;
                             m_deferred = deferred;
                         }
                         #region IDeferred implementation
                         public void Resolve(T value) {
                             if (m_remap != null) {
                                 try {
                                     m_deferred.Resolve(m_remap(value));
                                 } catch (Exception ex) {
                                     Reject(ex);
                                 }
                             }
                         }
                         public void Reject(Exception error) {
                             if (m_failed != null) {
                                 try {
                                     m_deferred.Resolve(m_failed(error));
                                 } catch (Exception ex) {
                                     m_deferred.Reject(ex);
                                 }
                             } else {
                                 m_deferred.Reject(error);
                             }
                         }
                         #endregion
                         #region ICancellable implementation
                         public void Cancel() {
                             if (m_cancel != null) {
                                 try {
                                     m_deferred.Resolve(m_cancel());
                                 } catch (Exception ex) {
                                     Reject(ex);
                                 }
                             } else {
                                 m_deferred.Cancel();
                             }
                         }
                         #endregion
                     }
                     class ListenerDescriptor : IDeferred<T> {
                         readonly Action m_handler;
                         readonly PromiseEventType m_events;
                         public ListenerDescriptor(Action handler, PromiseEventType events) {
                             Debug.Assert(handler != null);
                             m_handler = handler;
                             m_events = events;
                         }
                         #region IDeferred implementation
                         public void Resolve(T value) {
                             if (m_events.HasFlag(PromiseEventType.Success)) {
                                 try {
                                     m_handler();
                                 // Analysis disable once EmptyGeneralCatchClause
                                 } catch {
                                 }
                             }
                         }
                         public void Reject(Exception error) {
                             if (m_events.HasFlag(PromiseEventType.Error)){
                                 try {
                                     m_handler();
                                     // Analysis disable once EmptyGeneralCatchClause
                                 } catch {
                                 }
                             }
                         }
                         #endregion
                         #region ICancellable implementation
                         public void Cancel() {
                             if (m_events.HasFlag(PromiseEventType.Cancelled)){
                                 try {
                                     m_handler();
                                     // Analysis disable once EmptyGeneralCatchClause
                                 } catch {
                                 }
                             }
                         }
                         #endregion
                     }
                     class ValueEventDescriptor : IDeferred<T> {
                         readonly Action<T> m_success;
                         readonly Action<Exception> m_failed;
                         readonly Action m_cancelled;
                         readonly IDeferred<T> m_deferred;
                         public ValueEventDescriptor(Action<T> success, Action<Exception> failed, Action cancelled, IDeferred<T> deferred) {
                             Debug.Assert(deferred != null);
                             m_success = success;
                             m_failed = failed;
                             m_cancelled = cancelled;
                             m_deferred = deferred;
                         }
                         #region IDeferred implementation
                         public void Resolve(T value) {
                             if (m_success != null) {
                                 try {
                                     m_success(value);
                                     m_deferred.Resolve(value);
                                 } catch (Exception ex) {
                                     Reject(ex);
                                 }
                             }
                         }
                         public void Reject(Exception error) {
                             if (m_failed != null) {
                                 try {
                                     m_failed(error);
                                     m_deferred.Resolve(default(T));
                                 } catch(Exception ex) {
                                     m_deferred.Reject(ex);
                                 }
                             } else {
                                 m_deferred.Reject(error);
                             }
                         }
                         #endregion
                         #region ICancellable implementation
                         public void Cancel() {
                             if (m_cancelled != null) {
                                 try {
                                     m_cancelled();
                                     m_deferred.Resolve(default(T));
                                 } catch(Exception ex) {
                                     Reject(ex);
                                 }
                             } else {
                                 m_deferred.Cancel();
                             }
                         }
                         #endregion
                     }
                     public class EventDescriptor : IDeferred<T> {
                         readonly Action m_success;
                         readonly Action<Exception> m_failed;
                         readonly Action m_cancelled;
                         readonly IDeferred<T> m_deferred;
                         public EventDescriptor(Action success, Action<Exception> failed, Action cancelled, IDeferred<T> deferred) {
                             Debug.Assert(deferred != null);
                             m_success = success;
                             m_failed = failed;
                             m_cancelled = cancelled;
                             m_deferred = deferred;
                         }
                         #region IDeferred implementation
                         public void Resolve(T value) {
                             if (m_success != null) {
                                 try {
                                     m_success();
                                     m_deferred.Resolve(value);
                                 } catch (Exception ex) {
                                     Reject(ex);
                                 }
                             }
                         }
                         public void Reject(Exception error) {
                             if (m_failed != null) {
                                 try {
                                     m_failed(error);
                                     m_deferred.Resolve(default(T));
                                 }catch (Exception ex)
                                 {
                                     m_deferred.Reject(ex);
                                 }
                             } else {
                                 m_deferred.Reject(error);
                             }
                         }
                         #endregion
                         #region ICancellable implementation
                         public void Cancel() {
                             if (m_cancelled != null) {
                                 try {
                                     m_cancelled();
                                     m_deferred.Resolve(default(T));
                                 } catch (Exception ex) {
                                     Reject(ex);
                                 }
                             } else {
                                 m_deferred.Cancel();
                             }
                         }
                         #endregion
                     }
                     T m_result;
                     public virtual void Resolve(T value) {
-                        BeginSetResult();
+                        if (BeginSetResult()) {
-                        m_result = value;
+                            m_result = value;
-                        EndSetResult();
+                            EndSetResult();
+                        }
                     }
                     public void Reject(Exception error) {
                         SetError(error);
                     }
                     public Type PromiseType {
                         get {
                             return typeof(T);
                         }
                     }
                     public new T Join() {
                         WaitResult(-1);
                         return m_result;
                     }
                     public new T Join(int timeout) {
                         WaitResult(timeout);
                         return m_result;
                     }
                     public IPromise<T> On(Action<T> success, Action<Exception> error, Action cancel) {
                         AddHandler(new ValueEventDescriptor(success, error, cancel, StubDeferred.instance));
                         return this;
                     }
                     public IPromise<T> On(Action<T> success, Action<Exception> error) {
                         AddHandler(new ValueEventDescriptor(success, error, null, StubDeferred.instance));
                         return this;
                     }
                     public IPromise<T> On(Action<T> success) {
                         AddHandler(new ValueEventDescriptor(success, null, null, StubDeferred.instance));
                         return this;
                     }
                     public IPromise<T> On(Action handler, PromiseEventType events) {
                         Listen(events, handler);
                         return this;
                     }
                     public IPromise<T2> Then<T2>(Func<T, T2> mapper, Func<Exception, T2> error, Func<T2> cancel) {
                         var promise = new Promise<T2>();
                         AddHandler(new RemapDescriptor<T2>(mapper, error, cancel, promise));
                         return promise;
                     }
                     public IPromise<T2> Then<T2>(Func<T, T2> mapper, Func<Exception, T2> error) {
                         var promise = new Promise<T2>();
                         AddHandler(new RemapDescriptor<T2>(mapper, error, null, promise));
                         return promise;
                     }
                     public IPromise<T2> Then<T2>(Func<T, T2> mapper) {
                         var promise = new Promise<T2>();
                         AddHandler(new RemapDescriptor<T2>(mapper, null, null, promise));
                         return promise;
                     }
                     public IPromise<T2> Chain<T2>(Func<T, IPromise<T2>> chained, Func<Exception, IPromise<T2>> error, Func<IPromise<T2>> cancel) {
                         // this promise will be resolved when an asyc operation is started
                         var promise = new Promise<IPromise<T2>>();
                         AddHandler(new RemapDescriptor<IPromise<T2>>(
                             chained,
                             error,
                             cancel,
                             promise
                         ));
                         var medium = new Promise<T2>();
                         if (chained != null)
                             medium.On(Cancel, PromiseEventType.Cancelled);
                         // we need to connect started async operation with the medium
                         // if the async operation hasn't been started by the some reason
                         // report is to the medium
                         promise.On(
                             result => ConnectPromise<T2>(result, medium),
                             medium.Reject,
                             medium.Cancel
                         );
                         return medium;
                     }
                     static void ConnectPromise<T2>(IPromise<T2> result, Promise<T2> medium) {
                         if (result != null) {
                             result.On(
                                 medium.Resolve,
                                 medium.Reject,
                                 () => medium.Reject(new OperationCanceledException())
                             );
                             medium.On(result.Cancel, PromiseEventType.Cancelled);
                         } else {
                             medium.Reject(
                                 new NullReferenceException(
                                     "The chained asynchronous operation returned" +
                                     " 'null' where the promise instance is expected"
                                 )
                             );
                         }
                     }
                     public IPromise<T2> Chain<T2>(Func<T, IPromise<T2>> chained, Func<Exception, IPromise<T2>> error) {
                         return Chain(chained, error, null);
                     }
                     public IPromise<T2> Chain<T2>(Func<T, IPromise<T2>> chained) {
                         return Chain(chained, null, null);
                     }
                     public IPromise<T2> Error<T2>(Func<Exception, T2> error) {
                         var promise = new Promise<T2>();
                         if (error != null)
                             On(
                                 (Action<T>)null,
                                 ex => {
                                     try {
                                         promise.Resolve(error(ex));
                                     } catch (Exception ex2) {
                                         promise.Reject(ex2);
                                     }
                                 }
                             );
                         else
                             Listen(PromiseEventType.Error, () => promise.Resolve(default(T2)));
                         return promise;
                     }
                     public IPromise<T2> Cancelled<T2>(Func<T2> handler) {
                         var promise = new Promise<T2>();
                         if (handler != null)
                             On(
                                 (Action<T>)null,
                                 null,
                                 () => {
                                     try {
                                         promise.Resolve(handler());
                                     } catch (Exception ex) {
                                         promise.Reject(ex);
                                     }
                             });
                         else
                             Listen(PromiseEventType.Cancelled, () => promise.Resolve(default(T2)));
                         return promise;
                     }
                     public IPromise Then(Action success, Action<Exception> error, Action cancel) {
                         var promise = new Promise<T>();
                         if (success != null)
                             promise.On(Cancel, PromiseEventType.Cancelled);
                         AddHandler(new EventDescriptor(success, error, cancel, promise));
                         return promise;
                     }
                     public IPromise Then(Action success, Action<Exception> error) {
                         return Then(success, error, null);
                     }
                     public IPromise Then(Action success) {
                         return Then(success, null, null);
                     }
                     public IPromise Chain(Func<IPromise> chained, Func<Exception, IPromise> error, Func<IPromise> cancel) {
                         var promise = new Promise<IPromise>();
                         AddHandler(
                             new RemapDescriptor<IPromise>(
                                 x => chained(),
                                 error,
                                 cancel,
                                 promise
                             )
                         );
                         var medium = new Promise();
                         if (chained != null)
                             medium.On(Cancel, PromiseEventType.Cancelled);
                         promise.On(
                             result => ConnectPromise(result, medium),
                             medium.Reject,
                             medium.Cancel
                         );
                         return medium;
                     }
                     static void ConnectPromise(IPromise result, Promise medium) {
                         if (result != null) {
                             result.On(
                                 medium.Resolve,
                                 medium.Reject,
                                 () => medium.Reject(new OperationCanceledException())
                             );
                             medium.On(result.Cancel, PromiseEventType.Cancelled);
                         } else {
                             medium.Reject(
                                 new NullReferenceException(
                                     "The chained asynchronous operation returned" +
                                     " 'null' where the promise instance is expected"
                                 )
                             );
                         }
                     }
                     public IPromise Chain(Func<IPromise> chained, Func<Exception, IPromise> error) {
                         return Chain(chained, error, null);
                     }
                     public IPromise Chain(Func<IPromise> chained) {
                         return Chain(chained, null, null);
                     }
                     public IPromise On(Action success, Action<Exception> error, Action cancel) {
                         AddHandler(new EventDescriptor(success,error,cancel, StubDeferred.instance));
                         return this;
                     }
                     public IPromise On(Action success, Action<Exception> error) {
                         AddHandler(new EventDescriptor(success, error, null, StubDeferred.instance));
                         return this;
                     }
                     public IPromise On(Action success) {
                         Listen(PromiseEventType.Success, success);
                         return this;
                     }
                     IPromise IPromise.On(Action handler, PromiseEventType events) {
                         Listen(events,handler);
                         return this;
                     }
                     public IPromise Error(Action<Exception> error) {
                         var promise = new Promise();
                         if (error != null)
                             On(
                                 (Action<T>)null,
                                 ex => {
                                 try {
                                     error(ex);
                                     promise.Resolve();
                                 } catch (Exception ex2) {
                                     promise.Reject(ex2);
                                 }
                             });
                         else
                             Listen(PromiseEventType.Error, promise.Resolve);
                         return promise;
                     }
                     public IPromise Cancelled(Action handler) {
                         var promise = new Promise();
                         if (handler != null)
                             On(
                                 (Action<T>)null,
                                 null,
                                 () => {
                                     try {
                                         handler();
                                         promise.Resolve();
                                     } catch (Exception ex) {
                                         promise.Reject(ex);
                                     }
                                 });
                         else
                             Listen(PromiseEventType.Cancelled, promise.Resolve);
                         return promise;
                     }
                     public IPromise<T2> Cast<T2>() {
                         return (IPromise<T2>)this;
                     }
                     #region implemented abstract members of AbstractPromise
                     protected override void SignalSuccess(IDeferred<T> handler) {
                         handler.Resolve(m_result);
                     }
                     protected override void SignalError(IDeferred<T> handler, Exception error) {
                         handler.Reject(error);
                     }
                     protected override void SignalCancelled(IDeferred<T> handler) {
                         handler.Cancel();
                     }
                     protected override void Listen(PromiseEventType events, Action handler) {
                         if (handler != null)
                             AddHandler(new ListenerDescriptor(handler, events));
                     }
                     #endregion
                     public static IPromise<T> ResultToPromise(T value) {
                         var p = new Promise<T>();
                         p.Resolve(value);
                         return p;
                     }
                     public static IPromise<T> ExceptionToPromise(Exception error) {
                         var p = new Promise<T>();
                         p.Reject(error);
                         return p;
                     }
                 }
             }

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