Mercurial > pub > ImplabNet

comparison Implab/Parallels/ArrayTraits.cs @ 15:0f982f9b7d4d promises

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
implemented parallel map and foreach for arrays rewritten WorkerPool with MTQueue for more efficiency
author cin
date Thu, 07 Nov 2013 03:41:32 +0400
parents
children 5a4b735ba669
comparison
equal deleted inserted replaced
14:e943453e5039 15:0f982f9b7d4d
1 using System;
2 using System.Collections.Generic;
3 using System.Diagnostics;
4 using System.Linq;
5 using System.Text;
6 using System.Threading;
7
8 namespace Implab.Parallels {
9 public static class ArrayTraits {
10 class ArrayIterator<TSrc> : DispatchPool<int> {
11 readonly Action<TSrc> m_action;
12 readonly TSrc[] m_source;
13 readonly Promise<int> m_promise = new Promise<int>();
14
15 int m_pending;
16 int m_next;
17
18 public ArrayIterator(TSrc[] source, Action<TSrc> action, int threads)
19 : base(threads) {
20
21 Debug.Assert(source != null);
22 Debug.Assert(action != null);
23
24 m_next = 0;
25 m_source = source;
26 m_pending = source.Length;
27 m_action = action;
28
29 m_promise.Anyway(() => Dispose());
30 m_promise.Cancelled(() => Dispose());
31
32 InitPool();
33 }
34
35 public Promise<int> Promise {
36 get {
37 return m_promise;
38 }
39 }
40
41 protected override bool TryDequeue(out int unit) {
42 int index;
43 unit = -1;
44 do {
45 index = m_next;
46 if (index >= m_source.Length)
47 return false;
48 } while (index != Interlocked.CompareExchange(ref m_next, index + 1, index));
49
50 unit = index;
51 return true;
52 }
53
54 protected override void InvokeUnit(int unit) {
55 try {
56 m_action(m_source[unit]);
57 int pending;
58 do {
59 pending = m_pending;
60 } while (pending != Interlocked.CompareExchange(ref m_pending, pending - 1, pending));
61 pending--;
62 if (pending == 0)
63 m_promise.Resolve(m_source.Length);
64 } catch (Exception e) {
65 m_promise.Reject(e);
66 }
67 }
68 }
69
70 class ArrayMapper<TSrc, TDst>: DispatchPool<int> {
71 readonly Func<TSrc, TDst> m_transform;
72 readonly TSrc[] m_source;
73 readonly TDst[] m_dest;
74 readonly Promise<TDst[]> m_promise = new Promise<TDst[]>();
75
76 int m_pending;
77 int m_next;
78
79 public ArrayMapper(TSrc[] source, Func<TSrc, TDst> transform, int threads)
80 : base(threads) {
81
82 Debug.Assert (source != null);
83 Debug.Assert( transform != null);
84
85 m_next = 0;
86 m_source = source;
87 m_dest = new TDst[source.Length];
88 m_pending = source.Length;
89 m_transform = transform;
90
91 m_promise.Anyway(() => Dispose());
92 m_promise.Cancelled(() => Dispose());
93
94 InitPool();
95 }
96
97 public Promise<TDst[]> Promise {
98 get {
99 return m_promise;
100 }
101 }
102
103 protected override bool TryDequeue(out int unit) {
104 int index;
105 unit = -1;
106 do {
107 index = m_next;
108 if (index >= m_source.Length)
109 return false;
110 } while (index != Interlocked.CompareExchange(ref m_next, index + 1, index));
111
112 unit = index;
113 return true;
114 }
115
116 protected override void InvokeUnit(int unit) {
117 try {
118 m_dest[unit] = m_transform(m_source[unit]);
119 int pending;
120 do {
121 pending = m_pending;
122 } while ( pending != Interlocked.CompareExchange(ref m_pending,pending -1, pending));
123 pending --;
124 if (pending == 0)
125 m_promise.Resolve(m_dest);
126 } catch (Exception e) {
127 m_promise.Reject(e);
128 }
129 }
130 }
131
132 public static Promise<TDst[]> ParallelMap<TSrc, TDst> (this TSrc[] source, Func<TSrc,TDst> transform, int threads) {
133 if (source == null)
134 throw new ArgumentNullException("source");
135 if (transform == null)
136 throw new ArgumentNullException("transform");
137
138 var mapper = new ArrayMapper<TSrc, TDst>(source, transform, threads);
139 return mapper.Promise;
140 }
141
142 public static Promise<int> ParallelForEach<TSrc>(this TSrc[] source, Action<TSrc> action, int threads) {
143 if (source == null)
144 throw new ArgumentNullException("source");
145 if (action == null)
146 throw new ArgumentNullException("action");
147
148 var iter = new ArrayIterator<TSrc>(source, action, threads);
149 return iter.Promise;
150 }
151 }
152 }