implab/ImplabNet Commit - r124:a336cb13c6a9

major update, added Drain mathod to AsyncQueue class

cin -

r124:a336cb13c6a9 v2

parent child

Context file:

r124:a336cb13c6a9

Collapse all files

Implab.Test/AsyncTests.cs +108 -5

@@ -249,7 +249,7 namespace Implab.Test {
249	for (int i = 0; i < writersCount; i++) {	249	for (int i = 0; i < writersCount; i++) {
250	Interlocked.Increment(ref writers);	250	Interlocked.Increment(ref writers);
251	AsyncPool	251	AsyncPool
252	.~~InvokeNew~~Thread(() => {	252	.RunThread(() => {
253	for (int ii = 0; ii < itemsPerWriter; ii++) {	253	for (int ii = 0; ii < itemsPerWriter; ii++) {
254	queue.Enqueue(1);	254	queue.Enqueue(1);
255	}	255	}
@@ -261,7 +261,7 namespace Implab.Test {
261	for (int i = 0; i < 10; i++) {	261	for (int i = 0; i < 10; i++) {
262	Interlocked.Increment(ref readers);	262	Interlocked.Increment(ref readers);
263	AsyncPool	263	AsyncPool
264	.~~InvokeNew~~Thread(() => {	264	.RunThread(() => {
265	int t;	265	int t;
266	do {	266	do {
267	while (queue.TryDequeue(out t))	267	while (queue.TryDequeue(out t))
@@ -336,7 +336,7 namespace Implab.Test {
336	Console.WriteLine("done reader #2: {0} ms", Environment.TickCount - t1);	336	Console.WriteLine("done reader #2: {0} ms", Environment.TickCount - t1);
337	}	337	}
338	)	338	)
339	.~~Combin~~e()	339	.Bundle()
340	.Join();	340	.Join();
341		341
342	Assert.AreEqual(count * 3, res1 + res2);	342	Assert.AreEqual(count * 3, res1 + res2);
@@ -414,7 +414,7 namespace Implab.Test {
414	Console.WriteLine("done reader #2: {0} ms", Environment.TickCount - t1);	414	Console.WriteLine("done reader #2: {0} ms", Environment.TickCount - t1);
415	}	415	}
416	)	416	)
417	.~~Combin~~e()	417	.Bundle()
418	.Join();	418	.Join();
419		419
420	Assert.AreEqual(summ , r1 + r2);	420	Assert.AreEqual(summ , r1 + r2);
@@ -490,7 +490,110 namespace Implab.Test {
490	Console.WriteLine("done reader #2: {0} ms, avg chunk size: {1}", Environment.TickCount - t1, avgchunk);	490	Console.WriteLine("done reader #2: {0} ms, avg chunk size: {1}", Environment.TickCount - t1, avgchunk);
491	}	491	}
492	)	492	)
493	.~~Combin~~e()	493	.Bundle()
		494	.Join();
		495
		496	Assert.AreEqual(summ , r1 + r2);
		497
		498	Console.WriteLine(
		499	"done: {0} ms, summ#1: {1}, summ#2: {2}, total: {3}, count: {4}",
		500	Environment.TickCount - t1,
		501	r1,
		502	r2,
		503	r1 + r2,
		504	total
		505	);
		506	}
		507
		508	[TestMethod]
		509	public void AsyncQueueDrainTest() {
		510	var queue = new AsyncQueue<int>();
		511
		512	const int wBatch = 11;
		513	const int wCount = 200000;
		514	const int total = wBatch * wCount * 3;
		515	const int summ = wBatch * wCount * 3;
		516
		517	int r1 = 0, r2 = 0;
		518	const int rBatch = 11;
		519	int read = 0;
		520
		521	var t1 = Environment.TickCount;
		522
		523	AsyncPool.RunThread(
		524	() => {
		525	var buffer = new int[wBatch];
		526	for(int i = 0; i<wBatch; i++)
		527	buffer[i] = 1;
		528
		529	for(int i =0; i < wCount; i++)
		530	queue.EnqueueRange(buffer,0,wBatch);
		531	Console.WriteLine("done writer #1: {0} ms", Environment.TickCount - t1);
		532	},
		533	() => {
		534	for(int i =0; i < wCount * wBatch; i++)
		535	queue.Enqueue(1);
		536	Console.WriteLine("done writer #2: {0} ms", Environment.TickCount - t1);
		537	},
		538	() => {
		539	var buffer = new int[wBatch];
		540	for(int i = 0; i<wBatch; i++)
		541	buffer[i] = 1;
		542
		543	for(int i =0; i < wCount; i++)
		544	queue.EnqueueRange(buffer,0,wBatch);
		545	Console.WriteLine("done writer #3: {0} ms", Environment.TickCount - t1);
		546	},
		547	/*() => {
		548	int temp;
		549	int count = 0;
		550	while (read < total)
		551	if (queue.TryDequeue(out temp)) {
		552	count++;
		553	r1 += temp;
		554	Interlocked.Increment(ref read);
		555	}
		556	Console.WriteLine("done reader #1: {0} ms, {1} count", Environment.TickCount - t1, count);
		557	},*/
		558	/*() => {
		559	var buffer = new int[rBatch];
		560	var count = 0;
		561	while(read < total) {
		562	int actual;
		563	if (queue.TryDequeueRange(buffer,0,rBatch,out actual)) {
		564	for(int i=0; i< actual; i++)
		565	r1 += buffer[i];
		566	Interlocked.Add(ref read, actual);
		567	count += actual;
		568	}
		569	}
		570
		571	Console.WriteLine("done reader #1: {0} ms, {1} items", Environment.TickCount - t1, count);
		572	},*/
		573	() => {
		574	var count = 0;
		575	while(read < total) {
		576	var buffer = queue.Drain();
		577	for(int i=0; i< buffer.Length; i++)
		578	r1 += buffer[i];
		579	Interlocked.Add(ref read, buffer.Length);
		580	count += buffer.Length;
		581	}
		582	Console.WriteLine("done reader #1: {0} ms, {1} items", Environment.TickCount - t1, count);
		583	},
		584	() => {
		585	var count = 0;
		586	while(read < total) {
		587	var buffer = queue.Drain();
		588	for(int i=0; i< buffer.Length; i++)
		589	r2 += buffer[i];
		590	Interlocked.Add(ref read, buffer.Length);
		591	count += buffer.Length;
		592	}
		593	Console.WriteLine("done reader #2: {0} ms, {1} items", Environment.TickCount - t1, count);
		594	}
		595	)
		596	.Bundle()
494	.Join();	597	.Join();
495		598
496	Assert.AreEqual(summ , r1 + r2);	599	Assert.AreEqual(summ , r1 + r2);

Implab/Parallels/ArrayTraits.cs +1 -1

                         int slots = threads;
                         // Analysis disable AccessToDisposedClosure
-                        AsyncPool.InvokeNewThread<int>(() => {
+                        AsyncPool.RunThread<int>(() => {
                             for (int i = 0; i < source.Length; i++) {
                                 if(promise.IsResolved)
                                     break; // stop processing in case of error or cancellation

Implab/Parallels/AsyncPool.cs +4 -4

             			return p;
             		}
-                    public static IPromise<T> InvokeNewThread<T>(Func<T> func) {
+                    public static IPromise<T> RunThread<T>(Func<T> func) {
                         var p = new Promise<T>();
                         var caller = TraceContext.Instance.CurrentOperation;
                     }
-                    public static IPromise InvokeNewThread(Action func) {
+                    public static IPromise RunThread(Action func) {
                         var p = new Promise();
                         var caller = TraceContext.Instance.CurrentOperation;
                     }
                     public static IPromise[] RunThread(params Action[] func) {
-                        return func.Select(f => InvokeNewThread(f)).ToArray();
+                        return func.Select(f => RunThread(f)).ToArray();
                     }
                     public static IPromise<T>[] RunThread<T>(params Func<T>[] func) {
-                        return func.Select(f => InvokeNewThread(f)).ToArray();
+                        return func.Select(f => RunThread(f)).ToArray();
                     }
             	}
             }

Implab/Parallels/AsyncQueue.cs +73 -24

             using System.Collections.Generic;
             using System;
             using System.Collections;
+            using System.Diagnostics;
             namespace Implab.Parallels {
                 public class AsyncQueue<T> : IEnumerable<T> {
                             return true;
                         }
+                        /// <summary>
+                        /// Prevents from allocating new space in the chunk and waits for all write operations to complete
+                        /// </summary>
+                        public void Commit() {
+                            var actual = Math.Min(Interlocked.Exchange(ref m_alloc, m_size + 1), m_size);
+                            while (m_hi != actual)
+                                Thread.MemoryBarrier();
+                        }
                         public bool TryDequeue(out T value, out bool recycle) {
                             int low;
                             do {
                         if (last != null)
                             last.next = chunk;
-                        else
+                        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) {
-                            // looks like this is the last chunk
                             if (first != Interlocked.CompareExchange(ref m_last, null, first)) {
+                                /*while (first.next == null)
+                                    Thread.MemoryBarrier();*/
                                 // race
-                                // maybe someone already recycled this chunk
+                                // someone already updated the tail, restore the pointer to the queue head
-                                // or a new chunk has been appedned to the queue
+                                m_first = first;
-                                return; // give up
                             }
                             // the tail is updated
                         }
                         // we need to update the head
-                        Interlocked.CompareExchange(ref m_first, next, first);
+                        //Interlocked.CompareExchange(ref m_first, next, first);
                         // if the head is already updated then give up
-                        return;
+                        //return;
                     }
                     public void Clear() {
                         // start the new queue
-                        var t = new Chunk(m_chunkSize);
+                        var chunk = new Chunk(m_chunkSize);
-                        Thread.MemoryBarrier();
-                        m_last = t;
+                        do {
                             Thread.MemoryBarrier();
+                            var first = m_first;
+                            var last = m_last;
-                        // make the new queue available to the readers, and stop the old one
+                            if (last == null) // nothing to clear
-                        m_first = t;
+                                return;
-                        Thread.MemoryBarrier();
+                            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 t = new Chunk(m_chunkSize);
+                        var chunk = new Chunk(m_chunkSize);
-                        Thread.MemoryBarrier();
-                        m_last = t;
-                        Thread.MemoryBarrier();
-                        // make the new queue available to the readers, and stop the old one
-                        Chunk first;
                         do {
-                            first = m_first;
-                        } while(first != Interlocked.CompareExchange(ref m_first
                             Thread.MemoryBarrier();
+                            var first = m_first;
+                            var last = m_last;
+                            if (last == null)
+                                return new T[0];
+                            if (first == null || (first.next == null && first != last))
+                                continue;
+                            // here we will create inconsistency which will force others to spin
+                            // and prevent from fetching. chunk.next = null
+                            if (first != Interlocked.CompareExchange(ref m_first, chunk, first))
+                                continue;// inconsistent
+                            last = Interlocked.Exchange(ref m_last, chunk);
+                            return ReadChunks(first, last);
+                        } while(true);
                     }
-                    T[] ReadChunks(Chunk chunk) {
+                    T[] ReadChunks(Chunk chunk, object last) {
                         var result = new List<T>();
                         var buffer = new T[m_chunkSize];
                         int actual;
                         bool recycle;
                         while (chunk != null) {
+                            // ensure all write operations on the chunk are complete
+                            chunk.Commit();
                             // we need to read the chunk using this way
                             // since some client still may completing the dequeue
                             // operation, such clients most likely won't get results
                             while (chunk.TryDequeueBatch(buffer, 0, buffer.Length, out actual, out recycle))
                                 result.AddRange(new ArraySegmentCollection(buffer, 0, actual));
+                            if (chunk == last) {
+                                chunk = null;
+                            } else {
+                                while (chunk.next == null)
+                                    Thread.MemoryBarrier();
                                 chunk = chunk.next;
                             }
+                        }
                         return result.ToArray();
                     }

Implab/PromiseExtensions.cs +56 -3

@@ -94,12 +94,18 namespace Implab {
94	return that;	94	return that;
95	}	95	}
96		96
97	public static IPromise ~~Combin~~e(this ICollection<IPromise> that) {	97	public static IPromise Bundle(this ICollection<IPromise> that) {
98	Safe.ArgumentNotNull(that, "that");	98	Safe.ArgumentNotNull(that, "that");
99		99
100	int count = that.Count;	100	int count = that.Count;
		101	int errors = 0;
101	var medium = new Promise();	102	var medium = new Promise();
102		103
		104	medium.On(() => {
		105	foreach(var p2 in that)
		106	p2.Cancel();
		107	}, PromiseEventType.ErrorOrCancel);
		108
103	foreach (var p in that)	109	foreach (var p in that)
104	p.On(	110	p.On(
105	() => {	111	() => {
@@ -107,16 +113,63 namespace Implab {
107	medium.Resolve();	113	medium.Resolve();
108	},	114	},
109	error => {	115	error => {
110	throw new Exception("The dependency promise is failed", error);	116	if (Interlocked.Increment(ref errors) == 1)
		117	medium.Reject(
		118	new Exception("The dependency promise is failed", error)
		119	);
111	},	120	},
112	() => {	121	() => {
113	throw new OperationCanceledException("The dependency promise is cancelled");	122	if (Interlocked.Increment(ref errors) == 1)
		123	medium.Reject(
		124	new Exception("The dependency promise is cancelled")
		125	);
114	}	126	}
115	);	127	);
116		128
117	return medium;	129	return medium;
118	}	130	}
119		131
		132	public static IPromise<T[]> Bundle<T>(this ICollection<IPromise<T>> that) {
		133	Safe.ArgumentNotNull(that, "that");
		134
		135	int count = that.Count;
		136	int errors = 0;
		137	var medium = new Promise<T[]>();
		138	var results = new T[that.Count];
		139
		140	medium.On(() => {
		141	foreach(var p2 in that)
		142	p2.Cancel();
		143	}, PromiseEventType.ErrorOrCancel);
		144
		145	int i = 0;
		146	foreach (var p in that) {
		147	var idx = i;
		148	p.On(
		149	x => {
		150	results[idx] = x;
		151	if (Interlocked.Decrement(ref count) == 0)
		152	medium.Resolve(results);
		153	},
		154	error => {
		155	if (Interlocked.Increment(ref errors) == 1)
		156	medium.Reject(
		157	new Exception("The dependency promise is failed", error)
		158	);
		159	},
		160	() => {
		161	if (Interlocked.Increment(ref errors) == 1)
		162	medium.Reject(
		163	new Exception("The dependency promise is cancelled")
		164	);
		165	}
		166	);
		167	i++;
		168	}
		169
		170	return medium;
		171	}
		172
120	#if NET_4_5	173	#if NET_4_5
121		174
122	public static Task<T> GetTask<T>(this IPromise<T> that) {	175	public static Task<T> GetTask<T>(this IPromise<T> that) {

MonoPlay/Program.cs +2 -2

                                 Console.WriteLine("done reader #2: {0} ms", Environment.TickCount - t1);
                             }
                         )
-                            .Combine()
+                            .Bundle()
                             .Join();
                         Console.WriteLine("done: {0} ms, summ#1: {1}, summ#2: {2}, total: {3}, count: {4}", Environment.TickCount - t1, res1, res2, res1 + res2, count*2);
                             }
                         )
-                            .Combine()
+                            .Bundle()
                             .Join();

General Comments 0

Write
Preview

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

No TODOs yet

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages