implab/ImplabNet Commit - r137:238e15580926

added the blocking queue

cin -

r137:238e15580926 v2

parent child

Context file:

r137:238e15580926

Collapse all files

Implab/Parallels/BlockingQueue.cs created 644 +87 0

@@ -0,0 +1,87
	1	using System;
	2	using System.Threading;
	3
	4	namespace Implab.Parallels {
	5	public class BlockingQueue<T> : AsyncQueue<T> {
	6	readonly object m_lock = new object();
	7
	8	public override void Enqueue(T value) {
	9	base.Enqueue(value);
	10	lock (m_lock)
	11	Monitor.Pulse(m_lock);
	12	}
	13
	14	public override void EnqueueRange(T[] data, int offset, int length) {
	15	base.EnqueueRange(data, offset, length);
	16	if (length > 1)
	17	lock (m_lock)
	18	Monitor.PulseAll(m_lock);
	19	else
	20	lock (m_lock)
	21	Monitor.Pulse(m_lock);
	22	}
	23
	24	public T GetItem(int timeout) {
	25	T item;
	26	var t1 = Environment.TickCount;
	27	var dt = timeout;
	28	while (!TryDequeue(out item)) {
	29	lock (m_lock)
	30	if (!Monitor.Wait(m_lock, dt))
	31	throw new TimeoutException();
	32	if (timeout >= 0) {
	33	dt = timeout - Environment.TickCount + t1;
	34	if (dt < 0)
	35	throw new TimeoutException();
	36	}
	37	}
	38	return item;
	39	}
	40
	41	public T GetItem() {
	42	T item;
	43	while (!TryDequeue(out item))
	44	lock (m_lock)
	45	Monitor.Wait(m_lock);
	46	return item;
	47	}
	48
	49	public T[] GetRange(int max, int timeout) {
	50	Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max");
	51
	52	var buffer = new T[max];
	53	int actual;
	54	var t1 = Environment.TickCount;
	55	var dt = timeout;
	56	while (!TryDequeueRange(buffer,0,max,out actual)) {
	57	lock (m_lock)
	58	if (!Monitor.Wait(m_lock, dt))
	59	throw new TimeoutException();
	60	if (timeout >= 0) {
	61	dt = timeout - Environment.TickCount + t1;
	62	if (dt < 0)
	63	throw new TimeoutException();
	64	}
	65	}
	66
	67	var data = new T[actual];
	68	Array.Copy(buffer, data, actual);
	69	return data;
	70	}
	71
	72	public T[] GetRange(int max) {
	73	Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max");
	74
	75	var buffer = new T[max];
	76	int actual;
	77	while (!TryDequeueRange(buffer, 0, max, out actual))
	78	lock (m_lock)
	79	Monitor.Wait(m_lock);
	80
	81	var data = new T[actual];
	82	Array.Copy(buffer, data, actual);
	83	return data;
	84	}
	85	}
	86	}
	87

Implab/Implab.csproj +1 0

             <?xml version="1.0" encoding="utf-8"?>
             <Project DefaultTargets="Build" ToolsVersion="4.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
               <PropertyGroup>
                 <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
                 <Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform>
                 <ProjectGuid>{F550F1F8-8746-4AD0-9614-855F4C4B7F05}</ProjectGuid>
                 <OutputType>Library</OutputType>
                 <RootNamespace>Implab</RootNamespace>
                 <AssemblyName>Implab</AssemblyName>
               </PropertyGroup>
               <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
                 <DebugSymbols>true</DebugSymbols>
                 <DebugType>full</DebugType>
                 <Optimize>false</Optimize>
                 <OutputPath>bin\Debug</OutputPath>
                 <DefineConstants>TRACE;DEBUG;</DefineConstants>
                 <ErrorReport>prompt</ErrorReport>
                 <WarningLevel>4</WarningLevel>
                 <ConsolePause>false</ConsolePause>
                 <RunCodeAnalysis>true</RunCodeAnalysis>
               </PropertyGroup>
               <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
                 <DebugType>full</DebugType>
                 <Optimize>true</Optimize>
                 <OutputPath>bin\Release</OutputPath>
                 <ErrorReport>prompt</ErrorReport>
                 <WarningLevel>4</WarningLevel>
                 <ConsolePause>false</ConsolePause>
               </PropertyGroup>
               <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug 4.5|AnyCPU' ">
                 <DebugSymbols>true</DebugSymbols>
                 <DebugType>full</DebugType>
                 <Optimize>false</Optimize>
                 <OutputPath>bin\Debug</OutputPath>
                 <DefineConstants>TRACE;DEBUG;NET_4_5</DefineConstants>
                 <ErrorReport>prompt</ErrorReport>
                 <WarningLevel>4</WarningLevel>
                 <RunCodeAnalysis>true</RunCodeAnalysis>
                 <ConsolePause>false</ConsolePause>
               </PropertyGroup>
               <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release 4.5|AnyCPU' ">
                 <Optimize>true</Optimize>
                 <OutputPath>bin\Release</OutputPath>
                 <ErrorReport>prompt</ErrorReport>
                 <WarningLevel>4</WarningLevel>
                 <ConsolePause>false</ConsolePause>
                 <DefineConstants>NET_4_5</DefineConstants>
               </PropertyGroup>
               <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'DebugMono|AnyCPU' ">
                 <DebugSymbols>true</DebugSymbols>
                 <DebugType>full</DebugType>
                 <Optimize>false</Optimize>
                 <OutputPath>bin\Debug</OutputPath>
                 <DefineConstants>TRACE;DEBUG;NET_4_5;MONO</DefineConstants>
                 <ErrorReport>prompt</ErrorReport>
                 <WarningLevel>4</WarningLevel>
                 <RunCodeAnalysis>true</RunCodeAnalysis>
                 <ConsolePause>false</ConsolePause>
               </PropertyGroup>
               <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'ReleaseMono|AnyCPU' ">
                 <Optimize>true</Optimize>
                 <OutputPath>bin\Release</OutputPath>
                 <DefineConstants>NET_4_5;MONO;</DefineConstants>
                 <ErrorReport>prompt</ErrorReport>
                 <WarningLevel>4</WarningLevel>
                 <ConsolePause>false</ConsolePause>
               </PropertyGroup>
               <ItemGroup>
                 <Reference Include="System" />
                 <Reference Include="System.Xml" />
               </ItemGroup>
               <ItemGroup>
                 <Compile Include="Component.cs" />
                 <Compile Include="CustomEqualityComparer.cs" />
                 <Compile Include="Diagnostics\ConsoleTraceListener.cs" />
                 <Compile Include="Diagnostics\EventText.cs" />
                 <Compile Include="Diagnostics\LogChannel.cs" />
                 <Compile Include="Diagnostics\LogicalOperation.cs" />
                 <Compile Include="Diagnostics\TextFileListener.cs" />
                 <Compile Include="Diagnostics\TraceLog.cs" />
                 <Compile Include="Diagnostics\TraceEvent.cs" />
                 <Compile Include="Diagnostics\TraceEventType.cs" />
                 <Compile Include="Disposable.cs" />
                 <Compile Include="ICancellable.cs" />
                 <Compile Include="IProgressHandler.cs" />
                 <Compile Include="IProgressNotifier.cs" />
                 <Compile Include="IPromiseT.cs" />
                 <Compile Include="IPromise.cs" />
                 <Compile Include="IServiceLocator.cs" />
                 <Compile Include="ITaskController.cs" />
                 <Compile Include="JSON\JSONElementContext.cs" />
                 <Compile Include="JSON\JSONElementType.cs" />
                 <Compile Include="JSON\JSONGrammar.cs" />
                 <Compile Include="JSON\JSONParser.cs" />
                 <Compile Include="JSON\JSONScanner.cs" />
                 <Compile Include="JSON\JsonTokenType.cs" />
                 <Compile Include="JSON\JSONWriter.cs" />
                 <Compile Include="JSON\JSONXmlReader.cs" />
                 <Compile Include="JSON\JSONXmlReaderOptions.cs" />
                 <Compile Include="JSON\StringTranslator.cs" />
                 <Compile Include="Parallels\DispatchPool.cs" />
                 <Compile Include="Parallels\ArrayTraits.cs" />
                 <Compile Include="Parallels\MTQueue.cs" />
                 <Compile Include="Parallels\WorkerPool.cs" />
                 <Compile Include="Parsing\Alphabet.cs" />
                 <Compile Include="Parsing\AlphabetBase.cs" />
                 <Compile Include="Parsing\AltToken.cs" />
                 <Compile Include="Parsing\BinaryToken.cs" />
                 <Compile Include="Parsing\CatToken.cs" />
                 <Compile Include="Parsing\CDFADefinition.cs" />
                 <Compile Include="Parsing\DFABuilder.cs" />
                 <Compile Include="Parsing\DFADefinitionBase.cs" />
                 <Compile Include="Parsing\DFAStateDescriptor.cs" />
                 <Compile Include="Parsing\DFAutomaton.cs" />
                 <Compile Include="Parsing\EDFADefinition.cs" />
                 <Compile Include="Parsing\EmptyToken.cs" />
                 <Compile Include="Parsing\EndToken.cs" />
                 <Compile Include="Parsing\EnumAlphabet.cs" />
                 <Compile Include="Parsing\Grammar.cs" />
                 <Compile Include="Parsing\IAlphabet.cs" />
                 <Compile Include="Parsing\IDFADefinition.cs" />
                 <Compile Include="Parsing\IVisitor.cs" />
                 <Compile Include="Parsing\ParserException.cs" />
                 <Compile Include="Parsing\Scanner.cs" />
                 <Compile Include="Parsing\StarToken.cs" />
                 <Compile Include="Parsing\SymbolToken.cs" />
                 <Compile Include="Parsing\Token.cs" />
                 <Compile Include="ServiceLocator.cs" />
                 <Compile Include="TaskController.cs" />
                 <Compile Include="ProgressInitEventArgs.cs" />
                 <Compile Include="Properties\AssemblyInfo.cs" />
                 <Compile Include="Parallels\AsyncPool.cs" />
                 <Compile Include="Safe.cs" />
                 <Compile Include="ValueEventArgs.cs" />
                 <Compile Include="PromiseExtensions.cs" />
                 <Compile Include="SyncContextPromise.cs" />
                 <Compile Include="Diagnostics\OperationContext.cs" />
                 <Compile Include="Diagnostics\TraceContext.cs" />
                 <Compile Include="Diagnostics\LogEventArgs.cs" />
                 <Compile Include="Diagnostics\LogEventArgsT.cs" />
                 <Compile Include="Diagnostics\Extensions.cs" />
                 <Compile Include="IComponentContainer.cs" />
                 <Compile Include="PromiseEventType.cs" />
                 <Compile Include="ComponentContainer.cs" />
                 <Compile Include="DisposablePool.cs" />
                 <Compile Include="ObjectPool.cs" />
                 <Compile Include="Parallels\AsyncQueue.cs" />
                 <Compile Include="PromiseT.cs" />
                 <Compile Include="IDeferred.cs" />
                 <Compile Include="IDeferredT.cs" />
                 <Compile Include="AbstractPromise.cs" />
                 <Compile Include="Promise.cs" />
                 <Compile Include="PromiseTransientException.cs" />
                 <Compile Include="Parallels\Signal.cs" />
                 <Compile Include="Parallels\SharedLock.cs" />
                 <Compile Include="Diagnostics\ILogWriter.cs" />
                 <Compile Include="Diagnostics\ListenerBase.cs" />
+                <Compile Include="Parallels\BlockingQueue.cs" />
               </ItemGroup>
               <Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" />
               <ItemGroup />
               <ProjectExtensions>
                 <MonoDevelop>
                   <Properties>
                     <Policies>
                       <CSharpFormattingPolicy IndentSwitchBody="True" NamespaceBraceStyle="EndOfLine" ClassBraceStyle="EndOfLine" InterfaceBraceStyle="EndOfLine" StructBraceStyle="EndOfLine" EnumBraceStyle="EndOfLine" MethodBraceStyle="EndOfLine" ConstructorBraceStyle="EndOfLine" DestructorBraceStyle="EndOfLine" BeforeMethodDeclarationParentheses="False" BeforeMethodCallParentheses="False" BeforeConstructorDeclarationParentheses="False" NewLineBeforeConstructorInitializerColon="NewLine" NewLineAfterConstructorInitializerColon="SameLine" BeforeIndexerDeclarationBracket="False" BeforeDelegateDeclarationParentheses="False" NewParentheses="False" SpacesBeforeBrackets="False" inheritsSet="Mono" inheritsScope="text/x-csharp" scope="text/x-csharp" />
                       <TextStylePolicy FileWidth="120" EolMarker="Unix" inheritsSet="VisualStudio" inheritsScope="text/plain" scope="text/x-csharp" />
                       <DotNetNamingPolicy DirectoryNamespaceAssociation="PrefixedHierarchical" ResourceNamePolicy="MSBuild" />
                       <TextStylePolicy FileWidth="120" TabsToSpaces="False" inheritsSet="VisualStudio" inheritsScope="text/plain" scope="application/xml" />
                       <XmlFormattingPolicy inheritsSet="Mono" inheritsScope="application/xml" scope="application/xml" />
                       <TextStylePolicy FileWidth="120" TabsToSpaces="False" inheritsSet="VisualStudio" inheritsScope="text/plain" scope="text/plain" />
                       <NameConventionPolicy>
                         <Rules>
                           <NamingRule Name="Namespaces" AffectedEntity="Namespace" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
                           <NamingRule Name="Types" AffectedEntity="Class, Struct, Enum, Delegate" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
                           <NamingRule Name="Interfaces" AffectedEntity="Interface" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True">
                             <RequiredPrefixes>
                               <String>I</String>
                             </RequiredPrefixes>
                           </NamingRule>
                           <NamingRule Name="Attributes" AffectedEntity="CustomAttributes" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True">
                             <RequiredSuffixes>
                               <String>Attribute</String>
                             </RequiredSuffixes>
                           </NamingRule>
                           <NamingRule Name="Event Arguments" AffectedEntity="CustomEventArgs" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True">
                             <RequiredSuffixes>
                               <String>EventArgs</String>
                             </RequiredSuffixes>
                           </NamingRule>
                           <NamingRule Name="Exceptions" AffectedEntity="CustomExceptions" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True">
                             <RequiredSuffixes>
                               <String>Exception</String>
                             </RequiredSuffixes>
                           </NamingRule>
                           <NamingRule Name="Methods" AffectedEntity="Methods" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
                           <NamingRule Name="Static Readonly Fields" AffectedEntity="ReadonlyField" VisibilityMask="Internal, Protected, Public" NamingStyle="CamelCase" IncludeInstanceMembers="False" IncludeStaticEntities="True" />
                           <NamingRule Name="Fields (Non Private)" AffectedEntity="Field" VisibilityMask="Internal, Public" NamingStyle="CamelCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
                           <NamingRule Name="ReadOnly Fields (Non Private)" AffectedEntity="ReadonlyField" VisibilityMask="Internal, Public" NamingStyle="CamelCase" IncludeInstanceMembers="True" IncludeStaticEntities="False" />
                           <NamingRule Name="Fields (Private)" AffectedEntity="Field, ReadonlyField" VisibilityMask="Private, Protected" NamingStyle="CamelCase" IncludeInstanceMembers="True" IncludeStaticEntities="False">
                             <RequiredPrefixes>
                               <String>m_</String>
                             </RequiredPrefixes>
                           </NamingRule>
                           <NamingRule Name="Static Fields (Private)" AffectedEntity="Field" VisibilityMask="Private" NamingStyle="CamelCase" IncludeInstanceMembers="False" IncludeStaticEntities="True">
                             <RequiredPrefixes>
                               <String>_</String>
                             </RequiredPrefixes>
                           </NamingRule>
                           <NamingRule Name="ReadOnly Fields (Private)" AffectedEntity="ReadonlyField" VisibilityMask="Private, Protected" NamingStyle="CamelCase" IncludeInstanceMembers="True" IncludeStaticEntities="False">
                             <RequiredPrefixes>
                               <String>m_</String>
                             </RequiredPrefixes>
                           </NamingRule>
                           <NamingRule Name="Constant Fields" AffectedEntity="ConstantField" VisibilityMask="VisibilityMask" NamingStyle="AllUpper" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
                           <NamingRule Name="Properties" AffectedEntity="Property" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
                           <NamingRule Name="Events" AffectedEntity="Event" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
                           <NamingRule Name="Enum Members" AffectedEntity="EnumMember" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
                           <NamingRule Name="Parameters" AffectedEntity="Parameter, LocalVariable" VisibilityMask="VisibilityMask" NamingStyle="CamelCase" IncludeInstanceMembers="True" IncludeStaticEntities="True" />
                           <NamingRule Name="Type Parameters" AffectedEntity="TypeParameter" VisibilityMask="VisibilityMask" NamingStyle="PascalCase" IncludeInstanceMembers="True" IncludeStaticEntities="True">
                             <RequiredPrefixes>
                               <String>T</String>
                             </RequiredPrefixes>
                           </NamingRule>
                         </Rules>
                       </NameConventionPolicy>
                     </Policies>
                   </Properties>
                 </MonoDevelop>
               </ProjectExtensions>
             </Project>

Implab/Parallels/AsyncQueue.cs +2 -2

             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) {
+                    public virtual 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) {
+                    public virtual 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
                 }
             }

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