implab/ImplabNet Commit - r233:d6fe09f5592c

Improved AsyncQueue...

cin -

r233:d6fe09f5592c v2

parent child

Context file:

r233:d6fe09f5592c

Collapse all files

Implab/Parallels/SimpleAsyncQueue.cs created 644 +131 0

@@ -0,0 +1,131
	1	using System.Threading;
	2	using System.Collections.Generic;
	3	using System;
	4	using System.Collections;
	5
	6	namespace Implab.Parallels {
	7	public class SimpleAsyncQueue<T> : IEnumerable<T> {
	8	class Node {
	9	public Node(T value) {
	10	this.value = value;
	11	}
	12	public readonly T value;
	13	public volatile Node next;
	14	}
	15
	16	// the reader and the writer are mainteined completely independent,
	17	// the reader can read next item when m_first.next is not null
	18	// the writer creates the a new node, moves m_last to this node and
	19	// only after that restores the reference from the previous node
	20	// making available the reader to read the new node.
	21
	22	Node m_first; // position on the node which is already read
	23	Node m_last; // position on the node which is already written
	24
	25	public SimpleAsyncQueue() {
	26	m_first = m_last = new Node(default(T));
	27	}
	28
	29	public void Enqueue(T value) {
	30	var next = new Node(value);
	31
	32	// Interlocaked.CompareExchange implies Thread.MemoryBarrier();
	33	// to ensure that the next node is completely constructed
	34	var last = Interlocked.Exchange(ref m_last, next);
	35
	36	// release-fence
	37	last.next = next;
	38
	39	}
	40
	41	public bool TryDequeue(out T value) {
	42	Node first;
	43	Node next;
	44
	45	Thread.MemoryBarrier(); // ensure m_first is fresh
	46	SpinWait spin = new SpinWait();
	47	do {
	48	first = m_first;
	49	// aquire-fence
	50	next = first.next;
	51	if (next == null) {
	52	value = default(T);
	53	return false;
	54	}
	55
	56	if (first == Interlocked.CompareExchange(ref m_first, next, first))
	57	// head succesfully updated
	58	break;
	59	spin.SpinOnce();
	60	} while (true);
	61
	62	value = next.value;
	63	return true;
	64	}
	65
	66	#region IEnumerable implementation
	67
	68	class Enumerator : IEnumerator<T> {
	69	Node m_current;
	70	Node m_first;
	71
	72	public Enumerator(Node first) {
	73	m_first = first;
	74	}
	75
	76	#region IEnumerator implementation
	77
	78	public bool MoveNext() {
	79	m_current = m_current == null ? m_first : m_current.next;
	80	return m_current != null;
	81	}
	82
	83	public void Reset() {
	84	m_current = null;
	85	}
	86
	87	object IEnumerator.Current {
	88	get {
	89	if (m_current == null)
	90	throw new InvalidOperationException();
	91	return m_current.value;
	92	}
	93	}
	94
	95	#endregion
	96
	97	#region IDisposable implementation
	98
	99	public void Dispose() {
	100	}
	101
	102	#endregion
	103
	104	#region IEnumerator implementation
	105
	106	public T Current {
	107	get {
	108	if (m_current == null)
	109	throw new InvalidOperationException();
	110	return m_current.value;
	111	}
	112	}
	113
	114	#endregion
	115	}
	116
	117	public IEnumerator<T> GetEnumerator() {
	118	return new Enumerator(m_first);
	119	}
	120
	121	#endregion
	122
	123	#region IEnumerable implementation
	124
	125	IEnumerator IEnumerable.GetEnumerator() {
	126	return GetEnumerator();
	127	}
	128
	129	#endregion
	130	}
	131	}

Implab.Playground/App.config +3 -3

-            <?xml version="1.0" encoding="utf-8" ?>
+            <?xml version="1.0" encoding="utf-8"?>
             <configuration>
                 <startup>
-                    <supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.5.2" />
+                    <supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.5"/>
                 </startup>
-            </configuration>
  No newline at end of file
+            </configuration>

Implab.Playground/Implab.Playground.csproj +2 -1

             <?xml version="1.0" encoding="utf-8"?>
             <Project ToolsVersion="14.0" DefaultTargets="Build" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
               <Import Project="$(MSBuildExtensionsPath)\$(MSBuildToolsVersion)\Microsoft.Common.props" Condition="Exists('$(MSBuildExtensionsPath)\$(MSBuildToolsVersion)\Microsoft.Common.props')" />
               <PropertyGroup>
                 <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
                 <Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform>
                 <ProjectGuid>{100DFEB0-75BE-436F-ADDF-1F46EF433F46}</ProjectGuid>
                 <OutputType>Exe</OutputType>
                 <AppDesignerFolder>Properties</AppDesignerFolder>
                 <RootNamespace>Implab.Playground</RootNamespace>
                 <AssemblyName>Implab.Playground</AssemblyName>
-                <TargetFrameworkVersion>v4.5.2</TargetFrameworkVersion>
+                <TargetFrameworkVersion>v4.5</TargetFrameworkVersion>
                 <FileAlignment>512</FileAlignment>
                 <AutoGenerateBindingRedirects>true</AutoGenerateBindingRedirects>
+                <TargetFrameworkProfile />
               </PropertyGroup>
               <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
                 <PlatformTarget>AnyCPU</PlatformTarget>
                 <DebugSymbols>true</DebugSymbols>
                 <DebugType>full</DebugType>
                 <Optimize>false</Optimize>
                 <OutputPath>bin\Debug\</OutputPath>
                 <DefineConstants>DEBUG;TRACE</DefineConstants>
                 <ErrorReport>prompt</ErrorReport>
                 <WarningLevel>4</WarningLevel>
               </PropertyGroup>
               <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
                 <PlatformTarget>AnyCPU</PlatformTarget>
                 <DebugType>pdbonly</DebugType>
                 <Optimize>true</Optimize>
                 <OutputPath>bin\Release\</OutputPath>
                 <DefineConstants>TRACE</DefineConstants>
                 <ErrorReport>prompt</ErrorReport>
                 <WarningLevel>4</WarningLevel>
                 <Prefer32Bit>true</Prefer32Bit>
                 <DebugSymbols>true</DebugSymbols>
               </PropertyGroup>
               <ItemGroup>
                 <Reference Include="System" />
                 <Reference Include="System.Core" />
                 <Reference Include="System.Xml.Linq" />
                 <Reference Include="System.Data.DataSetExtensions" />
                 <Reference Include="Microsoft.CSharp" />
                 <Reference Include="System.Data" />
                 <Reference Include="System.Net.Http" />
                 <Reference Include="System.Xml" />
               </ItemGroup>
               <ItemGroup>
                 <Compile Include="Program.cs" />
                 <Compile Include="Properties\AssemblyInfo.cs" />
               </ItemGroup>
               <ItemGroup>
                 <None Include="App.config" />
               </ItemGroup>
               <ItemGroup>
                 <ProjectReference Include="..\Implab\Implab.csproj">
                   <Project>{f550f1f8-8746-4ad0-9614-855f4c4b7f05}</Project>
                   <Name>Implab</Name>
                 </ProjectReference>
               </ItemGroup>
               <Import Project="$(MSBuildToolsPath)\Microsoft.CSharp.targets" />
               <!-- To modify your build process, add your task inside one of the targets below and uncomment it.
                    Other similar extension points exist, see Microsoft.Common.targets.
               <Target Name="BeforeBuild">
               </Target>
               <Target Name="AfterBuild">
               </Target>
               -->
             </Project>

Implab.Playground/Program.cs +157 -16

@@ -1,42 +1,183
1	using Implab.Formats.Json;	1	using Implab.Formats.Json;
		2	using Implab.Parallels;
2	using Implab.Xml;	3	using Implab.Xml;
3	using System;	4	using System;
		5	using System.Collections.Concurrent;
4	using System.Collections.Generic;	6	using System.Collections.Generic;
5	using System.IO;	7	using System.IO;
6	using System.Linq;	8	using System.Linq;
7	using System.Text;	9	using System.Text;
		10	using System.Threading;
8	using System.Threading.Tasks;	11	using System.Threading.Tasks;
9	using System.Xml;	12	using System.Xml;
10	using System.Xml.Serialization;	13	using System.Xml.Serialization;
11		14
12	namespace Implab.Playground {	15	namespace Implab.Playground {
13	public class Program {	16	public class Program {
14		17
15	[XmlRoot(Namespace = "XmlSimpleData")]	18	static void EnqueueRange<T>(ConcurrentQueue<T> q, T[] data, int offset, int len) {
16	public class XmlSimpleModel {	19	for (var i = offset; i < offset + len; i++)
17	[XmlElement]	20	q.Enqueue(data[i]);
18	public string Name { get; set; }	21	}
		22
		23	static bool TryDequeueRange<T>(ConcurrentQueue<T> q,T[] buffer,int offset, int len, out int actual) {
		24	actual = 0;
		25	T res;
		26	while(q.TryDequeue(out res)) {
		27	buffer[offset + actual] = res;
		28	actual++;
		29	if (actual == len)
		30	break;
		31	}
		32	return actual != 0;
		33	}
		34
		35	static void EnqueueRange<T>(SimpleAsyncQueue<T> q, T[] data, int offset, int len) {
		36	for (var i = offset; i < offset + len; i++)
		37	q.Enqueue(data[i]);
		38	}
19		39
20	[XmlElement]	40	static bool TryDequeueRange<T>(SimpleAsyncQueue<T> q, T[] buffer, int offset, int len, out int actual) {
21	public int Order { get; set; }	41	actual = 0;
		42	T res;
		43	while (q.TryDequeue(out res)) {
		44	buffer[offset + actual] = res;
		45	actual++;
		46	if (actual == len)
		47	break;
		48	}
		49	return actual != 0;
		50	}
22		51
23	[XmlElement]	52	/*
24	public string[] Items { get; set; }	53	static void EnqueueRange<T>(AsyncQueue<T> q, T[] data, int offset, int len) {
		54	for (var i = offset; i < offset + len; i++)
		55	q.Enqueue(data[i]);
		56	}
25		57
		58	static bool TryDequeueRange<T>(AsyncQueue<T> q, T[] buffer, int offset, int len, out int actual) {
		59	actual = 0;
		60	T res;
		61	while (q.TryDequeue(out res)) {
		62	buffer[offset + actual] = res;
		63	actual++;
		64	if (actual == len)
		65	break;
		66	}
		67	return actual != 0;
		68	}
		69	*/
		70
		71	static void EnqueueRange<T>(AsyncQueue<T> q, T[] data, int offset, int len) {
		72	q.EnqueueRange(data, offset, len);
26	}	73	}
27		74
		75	static bool TryDequeueRange<T>(AsyncQueue<T> q, T[] buffer, int offset, int len, out int actual) {
		76	return q.TryDequeueRange(buffer, offset, len, out actual);
		77	}
		78
		79
28	static void Main(string[] args) {	80	static void Main(string[] args) {
29	var model = new XmlSimpleModel {
30	Name = "Tablet",
31	Order = 10,
32	Items = new string[] { "z1", "z2", "z3" }
33	};
34		81
35	var doc = SerializationHelpers.SerializeAsXmlDocument(model);	82	//var queue = new ConcurrentQueue<int>();
36		83	var queue = new AsyncQueue<int>();
37	var m2 = SerializationHelpers.DeserializeFromXmlNode<XmlSimpleModel>(doc.DocumentElement);	84	//var queue = new SimpleAsyncQueue<int>();
		85
		86	const int wBatch = 32;
		87	const long wCount = 1000000;
		88	const long total = wBatch * wCount * 3;
		89
		90	long r1 = 0, r2 = 0, r3 = 0;
		91	const int rBatch = 1000;
		92	long read = 0;
		93
		94	var t1 = Environment.TickCount;
		95
		96	AsyncPool.RunThread(
		97	() => {
		98	var buffer = new int[wBatch];
		99	for (int i = 0; i < wBatch; i++)
		100	buffer[i] = 1;
		101
		102	for (int i = 0; i < wCount; i++)
		103	EnqueueRange(queue, buffer, 0, wBatch);
		104	Console.WriteLine("done writer #1: {0} ms", Environment.TickCount - t1);
		105	},
		106	() => {
		107	var buffer = new int[wBatch];
		108	for (int i = 0; i < wBatch; i++)
		109	buffer[i] = 1;
		110
		111	for (int i = 0; i < wCount; i++)
		112	EnqueueRange(queue, buffer, 0, wBatch);
		113	Console.WriteLine("done writer #2: {0} ms", Environment.TickCount - t1);
		114	},
		115	() => {
		116	var buffer = new int[wBatch];
		117	for (int i = 0; i < wBatch; i++)
		118	buffer[i] = 1;
		119
		120	for (int i = 0; i < wCount; i++)
		121	EnqueueRange(queue, buffer, 0, wBatch);
		122	Console.WriteLine("done writer #3: {0} ms", Environment.TickCount - t1);
		123	},
		124	() => {
		125	var buffer = new int[rBatch];
		126
		127	while (read < total) {
		128	int actual;
		129	if (TryDequeueRange(queue, buffer, 0, rBatch, out actual)) {
		130	for (int i = 0; i < actual; i++)
		131	r1 += buffer[i];
		132	Interlocked.Add(ref read, actual);
		133	}
		134	}
		135
		136	Console.WriteLine("done reader #1: {0} ms", Environment.TickCount - t1);
		137	}/*,
		138	() => {
		139	var buffer = new int[rBatch];
		140
		141	while (read < total) {
		142	int actual;
		143	if (TryDequeueRange(queue, buffer, 0, rBatch, out actual)) {
		144	for (int i = 0; i < actual; i++)
		145	r2 += buffer[i];
		146	Interlocked.Add(ref read, actual);
		147	}
		148	}
		149
		150	Console.WriteLine("done reader #2: {0} ms", Environment.TickCount - t1);
		151	}//,
		152	() => {
		153	var buffer = new int[rBatch];
		154
		155	while (read < total) {
		156	int actual;
		157	if (TryDequeueRange(queue, buffer, 0, rBatch, out actual)) {
		158	for (int i = 0; i < actual; i++)
		159	r3 += buffer[i];
		160	Interlocked.Add(ref read, actual);
		161	}
		162	}
		163
		164	Console.WriteLine("done reader #3: {0} ms", Environment.TickCount - t1);
		165	}*/
		166	)
		167	.PromiseAll()
		168	.Join();
		169
		170
		171	Console.WriteLine(
		172	"done: {0} ms, summ#1: {1}, summ#2: {2}, total: {3}, count: {4}",
		173	Environment.TickCount - t1,
		174	r1,
		175	r2,
		176	r1 + r2 + r3,
		177	total
		178	);
38		179
39	Console.WriteLine("done");	180	Console.WriteLine("done");
40	}	181	}
41	}	182	}
42	}	183	}

Implab.Playground2.psess +2 -7

             <?xml version="1.0" encoding="UTF-8"?>
             <VSPerformanceSession Version="1.00">
               <Options>
                 <Solution>Implab.sln</Solution>
                 <CollectionMethod>Sampling</CollectionMethod>
                 <AllocationMethod>None</AllocationMethod>
                 <AddReport>true</AddReport>
                 <ResourceBasedAnalysisSelected>true</ResourceBasedAnalysisSelected>
                 <UniqueReport>Timestamp</UniqueReport>
                 <SamplingMethod>Cycles</SamplingMethod>
                 <CycleCount>50000</CycleCount>
                 <PageFaultCount>10</PageFaultCount>
                 <SysCallCount>10</SysCallCount>
                 <SamplingCounter Name="" ReloadValue="00000000000f4240" DisplayName="" />
                 <RelocateBinaries>false</RelocateBinaries>
                 <HardwareCounters EnableHWCounters="false" />
                 <EtwSettings />
                 <PdhSettings>
                   <PdhCountersEnabled>false</PdhCountersEnabled>
                   <PdhCountersRate>500</PdhCountersRate>
                   <PdhCounters>
                     <PdhCounter>\Память\Обмен страниц/с</PdhCounter>
                     <PdhCounter>\Процессор(_Total)\% загруженности процессора</PdhCounter>
                     <PdhCounter>\Физический диск(_Total)\Средняя длина очереди диска</PdhCounter>
                   </PdhCounters>
                 </PdhSettings>
               </Options>
               <ExcludeSmallFuncs>true</ExcludeSmallFuncs>
               <InteractionProfilingEnabled>false</InteractionProfilingEnabled>
               <JScriptProfilingEnabled>false</JScriptProfilingEnabled>
               <PreinstrumentEvent>
                 <InstrEventExclude>false</InstrEventExclude>
               </PreinstrumentEvent>
               <PostinstrumentEvent>
                 <InstrEventExclude>false</InstrEventExclude>
               </PostinstrumentEvent>
               <Binaries>
                 <ProjBinary>
                   <Path>Implab.Playground\obj\Debug\Implab.Playground.exe</Path>
                   <ArgumentTimestamp>01/01/0001 00:00:00</ArgumentTimestamp>
                   <Instrument>true</Instrument>
                   <Sample>true</Sample>
                   <ExternalWebsite>false</ExternalWebsite>
                   <InteractionProfilingEnabled>false</InteractionProfilingEnabled>
                   <IsLocalJavascript>false</IsLocalJavascript>
                   <IsWindowsStoreApp>false</IsWindowsStoreApp>
                   <IsWWA>false</IsWWA>
                   <LaunchProject>true</LaunchProject>
                   <OverrideProjectSettings>false</OverrideProjectSettings>
                   <LaunchMethod>Executable</LaunchMethod>
-                  <ExecutablePath>Implab.Playground\bin\Release\Implab.Playground.exe</ExecutablePath>
+                  <ExecutablePath>Implab.Playground\bin\Debug\Implab.Playground.exe</ExecutablePath>
-                  <StartupDirectory>Implab.Playground\bin\Release\</StartupDirectory>
+                  <StartupDirectory>Implab.Playground\bin\Debug\</StartupDirectory>
                   <Arguments>
                   </Arguments>
                   <NetAppHost>IIS</NetAppHost>
                   <NetBrowser>InternetExplorer</NetBrowser>
                   <ExcludeSmallFuncs>true</ExcludeSmallFuncs>
                   <JScriptProfilingEnabled>false</JScriptProfilingEnabled>
                   <PreinstrumentEvent>
                     <InstrEventExclude>false</InstrEventExclude>
                   </PreinstrumentEvent>
                   <PostinstrumentEvent>
                     <InstrEventExclude>false</InstrEventExclude>
                   </PostinstrumentEvent>
                   <ProjRef>{100DFEB0-75BE-436F-ADDF-1F46EF433F46}|Implab.Playground\Implab.Playground.csproj</ProjRef>
                   <ProjPath>Implab.Playground\Implab.Playground.csproj</ProjPath>
                   <ProjName>Implab.Playground</ProjName>
                 </ProjBinary>
               </Binaries>
-              <Launches>
-                <ProjBinary>
-                  <Path>:PB:{100DFEB0-75BE-436F-ADDF-1F46EF433F46}|Implab.Playground\Implab.Playground.csproj</Path>
-                </ProjBinary>
-              </Launches>
             </VSPerformanceSession>

Implab.Test/AsyncTests.cs +34 -19

             using System;
             using System.Reflection;
             using System.Threading;
             using Implab.Parallels;
             #if MONO
             using NUnit.Framework;
             using TestClassAttribute = NUnit.Framework.TestFixtureAttribute;
             using TestMethodAttribute = NUnit.Framework.TestAttribute;
             #else
             using Microsoft.VisualStudio.TestTools.UnitTesting;
             #endif
             namespace Implab.Test {
                 [TestClass]
                 public class AsyncTests {
                     [TestMethod]
                     public void ResolveTest() {
                         int res = -1;
                         var p = new Promise<int>();
                         p.Then(x => res = x);
                         p.Resolve(100);
                         Assert.AreEqual(100, res);
                     }
                     [TestMethod]
                     public void RejectTest() {
                         int res = -1;
                         Exception err = null;
                         var p = new Promise<int>();
                         p.Then(
                             x => res = x,
                             e => {
                                 err = e;
                                 return -2;
                             }
                         );
                         p.Reject(new ApplicationException("error"));
                         Assert.AreEqual(res, -1);
                         Assert.AreEqual(err.Message, "error");
                     }
                     [TestMethod]
                     public void CancelExceptionTest() {
                         var p = new Promise<bool>();
                         p.CancelOperation(null);
                         var p2 = p.Then(x => x, null, reason => {
                             throw new ApplicationException("CANCELLED");
                         });
                         try {
                             p2.Join();
                             Assert.Fail();
                         } catch (ApplicationException err) {
                             Assert.AreEqual("CANCELLED", err.InnerException.Message);
                         }
                     }
                     [TestMethod]
                     public void ContinueOnCancelTest() {
                         var p = new Promise<bool>();
                         p.CancelOperation(null);
                         var p2 = p
                             .Then(x => x, null, reason => {
                                 throw new ApplicationException("CANCELLED");
                             })
                             .Then(x => x, e => true);
                         Assert.AreEqual(true, p2.Join());
                     }
                     [TestMethod]
                     public void JoinSuccessTest() {
                         var p = new Promise<int>();
                         p.Resolve(100);
                         Assert.AreEqual(p.Join(), 100);
                     }
                     [TestMethod]
                     public void JoinFailTest() {
                         var p = new Promise<int>();
                         p.Reject(new ApplicationException("failed"));
                         try {
                             p.Join();
                             throw new ApplicationException("WRONG!");
                         } catch (TargetInvocationException err) {
                             Assert.AreEqual(err.InnerException.Message, "failed");
                         } catch {
                             Assert.Fail("Got wrong excaption");
                         }
                     }
                     [TestMethod]
                     public void MapTest() {
                         var p = new Promise<int>();
                         var p2 = p.Then(x => x.ToString());
                         p.Resolve(100);
                         Assert.AreEqual(p2.Join(), "100");
                     }
                     [TestMethod]
                     public void FixErrorTest() {
                         var p = new Promise<int>();
                         var p2 = p.Then(x => x, e => 101);
                         p.Reject(new Exception());
                         Assert.AreEqual(p2.Join(), 101);
                     }
                     [TestMethod]
                     public void ChainTest() {
                         var p1 = new Promise<int>();
                         var p3 = p1.Chain(x => {
                             var p2 = new Promise<string>();
                             p2.Resolve(x.ToString());
                             return p2;
                         });
                         p1.Resolve(100);
                         Assert.AreEqual(p3.Join(), "100");
                     }
                     [TestMethod]
                     public void ChainFailTest() {
                         var p1 = new Promise<int>();
                         var p3 = p1.Chain(x => {
                             var p2 = new Promise<string>();
                             p2.Reject(new Exception("DIE!!!"));
                             return p2;
                         });
                         p1.Resolve(100);
                         Assert.IsTrue(p3.IsResolved);
                     }
                     [TestMethod]
                     public void PoolTest() {
                         var pid = Thread.CurrentThread.ManagedThreadId;
                         var p = AsyncPool.Invoke(() => Thread.CurrentThread.ManagedThreadId);
                         Assert.AreNotEqual(pid, p.Join());
                     }
                     [TestMethod]
                     public void WorkerPoolSizeTest() {
                         var pool = new WorkerPool(5, 10, 1);
                         Assert.AreEqual(5, pool.PoolSize);
                         pool.Invoke(() => { Thread.Sleep(100000000); return 10; });
                         pool.Invoke(() => { Thread.Sleep(100000000); return 10; });
                         pool.Invoke(() => { Thread.Sleep(100000000); return 10; });
                         Assert.AreEqual(5, pool.PoolSize);
                         for (int i = 0; i < 100; i++)
                             pool.Invoke(() => { Thread.Sleep(100000000); return 10; });
                         Thread.Sleep(200);
                         Assert.AreEqual(10, pool.PoolSize);
                         pool.Dispose();
                     }
                     [TestMethod]
                     public void WorkerPoolCorrectTest() {
                         var pool = new WorkerPool(0,1000,100);
                         const int iterations = 1000;
                         int pending = iterations;
                         var stop = new ManualResetEvent(false);
                         var count = 0;
                         for (int i = 0; i < iterations; i++) {
                             pool
                                 .Invoke(() => 1)
                                 .Then(x => Interlocked.Add(ref count, x))
                                 .Then(x => Math.Log10(x))
                                 .On(() => {
                                     Interlocked.Decrement(ref pending);
                                     if (pending == 0)
                                         stop.Set();
                                 }, PromiseEventType.All);
                         }
                         stop.WaitOne();
                         Assert.AreEqual(iterations, count);
                         Console.WriteLine("Max threads: {0}", pool.MaxRunningThreads);
                         pool.Dispose();
                     }
                     [TestMethod]
                     public void WorkerPoolDisposeTest() {
                         var pool = new WorkerPool(5, 20);
                         Assert.AreEqual(5, pool.PoolSize);
                         pool.Dispose();
                         Thread.Sleep(500);
                         Assert.AreEqual(0, pool.PoolSize);
                         pool.Dispose();
                     }
                     [TestMethod]
                     public void MTQueueTest() {
-                        var queue = new MTQueue<int>();
+                        var queue = new SimpleAsyncQueue<int>();
                         int res;
                         queue.Enqueue(10);
                         Assert.IsTrue(queue.TryDequeue(out res));
                         Assert.AreEqual(10, res);
                         Assert.IsFalse(queue.TryDequeue(out res));
                         for (int i = 0; i < 1000; i++)
                             queue.Enqueue(i);
                         for (int i = 0; i < 1000; i++) {
                             queue.TryDequeue(out res);
                             Assert.AreEqual(i, res);
                         }
                         int writers = 0;
                         int readers = 0;
                         var stop = new ManualResetEvent(false);
                         int total = 0;
+                        var ticks = Environment.TickCount;
-                        const int itemsPerWriter = 10000;
+                        const int itemsPerWriter = 1000000;
                         const int writersCount = 10;
                         for (int i = 0; i < writersCount; i++) {
                             Interlocked.Increment(ref writers);
                             AsyncPool
                                 .RunThread(() => {
                                     for (int ii = 0; ii < itemsPerWriter; ii++) {
                                         queue.Enqueue(1);
                                     }
                                     return 1;
                                 })
                                 .On(() => Interlocked.Decrement(ref writers), PromiseEventType.All);
                         }
                         for (int i = 0; i < 10; i++) {
                             Interlocked.Increment(ref readers);
                             AsyncPool
                                 .RunThread(() => {
                                     int t;
                                     do {
                                         while (queue.TryDequeue(out t))
                                             Interlocked.Add(ref total, t);
                                     } while (writers > 0);
                                     return 1;
                                 })
                                 .On(() => {
                                     Interlocked.Decrement(ref readers);
                                     if (readers == 0)
                                         stop.Set();
                                 }, PromiseEventType.All);
                         }
                         stop.WaitOne();
-                        Assert.AreEqual(100000, total);
+                        Console.WriteLine("{0} in {1}ms", total, Environment.TickCount - ticks);
+                        Assert.AreEqual(itemsPerWriter * writersCount, total);
                     }
                     [TestMethod]
                     public void AsyncQueueTest() {
                         var queue = new AsyncQueue<int>();
                         int res;
                         queue.Enqueue(10);
                         Assert.IsTrue(queue.TryDequeue(out res));
                         Assert.AreEqual(10, res);
                         Assert.IsFalse(queue.TryDequeue(out res));
                         for (int i = 0; i < 1000; i++)
                             queue.Enqueue(i);
                         for (int i = 0; i < 1000; i++) {
                             queue.TryDequeue(out res);
                             Assert.AreEqual(i, res);
                         }
                         const int count = 10000000;
                         int res1 = 0, res2 = 0;
                         var t1 = Environment.TickCount;
                         AsyncPool.RunThread(
                             () => {
                                 for (var i = 0; i < count; i++)
                                     queue.Enqueue(1);
                                 Console.WriteLine("done writer #1: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 for (var i = 0; i < count; i++)
                                     queue.Enqueue(2);
                                 Console.WriteLine("done writer #2: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 int temp;
                                 int i = 0;
                                 while (i < count)
                                     if (queue.TryDequeue(out temp)) {
                                         i++;
                                         res1 += temp;
                                     }
                                 Console.WriteLine("done reader #1: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 int temp;
                                 int i = 0;
                                 while (i < count)
                                     if (queue.TryDequeue(out temp)) {
                                         i++;
                                         res2 += temp;
                                     }
                                 Console.WriteLine("done reader #2: {0} ms", Environment.TickCount - t1);
                             }
                         )
                             .PromiseAll()
                             .Join();
                         Assert.AreEqual(count * 3, res1 + res2);
                         Console.WriteLine(
                             "done: {0} ms, summ#1: {1}, summ#2: {2}, total: {3}, count: {4}",
                             Environment.TickCount - t1,
                             res1,
                             res2,
                             res1 + res2,
                             count
                         );
                     }
                     [TestMethod]
                     public void AsyncQueueBatchTest() {
                         var queue = new AsyncQueue<int>();
                         const int wBatch = 29;
                         const int wCount = 400000;
                         const int total = wBatch * wCount * 2;
                         const int summ = wBatch * wCount * 3;
                         int r1 = 0, r2 = 0;
                         const int rBatch = 111;
                         int read = 0;
                         var t1 = Environment.TickCount;
                         AsyncPool.RunThread(
                             () => {
                                 var buffer = new int[wBatch];
                                 for(int i = 0; i<wBatch; i++)
                                     buffer[i] = 1;
                                 for(int i =0; i < wCount; i++)
                                     queue.EnqueueRange(buffer,0,wBatch);
                                 Console.WriteLine("done writer #1: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 var buffer = new int[wBatch];
                                 for(int i = 0; i<wBatch; i++)
                                     buffer[i] = 2;
                                 for(int i =0; i < wCount; i++)
                                     queue.EnqueueRange(buffer,0,wBatch);
                                 Console.WriteLine("done writer #2: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 var buffer = new int[rBatch];
                                 while(read < total) {
                                     int actual;
                                     if (queue.TryDequeueRange(buffer,0,rBatch,out actual)) {
                                         for(int i=0; i< actual; i++)
                                             r1 += buffer[i];
                                         Interlocked.Add(ref read, actual);
                                     }
                                 }
                                 Console.WriteLine("done reader #1: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 var buffer = new int[rBatch];
                                 while(read < total) {
                                     int actual;
                                     if (queue.TryDequeueRange(buffer,0,rBatch,out actual)) {
                                         for(int i=0; i< actual; i++)
                                             r2 += buffer[i];
                                         Interlocked.Add(ref read, actual);
                                     }
                                 }
                                 Console.WriteLine("done reader #2: {0} ms", Environment.TickCount - t1);
                             }
                         )
                             .PromiseAll()
                             .Join();
                         Assert.AreEqual(summ , r1 + r2);
                         Console.WriteLine(
                             "done: {0} ms, summ#1: {1}, summ#2: {2}, total: {3}, count: {4}",
                             Environment.TickCount - t1,
                             r1,
                             r2,
                             r1 + r2,
                             total
                         );
                     }
                     [TestMethod]
                     public void AsyncQueueChunkDequeueTest() {
                         var queue = new AsyncQueue<int>();
                         const int wBatch = 31;
                         const int wCount = 200000;
                         const int total = wBatch * wCount * 3;
                         const int summ = wBatch * wCount * 6;
                         int r1 = 0, r2 = 0;
                         const int rBatch = 1024;
                         int read = 0;
                         var t1 = Environment.TickCount;
                         AsyncPool.RunThread(
                             () => {
                                 var buffer = new int[wBatch];
                                 for(int i = 0; i<wBatch; i++)
                                     buffer[i] = 1;
                                 for(int i =0; i < wCount; i++)
                                     queue.EnqueueRange(buffer,0,wBatch);
                                 Console.WriteLine("done writer #1: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 var buffer = new int[wBatch];
                                 for(int i = 0; i<wBatch; i++)
                                     buffer[i] = 2;
                                 for(int i =0; i < wCount; i++)
                                     queue.EnqueueRange(buffer,0,wBatch);
                                 Console.WriteLine("done writer #2: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 var buffer = new int[wBatch];
                                 for(int i = 0; i<wBatch; i++)
                                     buffer[i] = 3;
                                 for(int i =0; i < wCount; i++)
                                     queue.EnqueueRange(buffer,0,wBatch);
                                 Console.WriteLine("done writer #3: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 var buffer = new int[rBatch];
                                 int count = 1;
                                 double avgchunk = 0;
                                 while(read < total) {
                                     int actual;
                                     if (queue.TryDequeueChunk(buffer,0,rBatch,out actual)) {
                                         for(int i=0; i< actual; i++)
                                             r2 += buffer[i];
                                         Interlocked.Add(ref read, actual);
                                         avgchunk = avgchunk*(count-1)/count + actual/(double)count;
                                         count ++;
                                     }
                                 }
                                 Console.WriteLine("done reader #2: {0} ms, avg chunk size: {1}", Environment.TickCount - t1, avgchunk);
                             }
                         )
                             .PromiseAll()
                             .Join();
                         Assert.AreEqual(summ , r1 + r2);
                         Console.WriteLine(
                             "done: {0} ms, summ#1: {1}, summ#2: {2}, total: {3}, count: {4}",
                             Environment.TickCount - t1,
                             r1,
                             r2,
                             r1 + r2,
                             total
                         );
                     }
                     [TestMethod]
                     public void AsyncQueueDrainTest() {
                         var queue = new AsyncQueue<int>();
-                        const int wBatch = 11;
+                        const int wBatch = 32;
                         const int wCount = 200000;
                         const int total = wBatch * wCount * 3;
                         const int summ = wBatch * wCount * 3;
                         int r1 = 0, r2 = 0;
-                        const int rBatch = 11;
                         int read = 0;
                         var t1 = Environment.TickCount;
                         AsyncPool.RunThread(
                             () => {
                                 var buffer = new int[wBatch];
                                 for(int i = 0; i<wBatch; i++)
                                     buffer[i] = 1;
                                 for(int i =0; i < wCount; i++)
                                     queue.EnqueueRange(buffer,0,wBatch);
                                 Console.WriteLine("done writer #1: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
-                                for(int i =0; i < wCount * wBatch; i++)
+                                var buffer = new int[wBatch];
-                                    queue.Enqueue(1);
+                                for (int i = 0; i < wBatch; i++)
+                                    buffer[i] = 1;
+                                for (int i = 0; i < wCount; i++)
+                                    queue.EnqueueRange(buffer, 0, wBatch);
                                 Console.WriteLine("done writer #2: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 var buffer = new int[wBatch];
                                 for(int i = 0; i<wBatch; i++)
                                     buffer[i] = 1;
                                 for(int i =0; i < wCount; i++)
                                     queue.EnqueueRange(buffer,0,wBatch);
                                 Console.WriteLine("done writer #3: {0} ms", Environment.TickCount - t1);
                             },
                             /*() => {
                                 int temp;
                                 int count = 0;
                                 while (read < total)
                                     if (queue.TryDequeue(out temp)) {
                                         count++;
                                         r1 += temp;
                                         Interlocked.Increment(ref read);
                                     }
                                 Console.WriteLine("done reader #1: {0} ms, {1} count", Environment.TickCount - t1, count);
                             },*/
                             /*() => {
                                 var buffer = new int[rBatch];
                                 var count = 0;
                                 while(read < total) {
                                     int actual;
                                     if (queue.TryDequeueRange(buffer,0,rBatch,out actual)) {
                                         for(int i=0; i< actual; i++)
                                             r1 += buffer[i];
                                         Interlocked.Add(ref read, actual);
                                         count += actual;
                                     }
                                 }
                                 Console.WriteLine("done reader #1: {0} ms, {1} items", Environment.TickCount - t1, count);
                             },*/
                             () => {
                                 var count = 0;
-                                while(read < total) {
+                                int emptyDrains = 0;
+                                while (read < total) {
                                     var buffer = queue.Drain();
-                                    for(int i=0; i< buffer.Length; i++)
+                                    if (buffer.Count == 0)
+                                        emptyDrains++;
+                                    for(int i=0; i< buffer.Count; i++)
                                         r1 += buffer[i];
-                                        Interlocked.Add(ref read, buffer.Length);
+                                        Interlocked.Add(ref read, buffer.Count);
-                                    count += buffer.Length;
+                                    count += buffer.Count;
                                 }
-                                Console.WriteLine("done reader #1: {0} ms, {1} items", Environment.TickCount - t1, count);
+                                Console.WriteLine("done reader #1: {0} ms, {1} items, empty: {2}", Environment.TickCount - t1, count, emptyDrains);
                             },
                             () => {
                                 var count = 0;
-                                while(read < total) {
+                                int emptyDrains = 0;
+                                while (read < total) {
                                     var buffer = queue.Drain();
-                                    for(int i=0; i< buffer.Length; i++)
+                                    if (buffer.Count == 0)
+                                        emptyDrains++;
+                                    for (int i=0; i< buffer.Count; i++)
                                         r2 += buffer[i];
-                                    Interlocked.Add(ref read, buffer.Length);
+                                    Interlocked.Add(ref read, buffer.Count);
-                                    count += buffer.Length;
+                                    count += buffer.Count;
                                 }
-                                Console.WriteLine("done reader #2: {0} ms, {1} items", Environment.TickCount - t1, count);
+                                Console.WriteLine("done reader #2: {0} ms, {1} items, empty: {2}", Environment.TickCount - t1, count, emptyDrains);
                             }
                         )
                             .PromiseAll()
                             .Join();
                         Assert.AreEqual(summ , r1 + r2);
                         Console.WriteLine(
                             "done: {0} ms, summ#1: {1}, summ#2: {2}, total: {3}, count: {4}",
                             Environment.TickCount - t1,
                             r1,
                             r2,
                             r1 + r2,
                             total
                         );
                     }
                     [TestMethod]
                     public void ParallelMapTest() {
                         const int count = 100000;
                         var args = new double[count];
                         var rand = new Random();
                         for (int i = 0; i < count; i++)
                             args[i] = rand.NextDouble();
                         var t = Environment.TickCount;
                         var res = args.ParallelMap(x => Math.Sin(x*x), 4).Join();
                         Console.WriteLine("Map complete in {0} ms", Environment.TickCount - t);
                         t = Environment.TickCount;
                         for (int i = 0; i < count; i++)
                             Assert.AreEqual(Math.Sin(args[i] * args[i]), res[i]);
                         Console.WriteLine("Verified in {0} ms", Environment.TickCount - t);
                     }
                     [TestMethod]
                     public void ChainedMapTest() {
                         using (var pool = new WorkerPool()) {
                             const int count = 10000;
                             var args = new double[count];
                             var rand = new Random();
                             for (int i = 0; i < count; i++)
                                 args[i] = rand.NextDouble();
                             var t = Environment.TickCount;
                             var res = args
                                 .ChainedMap(
                                     // Analysis disable once AccessToDisposedClosure
                                     x => pool.Invoke(
                                         () => Math.Sin(x * x)
                                     ),
                                 )
                                 .Join();
                             Console.WriteLine("Map complete in {0} ms", Environment.TickCount - t);
                             t = Environment.TickCount;
                             for (int i = 0; i < count; i++)
                                 Assert.AreEqual(Math.Sin(args[i] * args[i]), res[i]);
                             Console.WriteLine("Verified in {0} ms", Environment.TickCount - t);
                             Console.WriteLine("Max workers: {0}", pool.MaxRunningThreads);
                         }
                     }
                     [TestMethod]
                     public void ParallelForEachTest() {
                         const int count = 100000;
                         var args = new int[count];
                         var rand = new Random();
                         for (int i = 0; i < count; i++)
                             args[i] = (int)(rand.NextDouble() * 100);
                         int result = 0;
                         var t = Environment.TickCount;
                         args.ParallelForEach(x => Interlocked.Add(ref result, x), 4).Join();
                         Console.WriteLine("Iteration complete in {0} ms, result: {1}", Environment.TickCount - t, result);
                         int result2 = 0;
                         t = Environment.TickCount;
                         for (int i = 0; i < count; i++)
                             result2 += args[i];
                         Assert.AreEqual(result2, result);
                         Console.WriteLine("Verified in {0} ms", Environment.TickCount - t);
                     }
                     [TestMethod]
                     public void ComplexCase1Test() {
                         var flags = new bool[3];
                         // op1 (aync 200ms) => op2 (async 200ms) => op3 (sync map)
                         var step1 = PromiseHelper
                             .Sleep(200, "Alan")
                             .On(() => flags[0] = true, PromiseEventType.Cancelled);
                         var p = step1
                             .Chain(x =>
                                 PromiseHelper
                                     .Sleep(200, "Hi, " + x)
                                     .Then(y => y)
                                     .On(() => flags[1] = true, PromiseEventType.Cancelled)
                             )
                             .On(() => flags[2] = true, PromiseEventType.Cancelled);
                         step1.Join();
                         p.Cancel();
                         try {
                             Assert.AreEqual(p.Join(), "Hi, Alan");
                             Assert.Fail("Shouldn't get here");
                         } catch (OperationCanceledException) {
                         }
                         Assert.IsFalse(flags[0]);
                         Assert.IsTrue(flags[1]);
                         Assert.IsTrue(flags[2]);
                     }
                     [TestMethod]
                     public void ChainedCancel1Test() {
                         // при отмене сцепленной асинхронной операции все обещание должно
                         // завершаться ошибкой OperationCanceledException
                         var p = PromiseHelper
                             .Sleep(1, "Hi, HAL!")
                             .Then(x => {
                                 // запускаем две асинхронные операции
                                 var result = PromiseHelper.Sleep(1000, "HEM ENABLED!!!");
                                 // вторая операция отменяет первую до завершения
                                 PromiseHelper
                                     .Sleep(100, "HAL, STOP!")
                                     .Then(result.Cancel);
                                 return result;
                             });
                         try {
                             p.Join();
                         } catch (TargetInvocationException err) {
                             Assert.IsTrue(err.InnerException is OperationCanceledException);
                         }
                     }
                     [TestMethod]
                     public void ChainedCancel2Test() {
                         // при отмене цепочки обещаний, вложенные операции также должны отменяться
                         var pSurvive = new Promise<bool>();
                         var hemStarted = new Signal();
                         var p = PromiseHelper
                             .Sleep(1, "Hi, HAL!")
                             .Chain(() => {
                                 hemStarted.Set();
                                 // запускаем две асинхронные операции
                                 var result = PromiseHelper
                                     .Sleep(2000, "HEM ENABLED!!!")
                                     .Then(() => pSurvive.Resolve(false));
                                 result
                                     .On(() => pSurvive.Resolve(true), PromiseEventType.Cancelled);
                                 return result;
                             });
                         hemStarted.Wait();
                         p.Cancel();
                         try {
                             p.Join();
                             Assert.Fail();
                         } catch (OperationCanceledException) {
                         }
                         Assert.IsTrue(pSurvive.Join());
                     }
                     [TestMethod]
                     public void SharedLockTest() {
                         var l = new SharedLock();
                         int shared = 0;
                         int exclusive = 0;
                         var s1 = new Signal();
                         var log = new AsyncQueue<string>();
                         try {
                             AsyncPool.RunThread(
                                 () => {
                                     log.Enqueue("Reader #1 started");
                                     try {
                                         l.LockShared();
                                         log.Enqueue("Reader #1 lock got");
                                         if (Interlocked.Increment(ref shared) == 2)
                                             s1.Set();
                                         s1.Wait();
                                         log.Enqueue("Reader #1 finished");
                                         Interlocked.Decrement(ref shared);
                                     } finally {
                                         l.Release();
                                         log.Enqueue("Reader #1 lock released");
                                     }
                                 },
                                 () => {
                                     log.Enqueue("Reader #2 started");
                                     try {
                                         l.LockShared();
                                         log.Enqueue("Reader #2 lock got");
                                         if (Interlocked.Increment(ref shared) == 2)
                                             s1.Set();
                                         s1.Wait();
                                         log.Enqueue("Reader #2 upgrading to writer");
                                         Interlocked.Decrement(ref shared);
                                         l.Upgrade();
                                         log.Enqueue("Reader #2 upgraded");
                                         Assert.AreEqual(1, Interlocked.Increment(ref exclusive));
                                         Assert.AreEqual(0, shared);
                                         log.Enqueue("Reader #2 finished");
                                         Interlocked.Decrement(ref exclusive);
                                     } finally {
                                         l.Release();
                                         log.Enqueue("Reader #2 lock released");
                                     }
                                 },
                                 () => {
                                     log.Enqueue("Writer #1 started");
                                     try {
                                         l.LockExclusive();
                                         log.Enqueue("Writer #1 got the lock");
                                         Assert.AreEqual(1, Interlocked.Increment(ref exclusive));
                                         Interlocked.Decrement(ref exclusive);
                                         log.Enqueue("Writer #1 is finished");
                                     } finally {
                                         l.Release();
                                         log.Enqueue("Writer #1 lock released");
                                     }
                                 }
                             ).PromiseAll().Join(1000);
                             log.Enqueue("Done");
                         } catch(Exception error) {
                             log.Enqueue(error.Message);
                             throw;
                         } finally {
                             foreach (var m in log)
                                 Console.WriteLine(m);
                         }
                     }
                     #if NET_4_5
                     [TestMethod]
                     public async void TaskInteropTest() {
                         var promise = new Promise<int>();
                         promise.Resolve(10);
                         var res = await promise;
                         Assert.AreEqual(10, res);
                     }
                     #endif
                 }
             }

Implab.sln 0 -10

             Microsoft Visual Studio Solution File, Format Version 12.00
             # Visual Studio 14
             VisualStudioVersion = 14.0.25420.1
             MinimumVisualStudioVersion = 10.0.40219.1
             Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Implab", "Implab\Implab.csproj", "{F550F1F8-8746-4AD0-9614-855F4C4B7F05}"
             EndProject
             Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Solution Items", "Solution Items", "{CE8D8D18-437A-445C-B662-4C2CE79A76F6}"
             	ProjectSection(SolutionItems) = preProject
             		Implab.vsmdi = Implab.vsmdi
             		Local.testsettings = Local.testsettings
             		TraceAndTestImpact.testsettings = TraceAndTestImpact.testsettings
             	EndProjectSection
             EndProject
             Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Implab.Test", "Implab.Test\Implab.Test.csproj", "{63F92C0C-61BF-48C0-A377-8D67C3C661D0}"
             EndProject
-            Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Implab.Fx", "Implab.Fx\Implab.Fx.csproj", "{06E706F8-6881-43EB-927E-FFC503AF6ABC}"
-            EndProject
             Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Implab.Format.Test", "Implab.Format.Test\Implab.Format.Test.csproj", "{4D364996-7ECD-4193-8F90-F223FFEA49DA}"
             EndProject
             Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Implab.Playground", "Implab.Playground\Implab.Playground.csproj", "{100DFEB0-75BE-436F-ADDF-1F46EF433F46}"
             EndProject
             Global
             	GlobalSection(Performance) = preSolution
             		HasPerformanceSessions = true
             	EndGlobalSection
             	GlobalSection(SolutionConfigurationPlatforms) = preSolution
             		Debug 4.5|Any CPU = Debug 4.5|Any CPU
             		Debug|Any CPU = Debug|Any CPU
             		Release 4.5|Any CPU = Release 4.5|Any CPU
             		Release|Any CPU = Release|Any CPU
             	EndGlobalSection
             	GlobalSection(ProjectConfigurationPlatforms) = postSolution
             		{F550F1F8-8746-4AD0-9614-855F4C4B7F05}.Debug 4.5|Any CPU.ActiveCfg = Debug 4.5|Any CPU
             		{F550F1F8-8746-4AD0-9614-855F4C4B7F05}.Debug 4.5|Any CPU.Build.0 = Debug 4.5|Any CPU
             		{F550F1F8-8746-4AD0-9614-855F4C4B7F05}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
             		{F550F1F8-8746-4AD0-9614-855F4C4B7F05}.Debug|Any CPU.Build.0 = Debug|Any CPU
             		{F550F1F8-8746-4AD0-9614-855F4C4B7F05}.Release 4.5|Any CPU.ActiveCfg = Release 4.5|Any CPU
             		{F550F1F8-8746-4AD0-9614-855F4C4B7F05}.Release 4.5|Any CPU.Build.0 = Release 4.5|Any CPU
             		{F550F1F8-8746-4AD0-9614-855F4C4B7F05}.Release|Any CPU.ActiveCfg = Release|Any CPU
             		{F550F1F8-8746-4AD0-9614-855F4C4B7F05}.Release|Any CPU.Build.0 = Release|Any CPU
             		{63F92C0C-61BF-48C0-A377-8D67C3C661D0}.Debug 4.5|Any CPU.ActiveCfg = Debug 4.5|Any CPU
             		{63F92C0C-61BF-48C0-A377-8D67C3C661D0}.Debug 4.5|Any CPU.Build.0 = Debug 4.5|Any CPU
             		{63F92C0C-61BF-48C0-A377-8D67C3C661D0}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
             		{63F92C0C-61BF-48C0-A377-8D67C3C661D0}.Debug|Any CPU.Build.0 = Debug|Any CPU
             		{63F92C0C-61BF-48C0-A377-8D67C3C661D0}.Release 4.5|Any CPU.ActiveCfg = Release 4.5|Any CPU
             		{63F92C0C-61BF-48C0-A377-8D67C3C661D0}.Release 4.5|Any CPU.Build.0 = Release 4.5|Any CPU
             		{63F92C0C-61BF-48C0-A377-8D67C3C661D0}.Release|Any CPU.ActiveCfg = Release|Any CPU
             		{63F92C0C-61BF-48C0-A377-8D67C3C661D0}.Release|Any CPU.Build.0 = Release|Any CPU
-            		{06E706F8-6881-43EB-927E-FFC503AF6ABC}.Debug 4.5|Any CPU.ActiveCfg = Debug 4.5|Any CPU
-            		{06E706F8-6881-43EB-927E-FFC503AF6ABC}.Debug 4.5|Any CPU.Build.0 = Debug 4.5|Any CPU
-            		{06E706F8-6881-43EB-927E-FFC503AF6ABC}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
-            		{06E706F8-6881-43EB-927E-FFC503AF6ABC}.Debug|Any CPU.Build.0 = Debug|Any CPU
-            		{06E706F8-6881-43EB-927E-FFC503AF6ABC}.Release 4.5|Any CPU.ActiveCfg = Release 4.5|Any CPU
-            		{06E706F8-6881-43EB-927E-FFC503AF6ABC}.Release 4.5|Any CPU.Build.0 = Release 4.5|Any CPU
-            		{06E706F8-6881-43EB-927E-FFC503AF6ABC}.Release|Any CPU.ActiveCfg = Release|Any CPU
-            		{06E706F8-6881-43EB-927E-FFC503AF6ABC}.Release|Any CPU.Build.0 = Release|Any CPU
             		{4D364996-7ECD-4193-8F90-F223FFEA49DA}.Debug 4.5|Any CPU.ActiveCfg = Debug|Any CPU
             		{4D364996-7ECD-4193-8F90-F223FFEA49DA}.Debug 4.5|Any CPU.Build.0 = Debug|Any CPU
             		{4D364996-7ECD-4193-8F90-F223FFEA49DA}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
             		{4D364996-7ECD-4193-8F90-F223FFEA49DA}.Debug|Any CPU.Build.0 = Debug|Any CPU
             		{4D364996-7ECD-4193-8F90-F223FFEA49DA}.Release 4.5|Any CPU.ActiveCfg = Release|Any CPU
             		{4D364996-7ECD-4193-8F90-F223FFEA49DA}.Release 4.5|Any CPU.Build.0 = Release|Any CPU
             		{4D364996-7ECD-4193-8F90-F223FFEA49DA}.Release|Any CPU.ActiveCfg = Release|Any CPU
             		{4D364996-7ECD-4193-8F90-F223FFEA49DA}.Release|Any CPU.Build.0 = Release|Any CPU
             		{100DFEB0-75BE-436F-ADDF-1F46EF433F46}.Debug 4.5|Any CPU.ActiveCfg = Debug|Any CPU
             		{100DFEB0-75BE-436F-ADDF-1F46EF433F46}.Debug 4.5|Any CPU.Build.0 = Debug|Any CPU
             		{100DFEB0-75BE-436F-ADDF-1F46EF433F46}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
             		{100DFEB0-75BE-436F-ADDF-1F46EF433F46}.Debug|Any CPU.Build.0 = Debug|Any CPU
             		{100DFEB0-75BE-436F-ADDF-1F46EF433F46}.Release 4.5|Any CPU.ActiveCfg = Release|Any CPU
             		{100DFEB0-75BE-436F-ADDF-1F46EF433F46}.Release 4.5|Any CPU.Build.0 = Release|Any CPU
             		{100DFEB0-75BE-436F-ADDF-1F46EF433F46}.Release|Any CPU.ActiveCfg = Release|Any CPU
             		{100DFEB0-75BE-436F-ADDF-1F46EF433F46}.Release|Any CPU.Build.0 = Release|Any CPU
             	EndGlobalSection
             	GlobalSection(SolutionProperties) = preSolution
             		HideSolutionNode = FALSE
             	EndGlobalSection
             	GlobalSection(MonoDevelopProperties) = preSolution
             		StartupItem = Implab\Implab.csproj
             		Policies = $0
             		$0.CSharpFormattingPolicy = $1
             		$1.IndentSwitchBody = True
             		$1.NamespaceBraceStyle = EndOfLine
             		$1.ClassBraceStyle = EndOfLine
             		$1.InterfaceBraceStyle = EndOfLine
             		$1.StructBraceStyle = EndOfLine
             		$1.EnumBraceStyle = EndOfLine
             		$1.MethodBraceStyle = EndOfLine
             		$1.ConstructorBraceStyle = EndOfLine
             		$1.DestructorBraceStyle = EndOfLine
             		$1.BeforeMethodDeclarationParentheses = False
             		$1.BeforeMethodCallParentheses = False
             		$1.BeforeConstructorDeclarationParentheses = False
             		$1.NewLineBeforeConstructorInitializerColon = NewLine
             		$1.NewLineAfterConstructorInitializerColon = SameLine
             		$1.BeforeIndexerDeclarationBracket = False
             		$1.BeforeDelegateDeclarationParentheses = False
             		$1.NewParentheses = False
             		$1.SpacesBeforeBrackets = False
             		$1.inheritsSet = Mono
             		$1.inheritsScope = text/x-csharp
             		$1.scope = text/x-csharp
             		$0.TextStylePolicy = $6
             		$2.FileWidth = 120
             		$2.EolMarker = Unix
             		$2.inheritsSet = VisualStudio
             		$2.inheritsScope = text/plain
             		$2.scope = text/x-csharp
             		$0.DotNetNamingPolicy = $3
             		$3.DirectoryNamespaceAssociation = PrefixedHierarchical
             		$3.ResourceNamePolicy = MSBuild
             		$4.FileWidth = 120
             		$4.TabsToSpaces = False
             		$4.inheritsSet = VisualStudio
             		$4.inheritsScope = text/plain
             		$4.scope = application/xml
             		$0.XmlFormattingPolicy = $5
             		$5.inheritsSet = Mono
             		$5.inheritsScope = application/xml
             		$5.scope = application/xml
             		$6.FileWidth = 120
             		$6.TabsToSpaces = False
             		$6.inheritsSet = VisualStudio
             		$6.inheritsScope = text/plain
             		$6.scope = text/plain
             		$0.NameConventionPolicy = $7
             		$7.Rules = $8
             		$8.NamingRule = $34
             		$9.Name = Namespaces
             		$9.AffectedEntity = Namespace
             		$9.VisibilityMask = VisibilityMask
             		$9.NamingStyle = PascalCase
             		$9.IncludeInstanceMembers = True
             		$9.IncludeStaticEntities = True
             		$10.Name = Types
             		$10.AffectedEntity = Class, Struct, Enum, Delegate
             		$10.VisibilityMask = VisibilityMask
             		$10.NamingStyle = PascalCase
             		$10.IncludeInstanceMembers = True
             		$10.IncludeStaticEntities = True
             		$11.Name = Interfaces
             		$11.RequiredPrefixes = $12
             		$12.String = I
             		$11.AffectedEntity = Interface
             		$11.VisibilityMask = VisibilityMask
             		$11.NamingStyle = PascalCase
             		$11.IncludeInstanceMembers = True
             		$11.IncludeStaticEntities = True
             		$13.Name = Attributes
             		$13.RequiredSuffixes = $14
             		$14.String = Attribute
             		$13.AffectedEntity = CustomAttributes
             		$13.VisibilityMask = VisibilityMask
             		$13.NamingStyle = PascalCase
             		$13.IncludeInstanceMembers = True
             		$13.IncludeStaticEntities = True
             		$15.Name = Event Arguments
             		$15.RequiredSuffixes = $16
             		$16.String = EventArgs
             		$15.AffectedEntity = CustomEventArgs
             		$15.VisibilityMask = VisibilityMask
             		$15.NamingStyle = PascalCase
             		$15.IncludeInstanceMembers = True
             		$15.IncludeStaticEntities = True
             		$17.Name = Exceptions
             		$17.RequiredSuffixes = $18
             		$18.String = Exception
             		$17.AffectedEntity = CustomExceptions
             		$17.VisibilityMask = VisibilityMask
             		$17.NamingStyle = PascalCase
             		$17.IncludeInstanceMembers = True
             		$17.IncludeStaticEntities = True
             		$19.Name = Methods
             		$19.AffectedEntity = Methods
             		$19.VisibilityMask = VisibilityMask
             		$19.NamingStyle = PascalCase
             		$19.IncludeInstanceMembers = True
             		$19.IncludeStaticEntities = True
             		$20.Name = Static Readonly Fields
             		$20.AffectedEntity = ReadonlyField
             		$20.VisibilityMask = Internal, Protected, Public
             		$20.NamingStyle = CamelCase
             		$20.IncludeInstanceMembers = False
             		$20.IncludeStaticEntities = True
             		$21.Name = Fields (Non Private)
             		$21.AffectedEntity = Field
             		$21.VisibilityMask = Internal, Public
             		$21.NamingStyle = CamelCase
             		$21.IncludeInstanceMembers = True
             		$21.IncludeStaticEntities = True
             		$22.Name = ReadOnly Fields (Non Private)
             		$22.AffectedEntity = ReadonlyField
             		$22.VisibilityMask = Internal, Public
             		$22.NamingStyle = CamelCase
             		$22.IncludeInstanceMembers = True
             		$22.IncludeStaticEntities = False
             		$23.Name = Fields (Private)
             		$23.RequiredPrefixes = $24
             		$24.String = m_
             		$23.AffectedEntity = Field, ReadonlyField
             		$23.VisibilityMask = Private, Protected
             		$23.NamingStyle = CamelCase
             		$23.IncludeInstanceMembers = True
             		$23.IncludeStaticEntities = False
             		$25.Name = Static Fields (Private)
             		$25.RequiredPrefixes = $26
             		$26.String = _
             		$25.AffectedEntity = Field
             		$25.VisibilityMask = Private
             		$25.NamingStyle = CamelCase
             		$25.IncludeInstanceMembers = False
             		$25.IncludeStaticEntities = True
             		$27.Name = ReadOnly Fields (Private)
             		$27.RequiredPrefixes = $28
             		$28.String = m_
             		$27.AffectedEntity = ReadonlyField
             		$27.VisibilityMask = Private, Protected
             		$27.NamingStyle = CamelCase
             		$27.IncludeInstanceMembers = True
             		$27.IncludeStaticEntities = False
             		$29.Name = Constant Fields
             		$29.AffectedEntity = ConstantField
             		$29.VisibilityMask = VisibilityMask
             		$29.NamingStyle = AllUpper
             		$29.IncludeInstanceMembers = True
             		$29.IncludeStaticEntities = True
             		$30.Name = Properties
             		$30.AffectedEntity = Property
             		$30.VisibilityMask = VisibilityMask
             		$30.NamingStyle = PascalCase
             		$30.IncludeInstanceMembers = True
             		$30.IncludeStaticEntities = True
             		$31.Name = Events
             		$31.AffectedEntity = Event
             		$31.VisibilityMask = VisibilityMask
             		$31.NamingStyle = PascalCase
             		$31.IncludeInstanceMembers = True
             		$31.IncludeStaticEntities = True
             		$32.Name = Enum Members
             		$32.AffectedEntity = EnumMember
             		$32.VisibilityMask = VisibilityMask
             		$32.NamingStyle = PascalCase
             		$32.IncludeInstanceMembers = True
             		$32.IncludeStaticEntities = True
             		$33.Name = Parameters
             		$33.AffectedEntity = Parameter, LocalVariable
             		$33.VisibilityMask = VisibilityMask
             		$33.NamingStyle = CamelCase
             		$33.IncludeInstanceMembers = True
             		$33.IncludeStaticEntities = True
             		$34.Name = Type Parameters
             		$34.RequiredPrefixes = $35
             		$35.String = T
             		$34.AffectedEntity = TypeParameter
             		$34.VisibilityMask = VisibilityMask
             		$34.NamingStyle = PascalCase
             		$34.IncludeInstanceMembers = True
             		$34.IncludeStaticEntities = True
             	EndGlobalSection
             	GlobalSection(TestCaseManagementSettings) = postSolution
             		CategoryFile = Implab.vsmdi
             	EndGlobalSection
             EndGlobal

Implab/AbstractEvent.cs +4 -4

             using System;
             using Implab.Parallels;
             using System.Threading;
             using System.Reflection;
             namespace Implab {
                 public abstract class AbstractEvent<THandler> : ICancellationToken, ICancellable {
                     const int UNRESOLVED_SATE = 0;
                     const int TRANSITIONAL_STATE = 1;
                     protected const int SUCCEEDED_STATE = 2;
                     protected const int REJECTED_STATE = 3;
                     protected const int CANCELLED_STATE = 4;
                     const int CANCEL_NOT_REQUESTED = 0;
                     const int CANCEL_REQUESTING = 1;
                     const int CANCEL_REQUESTED = 2;
                     const int RESERVED_HANDLERS_COUNT = 4;
                     int m_state;
                     Exception m_error;
                     int m_handlersCount;
                     //readonly THandler[] m_handlers = new THandler[RESERVED_HANDLERS_COUNT];
                     THandler[] m_handlers;
-                    MTQueue<THandler> m_extraHandlers;
+                    SimpleAsyncQueue<THandler> m_extraHandlers;
                     int m_handlerPointer = -1;
                     int m_handlersCommited;
                     int m_cancelRequest;
                     Exception m_cancelationReason;
-                    MTQueue<Action<Exception>> m_cancelationHandlers;
+                    SimpleAsyncQueue<Action<Exception>> m_cancelationHandlers;
                     #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 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);
                         Signal();
                     }
                     /// <summary>
                     /// Выполняет обещание, сообщая об ошибке
                     /// </summary>
                     /// <remarks>
                     /// Поскольку обещание должно работать в многопточной среде, при его выполнении сразу несколько потоков
                     /// могу вернуть ошибку, при этом только первая будет использована в качестве результата, остальные
                     /// будут проигнорированы.
                     /// </remarks>
                     /// <param name="error">Исключение возникшее при выполнении операции</param>
                     /// <exception cref="InvalidOperationException">Данное обещание уже выполнено</exception>
                     protected void SetError(Exception error) {
                         if (BeginTransit()) {
                             m_error = error;
                             CompleteTransit(REJECTED_STATE);
                             Signal();
                         } 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(Exception reason) {
                         if (BeginTransit()) {
                             m_error = reason;
                             CompleteTransit(CANCELLED_STATE);
                             Signal();
                         }
                     }
                     protected abstract void SignalHandler(THandler handler, int signal);
                     void Signal() {
                         var hp = m_handlerPointer;
                         var slot = hp +1 ;
                         while (slot < m_handlersCommited) {
                             if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) == hp) {
                                 SignalHandler(m_handlers[slot], m_state);
                             }
                             hp = m_handlerPointer;
                             slot = hp +1 ;
                         }
                         if (m_extraHandlers != null) {
                             THandler handler;
                             while (m_extraHandlers.TryDequeue(out handler))
                                 SignalHandler(handler, m_state);
                         }
                     }
                     #endregion
                     protected abstract Signal GetResolveSignal();
                     #region synchronization traits
                     protected void WaitResult(int timeout) {
                         if (!(IsResolved || GetResolveSignal().Wait(timeout)))
                             throw new TimeoutException();
                         switch (m_state) {
                             case SUCCEEDED_STATE:
                                 return;
                             case CANCELLED_STATE:
                                 throw new OperationCanceledException("The operation has been cancelled", m_error);
                             case REJECTED_STATE:
                                 throw new TargetInvocationException(m_error);
                             default:
                                 throw new ApplicationException(String.Format("The promise state {0} is invalid", 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
                             SignalHandler(handler, m_state);
                         } else {
                             var slot = Interlocked.Increment(ref m_handlersCount) - 1;
                             if (slot < RESERVED_HANDLERS_COUNT) {
                                 if (slot == 0) {
                                     m_handlers = new THandler[RESERVED_HANDLERS_COUNT];
                                 } else {
                                     while (m_handlers == null)
                                         Thread.MemoryBarrier();
                                 }
                                 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;
                                             SignalHandler(m_handlers[slot], m_state);
                                         }
                                         break;
                                     } while(true);
                                 }
                             } else {
                                 if (slot == RESERVED_HANDLERS_COUNT) {
-                                    m_extraHandlers = new MTQueue<THandler>();
+                                    m_extraHandlers = new SimpleAsyncQueue<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 :)
                                     SignalHandler(handler, m_state);
                             }
                         }
                     }
                     #endregion
                     #region IPromise implementation
                     public bool IsResolved {
                         get {
                             Thread.MemoryBarrier();
                             return m_state > 1;
                         }
                     }
                     public bool IsCancelled {
                         get {
                             Thread.MemoryBarrier();
                             return m_state == CANCELLED_STATE;
                         }
                     }
                     #endregion
                     public Exception Error {
                         get {
                             return m_error;
                         }
                     }
                     public bool CancelOperationIfRequested() {
                         if (IsCancellationRequested) {
                             CancelOperation(CancellationReason);
                             return true;
                         }
                         return false;
                     }
                     public virtual void CancelOperation(Exception reason) {
                         SetCancelled(reason);
                     }
                     public void CancellationRequested(Action<Exception> handler) {
                         Safe.ArgumentNotNull(handler, "handler");
                         if (IsCancellationRequested)
                             handler(CancellationReason);
                         if (m_cancelationHandlers == null)
-                            Interlocked.CompareExchange(ref m_cancelationHandlers, new MTQueue<Action<Exception>>(), null);
+                            Interlocked.CompareExchange(ref m_cancelationHandlers, new SimpleAsyncQueue<Action<Exception>>(), null);
                         m_cancelationHandlers.Enqueue(handler);
                         if (IsCancellationRequested && m_cancelationHandlers.TryDequeue(out handler))
                             // TryDeque implies MemoryBarrier()
                             handler(m_cancelationReason);
                     }
                     public bool IsCancellationRequested {
                         get {
                             do {
                                 if (m_cancelRequest == CANCEL_NOT_REQUESTED)
                                     return false;
                                 if (m_cancelRequest == CANCEL_REQUESTED)
                                     return true;
                                 Thread.MemoryBarrier();
                             } while(true);
                         }
                     }
                     public Exception CancellationReason {
                         get {
                             do {
                                 Thread.MemoryBarrier();
                             } while(m_cancelRequest == CANCEL_REQUESTING);
                             return m_cancelationReason;
                         }
                     }
                     #region ICancellable implementation
                     public void Cancel() {
                         Cancel(null);
                     }
                     public void Cancel(Exception reason) {
                         if (CANCEL_NOT_REQUESTED == Interlocked.CompareExchange(ref m_cancelRequest, CANCEL_REQUESTING, CANCEL_NOT_REQUESTED)) {
                             m_cancelationReason = reason;
                             m_cancelRequest = CANCEL_REQUESTED;
                             if (m_cancelationHandlers != null) {
                                 Action<Exception> handler;
                                 while (m_cancelationHandlers.TryDequeue(out handler))
                                     handler(m_cancelationReason);
                             }
                         }
                     }
                     #endregion
                 }
             }

Implab/Automaton/AutomatonTransition.cs +1 -1

             using System;
             namespace Implab.Automaton {
-                public struct AutomatonTransition : IEquatable<AutomatonTransition> {
+                public class AutomatonTransition : IEquatable<AutomatonTransition> {
                     public readonly int s1;
                     public readonly int s2;
                     public readonly int edge;
                     public AutomatonTransition(int s1, int s2, int edge) {
                         this.s1 = s1;
                         this.s2 = s2;
                         this.edge = edge;
                     }
                     #region IEquatable implementation
                     public bool Equals(AutomatonTransition other) {
                         return other.s1 == s1 && other.s2 == s2 && other.edge == edge ;
                     }
                     #endregion
                     public override bool Equals(object obj) {
                         if (obj is AutomatonTransition)
                             return Equals((AutomatonTransition)obj);
                         return base.Equals(obj);
                     }
                     public override int GetHashCode() {
                         return s1 + s2 + edge;
                     }
                     public static bool operator == (AutomatonTransition rv, AutomatonTransition lv) {
                         return rv.Equals(lv);
                     }
                     public static bool operator !=(AutomatonTransition rv, AutomatonTransition lv) {
                         return rv.Equals(lv);
                     }
                 }
             }

Implab/Implab.csproj +3 -2

             <?xml version="1.0" encoding="utf-8"?>
-            <Project DefaultTargets="Build" ToolsVersion="4.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
+            <Project DefaultTargets="Build" ToolsVersion="12.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>
                 <TargetFrameworkVersion>v4.5</TargetFrameworkVersion>
+                <TargetFrameworkProfile />
               </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" />
                 <Reference Include="mscorlib" />
                 <Reference Include="System.Xml.Linq" />
               </ItemGroup>
               <ItemGroup>
                 <Compile Include="Components\StateChangeEventArgs.cs" />
                 <Compile Include="CustomEqualityComparer.cs" />
                 <Compile Include="Diagnostics\ConsoleTraceListener.cs" />
                 <Compile Include="Diagnostics\LogChannel.cs" />
                 <Compile Include="Diagnostics\LogicalOperation.cs" />
                 <Compile Include="Diagnostics\TextFileListener.cs" />
                 <Compile Include="Diagnostics\Trace.cs" />
                 <Compile Include="Diagnostics\TraceLog.cs" />
                 <Compile Include="Diagnostics\TraceEvent.cs" />
                 <Compile Include="Diagnostics\TraceEventType.cs" />
                 <Compile Include="Diagnostics\TraceSourceAttribute.cs" />
                 <Compile Include="Formats\CharMap.cs" />
                 <Compile Include="Formats\InputScanner.cs" />
                 <Compile Include="Formats\Json\JsonStringScanner.cs" />
                 <Compile Include="Formats\Json\JsonTextScanner.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="Parallels\DispatchPool.cs" />
                 <Compile Include="Parallels\ArrayTraits.cs" />
-                <Compile Include="Parallels\MTQueue.cs" />
+                <Compile Include="Parallels\SimpleAsyncQueue.cs" />
                 <Compile Include="Parallels\WorkerPool.cs" />
                 <Compile Include="ProgressInitEventArgs.cs" />
                 <Compile Include="Properties\AssemblyInfo.cs" />
                 <Compile Include="Parallels\AsyncPool.cs" />
                 <Compile Include="Safe.cs" />
                 <Compile Include="SyncContextPromise.cs" />
                 <Compile Include="ValueEventArgs.cs" />
                 <Compile Include="PromiseExtensions.cs" />
                 <Compile Include="SyncContextPromiseT.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="PromiseEventType.cs" />
                 <Compile Include="Parallels\AsyncQueue.cs" />
                 <Compile Include="PromiseT.cs" />
                 <Compile Include="IDeferred.cs" />
                 <Compile Include="IDeferredT.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" />
                 <Compile Include="AbstractEvent.cs" />
                 <Compile Include="AbstractPromise.cs" />
                 <Compile Include="AbstractPromiseT.cs" />
                 <Compile Include="FuncTask.cs" />
                 <Compile Include="FuncTaskBase.cs" />
                 <Compile Include="FuncTaskT.cs" />
                 <Compile Include="ActionChainTaskBase.cs" />
                 <Compile Include="ActionChainTask.cs" />
                 <Compile Include="ActionChainTaskT.cs" />
                 <Compile Include="FuncChainTaskBase.cs" />
                 <Compile Include="FuncChainTask.cs" />
                 <Compile Include="FuncChainTaskT.cs" />
                 <Compile Include="ActionTaskBase.cs" />
                 <Compile Include="ActionTask.cs" />
                 <Compile Include="ActionTaskT.cs" />
                 <Compile Include="ICancellationToken.cs" />
                 <Compile Include="SuccessPromise.cs" />
                 <Compile Include="SuccessPromiseT.cs" />
                 <Compile Include="PromiseAwaiterT.cs" />
                 <Compile Include="PromiseAwaiter.cs" />
                 <Compile Include="Components\ComponentContainer.cs" />
                 <Compile Include="Components\Disposable.cs" />
                 <Compile Include="Components\DisposablePool.cs" />
                 <Compile Include="Components\ObjectPool.cs" />
                 <Compile Include="Components\ServiceLocator.cs" />
                 <Compile Include="Components\IInitializable.cs" />
                 <Compile Include="TaskController.cs" />
                 <Compile Include="Components\App.cs" />
                 <Compile Include="Components\IRunnable.cs" />
                 <Compile Include="Components\ExecutionState.cs" />
                 <Compile Include="Components\RunnableComponent.cs" />
                 <Compile Include="Components\IFactory.cs" />
                 <Compile Include="Automaton\IAlphabet.cs" />
                 <Compile Include="Automaton\ParserException.cs" />
                 <Compile Include="Automaton\IndexedAlphabetBase.cs" />
                 <Compile Include="Automaton\IAlphabetBuilder.cs" />
                 <Compile Include="Automaton\RegularExpressions\AltToken.cs" />
                 <Compile Include="Automaton\RegularExpressions\BinaryToken.cs" />
                 <Compile Include="Automaton\RegularExpressions\CatToken.cs" />
                 <Compile Include="Automaton\RegularExpressions\StarToken.cs" />
                 <Compile Include="Automaton\RegularExpressions\SymbolToken.cs" />
                 <Compile Include="Automaton\RegularExpressions\EmptyToken.cs" />
                 <Compile Include="Automaton\RegularExpressions\Token.cs" />
                 <Compile Include="Automaton\RegularExpressions\IVisitor.cs" />
                 <Compile Include="Automaton\AutomatonTransition.cs" />
                 <Compile Include="Formats\Json\JsonElementContext.cs" />
                 <Compile Include="Formats\Json\JsonElementType.cs" />
                 <Compile Include="Formats\Json\JsonGrammar.cs" />
                 <Compile Include="Formats\Json\JsonReader.cs" />
                 <Compile Include="Formats\Json\JsonScanner.cs" />
                 <Compile Include="Formats\Json\JsonTokenType.cs" />
                 <Compile Include="Formats\Json\JsonWriter.cs" />
                 <Compile Include="Formats\Json\StringTranslator.cs" />
                 <Compile Include="Automaton\MapAlphabet.cs" />
                 <Compile Include="Formats\CharAlphabet.cs" />
                 <Compile Include="Formats\ByteAlphabet.cs" />
                 <Compile Include="Automaton\IDFATable.cs" />
                 <Compile Include="Automaton\IDFATableBuilder.cs" />
                 <Compile Include="Automaton\DFATable.cs" />
                 <Compile Include="Automaton\RegularExpressions\RegularExpressionVisitor.cs" />
                 <Compile Include="Automaton\RegularExpressions\ITaggedDFABuilder.cs" />
                 <Compile Include="Formats\Grammar.cs" />
                 <Compile Include="Automaton\RegularExpressions\EndTokenT.cs" />
                 <Compile Include="Automaton\RegularExpressions\EndToken.cs" />
                 <Compile Include="Automaton\RegularExpressions\RegularExpressionVisitorT.cs" />
                 <Compile Include="Automaton\AutomatonConst.cs" />
                 <Compile Include="Automaton\RegularExpressions\RegularDFA.cs" />
                 <Compile Include="Components\LazyAndWeak.cs" />
                 <Compile Include="AbstractTask.cs" />
                 <Compile Include="AbstractTaskT.cs" />
                 <Compile Include="FailedPromise.cs" />
                 <Compile Include="FailedPromiseT.cs" />
                 <Compile Include="Components\PollingComponent.cs" />
                 <Compile Include="Xml\JsonXmlReader.cs" />
                 <Compile Include="Xml\JsonXmlReaderOptions.cs" />
                 <Compile Include="Xml\JsonXmlReaderPosition.cs" />
                 <Compile Include="Xml\SerializationHelpers.cs" />
                 <Compile Include="Xml\SerializersPool.cs" />
                 <Compile Include="Xml\XmlSimpleAttribute.cs" />
                 <Compile Include="Xml\XmlNameContext.cs" />
               </ItemGroup>
               <Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" />
               <ItemGroup />
               <ProjectExtensions>
                 <MonoDevelop>
                   <Properties>
                     <Policies>
                       <CSharpFormattingPolicy IndentBlock="True" IndentBraces="False" IndentSwitchSection="False" IndentSwitchCaseSection="True" LabelPositioning="OneLess" NewLinesForBracesInProperties="False" NewLinesForBracesInAccessors="False" NewLinesForBracesInAnonymousMethods="False" NewLinesForBracesInControlBlocks="False" NewLinesForBracesInAnonymousTypes="False" NewLinesForBracesInObjectCollectionArrayInitializers="False" NewLinesForBracesInLambdaExpressionBody="False" NewLineForElse="False" NewLineForCatch="False" NewLineForFinally="False" NewLineForMembersInObjectInit="False" NewLineForMembersInAnonymousTypes="False" NewLineForClausesInQuery="False" SpaceWithinMethodDeclarationParenthesis="False" SpaceBetweenEmptyMethodDeclarationParentheses="False" SpaceWithinMethodCallParentheses="False" SpaceBetweenEmptyMethodCallParentheses="False" SpaceAfterControlFlowStatementKeyword="True" SpaceWithinExpressionParentheses="False" SpaceWithinCastParentheses="False" SpaceWithinOtherParentheses="False" SpaceAfterCast="False" SpacesIgnoreAroundVariableDeclaration="False" SpaceBetweenEmptySquareBrackets="False" SpaceWithinSquareBrackets="False" SpaceAfterColonInBaseTypeDeclaration="True" SpaceAfterComma="True" SpaceAfterDot="False" SpaceAfterSemicolonsInForStatement="True" SpaceBeforeComma="False" SpaceBeforeDot="False" SpaceBeforeSemicolonsInForStatement="False" SpacingAroundBinaryOperator="Single" WrappingPreserveSingleLine="True" WrappingKeepStatementsOnSingleLine="True" PlaceSystemDirectiveFirst="True" NewLinesForBracesInTypes="True" NewLinesForBracesInMethods="True" SpacingAfterMethodDeclarationName="True" SpaceAfterMethodCallName="True" SpaceBeforeOpenSquareBracket="True" SpaceBeforeColonInBaseTypeDeclaration="True" scope="text/x-csharp" />
                       <TextStylePolicy FileWidth="120" TabWidth="4" IndentWidth="4" RemoveTrailingWhitespace="True" NoTabsAfterNonTabs="False" TabsToSpaces="True" EolMarker="Unix" scope="text/x-csharp" />
                       <DotNetNamingPolicy DirectoryNamespaceAssociation="PrefixedHierarchical" ResourceNamePolicy="MSBuild" />
                       <TextStylePolicy FileWidth="120" TabWidth="4" TabsToSpaces="False" IndentWidth="4" RemoveTrailingWhitespace="True" NoTabsAfterNonTabs="False" EolMarker="Native" scope="application/xml" />
                       <XmlFormattingPolicy scope="application/xml">
                         <DefaultFormat OmitXmlDeclaration="False" NewLineChars="&#xA;" IndentContent="True" ContentIndentString="	" AttributesInNewLine="False" MaxAttributesPerLine="10" AttributesIndentString="	" WrapAttributes="False" AlignAttributes="False" AlignAttributeValues="False" QuoteChar="&quot;" SpacesBeforeAssignment="0" SpacesAfterAssignment="0" EmptyLinesBeforeStart="0" EmptyLinesAfterStart="0" EmptyLinesBeforeEnd="0" EmptyLinesAfterEnd="0" />
                       </XmlFormattingPolicy>
                       <TextStylePolicy FileWidth="120" TabWidth="4" TabsToSpaces="False" IndentWidth="4" RemoveTrailingWhitespace="True" NoTabsAfterNonTabs="False" EolMarker="Native" 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>
               <ItemGroup />
             </Project>

Implab/Parallels/AsyncQueue.cs +162 -223

             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;
+                        public volatile Chunk next;
-                        int m_low;
+                        volatile int m_low;
-                        int m_hi;
+                        volatile int m_hi;
-                        int m_alloc;
+                        volatile 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) {
+                        public Chunk(int size, int allocated) {
                             m_size = size;
-                            m_hi = length;
+                            m_hi = allocated;
-                            m_alloc = alloc;
+                            m_alloc = allocated;
                             m_data = new T[size];
-                            Array.Copy(data, offset, m_data, 0, length);
+                        public void WriteData(T[] data, int offset, int dest, int length) {
+                            Array.Copy(data, offset, m_data, dest, 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) {
+                        public bool TryEnqueue(T value) {
-                            var alloc = Interlocked.Increment(ref m_alloc) - 1;
+                            int alloc;
+                            do {
-                            if (alloc >= m_size) {
+                                alloc = m_alloc;
-                                extend = alloc == m_size;
+                                if (alloc >= m_size)
-                                return false;
+                                    return false;
+                            } while(alloc != Interlocked.CompareExchange(ref m_alloc, alloc + 1, alloc));
-                            extend = false;
                             m_data[alloc] = value;
-                            while (alloc != Interlocked.CompareExchange(ref m_hi, alloc + 1, alloc)) {
+                            SpinWait spin = new SpinWait();
+                            // m_hi is volatile
+                            while (alloc != m_hi) {
                                 // spin wait for commit
+                                spin.SpinOnce();
                             }
+                            m_hi = alloc + 1;
                             return true;
                         }
                         /// <summary>
                         /// Prevents from allocating new space in the chunk and waits for all write operations to complete
                         /// </summary>
-                        public void Commit() {
+                        public void Seal() {
-                            var actual = Math.Min(Interlocked.Exchange(ref m_alloc, m_size + 1), m_size);
+                            var actual = Math.Min(Interlocked.Exchange(ref m_alloc, m_size), m_size);
+                            SpinWait spin = new SpinWait();
-                            while (m_hi != actual)
+                            while (m_hi != actual) {
-                                Thread.MemoryBarrier();
+                                spin.SpinOnce();
+                            }
                         }
                         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));
+                            } while (low != Interlocked.CompareExchange(ref m_low, low + 1, low));
-                            recycle = (low == m_size - 1);
+                            recycle = (low + 1 == m_size);
                             value = m_data[low];
                             return true;
                         }
-                        public bool TryEnqueueBatch(T[] batch, int offset, int length, out int enqueued, out bool extend) {
+                        public bool TryEnqueueBatch(T[] batch, int offset, int length, out int enqueued) {
-                            //int alloc;
+                            int alloc;
-                            //int allocSize;
+                            do {
+                                alloc = m_alloc;
+                                if (alloc >= m_size) {
+                                    enqueued = 0;
+                                    return false;
+                                } else {
+                                    enqueued = Math.Min(length, m_size - alloc);
+                                }
+                            } while (alloc != Interlocked.CompareExchange(ref m_alloc, alloc + enqueued, alloc));
+                            Array.Copy(batch, offset, m_data, alloc, enqueued);
-                            var alloc = Interlocked.Add(ref m_alloc, length) - length;
+                            SpinWait spin = new SpinWait();
-                            if (alloc > m_size) {
+                            while (alloc != m_hi) {
-                                // the chunk is full and someone already
+                                spin.SpinOnce();
-                                // 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);
+                            m_hi = alloc + enqueued;
-                            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) {
+                        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
+                                    recycle = (low == m_size);
                                     return false;
                                 }
                                 batchSize = Math.Min(hi - low, length);
-                            } while(low != Interlocked.CompareExchange(ref m_low, low + batchSize, low));
+                            } while (low != Interlocked.CompareExchange(ref m_low, low + batchSize, low));
-                            recycle = (low == m_size - batchSize);
                             dequeued = batchSize;
+                            recycle = (low + batchSize == m_size);
                             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;
+                    public const int MAX_CHUNK_SIZE = 256;
                     Chunk m_first;
                     Chunk m_last;
+                    public AsyncQueue() {
+                        m_first = m_last = new Chunk(DEFAULT_CHUNK_SIZE);
+                    }
                     /// <summary>
                     /// Adds the specified value to the queue.
                     /// </summary>
                     /// <param name="value">Tha value which will be added to the queue.</param>
-                    public virtual void Enqueue(T value) {
+                    public void Enqueue(T value) {
                         var last = m_last;
-                        // spin wait to the new chunk
+                        SpinWait spin = new SpinWait();
-                        bool extend = true;
+                        while (!last.TryEnqueue(value)) {
-                        while (last == null || !last.TryEnqueue(value, out extend)) {
                             // try to extend queue
-                            if (extend || last == null) {
+                            var chunk = new Chunk(DEFAULT_CHUNK_SIZE, value);
-                                var chunk = new Chunk(DEFAULT_CHUNK_SIZE, value);
+                            var t = Interlocked.CompareExchange(ref m_last, chunk, last);
-                                if (EnqueueChunk(last, chunk))
+                            if (t == last) {
-                                    break; // success! exit!
+                                last.next = chunk;
-                                last = m_last;
+                                break;
                             } else {
-                                while (last == m_last) {
+                                last = t;
-                                    Thread.MemoryBarrier();
-                                last = m_last;
                             }
+                            spin.SpinOnce();
                         }
                     }
                     /// <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 virtual void EnqueueRange(T[] data, int offset, int length) {
+                    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;
+                        while (length > 0) {
+                            var last = m_last;
+                            int enqueued;
-                        bool extend;
+                            if (last.TryEnqueueBatch(data, offset, length, out enqueued)) {
-                        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) {
+                            if (length > 0) {
-                                // there was no enough space in the chunk
+                                // we have something to enqueue
-                                // or there was no chunks in the queue
-                                while (length > 0) {
+                                var tail = length % MAX_CHUNK_SIZE;
-                                    var size = Math.Min(length, MAX_CHUNK_SIZE);
-                                    var chunk = new Chunk(
+                                var chunk = new Chunk(Math.Max(tail, DEFAULT_CHUNK_SIZE), tail);
-                                        Math.Max(size, DEFAULT_CHUNK_SIZE),
-                                        data,
+                                if (last != Interlocked.CompareExchange(ref m_last, chunk, last))
-                                        offset,
+                                    continue; // we wasn't able to catch the writer, roundtrip
-                                        size,
-                                        length // length >= size
-                                    );
-                                    if (!EnqueueChunk(last, chunk)) {
+                                // we are lucky
-                                        // looks like the queue has been updated then proceed from the beginning
+                                // we can exclusively write our batch, the other writers will continue their work
-                                        last = m_last;
-                                        break;
+                                length -= tail;
-                                    // we have successfully added the new chunk
-                                    last = chunk;
+                                for(var i = 0; i < length; i+= MAX_CHUNK_SIZE) {
-                                    length -= size;
+                                    var node = new Chunk(MAX_CHUNK_SIZE, MAX_CHUNK_SIZE);
-                                    offset += size;
+                                    node.WriteData(data, offset, 0, MAX_CHUNK_SIZE);
+                                    offset += MAX_CHUNK_SIZE;
-                            } else {
+                                    // fence last.next is volatile
-                                // we don't need to extend the queue, if we successfully enqueued data
+                                    last.next = node;
-                                if (length == 0)
+                                    last = node;
-                                    break;
-                                // if we need to wait while someone is extending the queue
-                                // spinwait
-                                while (last == m_last) {
-                                    Thread.MemoryBarrier();
                                 }
-                                last = m_last;
+                                if (tail > 0)
+                                    chunk.WriteData(data, offset, 0, tail);
+                                // fence last.next is volatile
+                                last.next = chunk;
+                                return;
                             }
                         }
                     }
                     /// <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;
+                        do {
-                        while (chunk != null) {
+                            bool recycle;
                             var result = chunk.TryDequeue(out value, out recycle);
-                            if (recycle) // this chunk is waste
+                            if (recycle && chunk.next != null) {
-                                RecycleFirstChunk(chunk);
+                                // this chunk is waste
-                            else
+                                chunk = Interlocked.CompareExchange(ref m_first, chunk.next, 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;
+                        } while (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) {
+                        do {
+                            bool recycle;
                             int actual;
                             if (chunk.TryDequeueBatch(buffer, offset, length, out actual, out recycle)) {
                                 offset += actual;
                                 length -= actual;
                                 dequeued += actual;
                             }
-                            if (recycle) // this chunk is waste
+                            if (recycle && chunk.next != null) {
-                                RecycleFirstChunk(chunk);
+                                // this chunk is waste
-                            else if (actual == 0)
+                                chunk = Interlocked.CompareExchange(ref m_first, chunk.next, chunk);
-                                break; // the chunk is usable, but doesn't contain any data (it's the last chunk in the queue)
+                            } else {
+                                chunk = null;
+                            }
                             if (length == 0)
                                 return true;
+                        } while (chunk != null);
-                            // 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;
+                        do {
-                        dequeued = 0;
+                            bool recycle;
+                            chunk.TryDequeueBatch(buffer, offset, length, out dequeued, out recycle);
-                        while (chunk != null) {
-                            int actual;
+                            if (recycle && chunk.next != null) {
-                            if (chunk.TryDequeueBatch(buffer, offset, length, out actual, out recycle)) {
+                                // this chunk is waste
-                                dequeued = actual;
+                                chunk = Interlocked.CompareExchange(ref m_first, chunk.next, chunk);
+                            } else {
+                                chunk = null;
                             }
-                            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
+                        } while (chunk != null);
-                            // 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)) {
-                                // race
-                                // someone already updated the tail, restore the pointer to the queue head
-                                m_first = first;
-                            // the tail is updated
                     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 (first.next == null && first != 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);
+                            m_last = chunk;
+                            return;
+                        } while (true);
+                    }
+                    public List<T> Drain() {
+                        // start the new queue
+                        var chunk = new Chunk(DEFAULT_CHUNK_SIZE);
+                        do {
+                            var first = m_first;
+                            // first.next is volatile
+                            if (first.next == null) {
+                                if (first != m_last)
+                                    continue;
+                                else if (first.Hi == first.Low)
+                                    return new List<T>();
+                            }
+                            // 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
+                            var last = Interlocked.Exchange(ref m_last, chunk);
                             return ReadChunks(first, last);
-                        } while(true);
+                        } while (true);
                     }
-                    static T[] ReadChunks(Chunk chunk, object last) {
+                    static List<T> ReadChunks(Chunk chunk, object last) {
                         var result = new List<T>();
-                        var buffer = new T[DEFAULT_CHUNK_SIZE];
+                        var buffer = new T[MAX_CHUNK_SIZE];
                         int actual;
                         bool recycle;
+                        SpinWait spin = new SpinWait();
                         while (chunk != null) {
                             // ensure all write operations on the chunk are complete
-                            chunk.Commit();
+                            chunk.Seal();
                             // 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();
+                                    spin.SpinOnce();
                                 chunk = chunk.next;
                             }
                         }
-                        return result.ToArray();
+                        return result;
                     }
                     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);
+                            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/BlockingQueue.cs +2 -2

             using System;
             using System.Threading;
             namespace Implab.Parallels {
                 public class BlockingQueue<T> : AsyncQueue<T> {
                     readonly object m_lock = new object();
-                    public override void Enqueue(T value) {
+                    public void EnqueuePulse(T value) {
                         base.Enqueue(value);
                         lock (m_lock)
                             Monitor.Pulse(m_lock);
                     }
-                    public override void EnqueueRange(T[] data, int offset, int length) {
+                    public void EnqueueRangePulse(T[] data, int offset, int length) {
                         base.EnqueueRange(data, offset, length);
                         if (length > 1)
                             lock (m_lock)
                                 Monitor.PulseAll(m_lock);
                         else
                             lock (m_lock)
                                 Monitor.Pulse(m_lock);
                     }
                     public T GetItem(int timeout) {
                         T item;
                         if (!TryDequeue(out item)) {
                             var t1 = Environment.TickCount;
                             var dt = timeout;
                             lock (m_lock) {
                                 while (!TryDequeue(out item)) {
                                     if (!Monitor.Wait(m_lock, dt))
                                         throw new TimeoutException();
                                     if (timeout >= 0) {
                                         dt = timeout - Environment.TickCount + t1;
                                         if (dt < 0)
                                             throw new TimeoutException();
                                     }
                                 }
                             }
                         }
                         return item;
                     }
                     public T GetItem() {
                         T item;
                         if (!TryDequeue(out item))
                             lock (m_lock) {
                                 while (!TryDequeue(out item))
                                     Monitor.Wait(m_lock);
                             }
                         return item;
                     }
                     public T[] GetRange(int max, int timeout) {
                         Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max");
                         var buffer = new T[max];
                         int actual;
                         if (!TryDequeueRange(buffer, 0, max, out actual)) {
                             var t1 = Environment.TickCount;
                             var dt = timeout;
                             lock (m_lock) {
                                 while (!TryDequeueRange(buffer, 0, max, out actual)) {
                                     if (!Monitor.Wait(m_lock, dt))
                                         throw new TimeoutException();
                                     if (timeout >= 0) {
                                         dt = timeout - Environment.TickCount + t1;
                                         if (dt < 0)
                                             throw new TimeoutException();
                                     }
                                 }
                             }
                         }
                         var data = new T[actual];
                         Array.Copy(buffer, data, actual);
                         return data;
                     }
                     public T[] GetRange(int max) {
                         Safe.ArgumentInRange(max, 1, Int32.MaxValue, "max");
                         var buffer = new T[max];
                         int actual;
                         if (!TryDequeueRange(buffer, 0, max, out actual))
                             lock (m_lock)
                                 while (!TryDequeueRange(buffer, 0, max, out actual))
                                     Monitor.Wait(m_lock);
                         var data = new T[actual];
                         Array.Copy(buffer, data, actual);
                         return data;
                     }
                 }
             }

Implab/Parallels/MTQueue.cs removed 0 -143

NO CONTENT: file was removed

General Comments 3

sergey

|

Auto status change to "Under Review"

sergey

|

ok, latest stable version should be in default

Write
Preview

You need to be logged in to leave comments. Login now

	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