implab/ImplabNet Commit - r125:f803565868a4

improved performance of promises

cin -

r125:f803565868a4 v2

parent child

Context file:

r125:f803565868a4

Collapse all files

Implab.Test/AsyncTests.cs +1 -1

                     [TestMethod]
                     public void ChainedMapTest() {
-                        using (var pool = new WorkerPool(0,10,1)) {
+                        using (var pool = new WorkerPool()) {
                             const int count = 10000;
                             var args = new double[count];

Implab/AbstractPromise.cs +80 -10

@@ -12,10 +12,16 namespace Implab {
12	const int REJECTED_STATE = 3;	12	const int REJECTED_STATE = 3;
13	const int CANCELLED_STATE = 4;	13	const int CANCELLED_STATE = 4;
14		14
		15	const int RESERVED_HANDLERS_COUNT = 4;
		16
15	int m_state;	17	int m_state;
16	Exception m_error;	18	Exception m_error;
		19	int m_handlersCount;
17		20
18	readonly ~~AsyncQueue~~<THandler> m_handlers = new ~~AsyncQueue~~<THandler~~>()~~;	21	readonly THandler[] m_handlers = new THandler[RESERVED_HANDLERS_COUNT];
		22	MTQueue<THandler> m_extraHandlers;
		23	int m_handlerPointer = -1;
		24	int m_handlersCommited;
19		25
20	#region state managment	26	#region state managment
21	bool BeginTransit() {	27	bool BeginTransit() {
@@ -88,22 +94,59 namespace Implab {
88	protected abstract void SignalCancelled(THandler handler);	94	protected abstract void SignalCancelled(THandler handler);
89		95
90	void OnSuccess() {	96	void OnSuccess() {
		97	var hp = m_handlerPointer;
		98	var slot = hp +1 ;
		99	while (slot < m_handlersCommited) {
		100	if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) == hp) {
		101	SignalSuccess(m_handlers[slot]);
		102	}
		103	hp = m_handlerPointer;
		104	slot = hp +1 ;
		105	}
		106
		107
		108	if (m_extraHandlers != null) {
91	THandler handler;	109	THandler handler;
92	while (m_handlers.TryDequeue(out handler))	110	while (m_extraHandlers.TryDequeue(out handler))
93	SignalSuccess(handler);	111	SignalSuccess(handler);
94	}	112	}
		113	}
95		114
96	void OnError() {	115	void OnError() {
		116	var hp = m_handlerPointer;
		117	var slot = hp +1 ;
		118	while (slot < m_handlersCommited) {
		119	if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) == hp) {
		120	SignalError(m_handlers[slot],m_error);
		121	}
		122	hp = m_handlerPointer;
		123	slot = hp +1 ;
		124	}
		125
		126	if (m_extraHandlers != null) {
97	THandler handler;	127	THandler handler;
98	while (m_handlers.TryDequeue(out handler))	128	while (m_extraHandlers.TryDequeue(out handler))
99	SignalError(handler,m_error);	129	SignalError(handler, m_error);
100	}	130	}
		131	}
101		132
102	void OnCancelled() {	133	void OnCancelled() {
		134	var hp = m_handlerPointer;
		135	var slot = hp +1 ;
		136	while (slot < m_handlersCommited) {
		137	if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) == hp) {
		138	SignalCancelled(m_handlers[slot]);
		139	}
		140	hp = m_handlerPointer;
		141	slot = hp +1 ;
		142	}
		143
		144	if (m_extraHandlers != null) {
103	THandler handler;	145	THandler handler;
104	while (m_handlers.TryDequeue(out handler))	146	while (m_extraHandlers.TryDequeue(out handler))
105	SignalCancelled(handler);	147	SignalCancelled(handler);
106	}	148	}
		149	}
107		150
108	#endregion	151	#endregion
109		152
@@ -145,15 +188,41 namespace Implab {
145		188
146	protected void AddHandler(THandler handler) {	189	protected void AddHandler(THandler handler) {
147		190
148	if (~~IsResolved~~) {	191	if (m_state > 1) {
149	InvokeHandler(handler);
150
151	} else {
152	// the promise is in the resolved state, just invoke the handler	192	// the promise is in the resolved state, just invoke the handler
153	~~m_h~~andlers.~~Enqueue~~(handler);	193	InvokeHandler(handler);
		194	} else {
		195	var slot = Interlocked.Increment(ref m_handlersCount) - 1;
		196
		197	if (slot < RESERVED_HANDLERS_COUNT) {
		198	m_handlers[slot] = handler;
		199
		200	while (slot != Interlocked.CompareExchange(ref m_handlersCommited, slot + 1, slot)) {
		201	}
154		202
		203	if (m_state > 1) {
		204	do {
		205	var hp = m_handlerPointer;
		206	slot = hp + 1;
		207	if (slot < m_handlersCommited) {
		208	if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) != hp)
		209	continue;
		210	InvokeHandler(m_handlers[slot]);
		211	}
		212	break;
		213	} while(true);
		214	}
		215	} else {
		216	if (slot == RESERVED_HANDLERS_COUNT) {
		217	m_extraHandlers = new MTQueue<THandler>();
		218	} else {
		219	while (m_extraHandlers == null)
		220	Thread.MemoryBarrier();
		221	}
155		222
156	if (~~IsResolved~~ && ~~m_handlers~~.~~TryDequeue~~(~~out~~ handler))	223	m_extraHandlers.Enqueue(handler);
		224
		225	if (m_state > 1 && m_extraHandlers.TryDequeue(out handler))
157	// if the promise have been resolved while we was adding the handler to the queue	226	// if the promise have been resolved while we was adding the handler to the queue
158	// we can't guarantee that someone is still processing it	227	// we can't guarantee that someone is still processing it
159	// therefore we need to fetch a handler from the queue and execute it	228	// therefore we need to fetch a handler from the queue and execute it
@@ -162,6 +231,7 namespace Implab {
162	InvokeHandler(handler);	231	InvokeHandler(handler);
163	}	232	}
164	}	233	}
		234	}
165		235
166	protected void InvokeHandler(THandler handler) {	236	protected void InvokeHandler(THandler handler) {
167	switch (m_state) {	237	switch (m_state) {

Implab/Parallels/AsyncQueue.cs +5 -11

                     public const int DEFAULT_CHUNK_SIZE = 32;
                     public const int MAX_CHUNK_SIZE = 262144;
-                    readonly int m_chunkSize = DEFAULT_CHUNK_SIZE;
                     Chunk m_first;
                     Chunk m_last;
-                    public AsyncQueue() {
-                        m_last = m_first = new Chunk(m_chunkSize);
                     /// <summary>
                     /// Adds the specified value to the queue.
                     /// </summary>
                         while (last == null || !last.TryEnqueue(value, out extend)) {
                             // try to extend queue
                             if (extend || last == null) {
-                                var chunk = new Chunk(m_chunkSize, value);
+                                var chunk = new Chunk(DEFAULT_CHUNK_SIZE, value);
                                 if (EnqueueChunk(last, chunk))
                                     break; // success! exit!
                                 last = m_last;
                                     var size = Math.Min(length, MAX_CHUNK_SIZE);
                                     var chunk = new Chunk(
-                                        Math.Max(size, m_chunkSize),
+                                        Math.Max(size, DEFAULT_CHUNK_SIZE),
                                         data,
                                         offset,
                                         size,
                     public void Clear() {
                         // start the new queue
-                        var chunk = new Chunk(m_chunkSize);
+                        var chunk = new Chunk(DEFAULT_CHUNK_SIZE);
                         do {
                             Thread.MemoryBarrier();
                     public T[] Drain() {
                         // start the new queue
-                        var chunk = new Chunk(m_chunkSize);
+                        var chunk = new Chunk(DEFAULT_CHUNK_SIZE);
                         do {
                             Thread.MemoryBarrier();
                     T[] ReadChunks(Chunk chunk, object last) {
                         var result = new List<T>();
-                        var buffer = new T[m_chunkSize];
+                        var buffer = new T[DEFAULT_CHUNK_SIZE];
                         int actual;
                         bool recycle;
                         while (chunk != null) {

Implab/Parallels/DispatchPool.cs +1 -3

                     }
                     protected DispatchPool() {
-                        int maxThreads, maxCP;
-                        ThreadPool.GetMaxThreads(out maxThreads, out maxCP);
                         m_minThreadsLimit = 0;
-                        m_maxThreadsLimit = maxThreads;
+                        m_maxThreadsLimit = Environment.ProcessorCount;
                     }
                     protected void InitPool() {

MonoPlay/Program.cs +6 -92

@@ -11,109 +11,23 namespace MonoPlay {
11	if (args == null)	11	if (args == null)
12	throw new ArgumentNullException("args");	12	throw new ArgumentNullException("args");
13		13
14	var q1 = new AsyncQueue<int>();
15	var q2 = new Queue<int>();
16
17	const int count = 10000000;	14	const int count = 10000000;
18		15
19	int res1 = 0, res2 = 0;
20	var t1 = Environment.TickCount;	16	var t1 = Environment.TickCount;
21		17
22	AsyncPool.RunThread(	18	for (int i = 0; i < count; i++) {
23	() => {	19	var p = new Promise<int>();
24	for (var i = 0; i < count; i++)
25	q1.Enqueue(1);
26	Console.WriteLine("done writer #1: {0} ms", Environment.TickCount - t1);
27	},
28	() => {
29	for (var i = 0; i < count; i++)
30	q1.Enqueue(2);
31	Console.WriteLine("done writer #2: {0} ms", Environment.TickCount - t1);
32	},
33	() => {
34	int temp = 0;
35	int i = 0;
36	while (i < count)
37	if (q1.TryDequeue(out temp)) {
38	i++;
39	res1 += temp;
40	}
41	Console.WriteLine("done reader #1: {0} ms", Environment.TickCount - t1);
42	},
43	() => {
44	int temp = 0;
45	int i = 0;
46	while (i < count)
47	if (q1.TryDequeue(out temp)) {
48	i++;
49	res2 += temp;
50	}
51	Console.WriteLine("done reader #2: {0} ms", Environment.TickCount - t1);
52	}
53	)
54	.Bundle()
55	.Join();
56
57	Console.WriteLine("done: {0} ms, summ#1: {1}, summ#2: {2}, total: {3}, count: {4}", Environment.TickCount - t1, res1, res2, res1 + res2, count*2);
58
59	var t2 = Environment.TickCount;
60	Console.WriteLine("MTQueue: {0} ms", t2 - t1);
61
62	t1 = Environment.TickCount;
63		20
64	for (var i = 0; i < count * 2; i++)	21	p.On(x => {}).On(x => {});
65	q2.Enqueue(i);
66
67	for (var i = 0; i < count * 2; i++)
68	q2.Dequeue();
69
70	t2 = Environment.TickCount;
71	Console.WriteLine("Queue: {0} ms", t2 - t1);
72
73	q2 = new Queue<int>();
74
75	t1 = Environment.TickCount;
76
77		22
78	AsyncPool.RunThread(	23	p.Resolve(i);
79	() => {
80	for (var i = 0; i < count; i++)
81	lock (q2)
82	q2.Enqueue(i);
83	},
84	() => {
85	for (var i = 0; i < count; i++)
86	lock (q2)
87	q2.Enqueue(i);
88	},
89	() => {
90	for (int i = 0; i < count ;)
91	lock (q2) {
92	if (q2.Count == 0)
93	continue;
94	q2.Dequeue();
95	i++;
96	}
97
98	},
99	() => {
100	for (int i = 0; i < count ;)
101	lock (q2) {
102	if (q2.Count == 0)
103	continue;
104	q2.Dequeue();
105	i++;
106	}
107		24
108	}	25	}
109	)
110	.Bundle()
111	.Join();
112		26
113		27
114		28
115	t2 = Environment.TickCount;	29	var t2 = Environment.TickCount;
116	Console.WriteLine("Queue+Lock: {0} ms", t2 - t1);	30	Console.WriteLine("done: {0} ms, {1:.00} Mb, {2} GC", t2 - t1, GC.GetTotalMemory(false) / (1024*1024), GC.CollectionCount(0) );
117		31
118	}	32	}
119	}	33	}

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