5 // Jérémie "Garuma" Laval <jeremie.laval@gmail.com>
7 // Copyright (c) 2010 Jérémie "Garuma" Laval
9 // Permission is hereby granted, free of charge, to any person obtaining a copy
10 // of this software and associated documentation files (the "Software"), to deal
11 // in the Software without restriction, including without limitation the rights
12 // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
13 // copies of the Software, and to permit persons to whom the Software is
14 // furnished to do so, subject to the following conditions:
16 // The above copyright notice and this permission notice shall be included in
17 // all copies or substantial portions of the Software.
19 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
20 // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
21 // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
22 // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
23 // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
24 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
29 using System.Threading;
30 using System.Threading.Tasks;
31 using System.Collections;
32 using System.Collections.Generic;
33 using System.Collections.Concurrent;
34 using System.Linq.Parallel.QueryNodes;
36 namespace System.Linq.Parallel
38 internal static class ParallelExecuter
40 internal static QueryOptions CheckQuery<T> (QueryBaseNode<T> startingNode)
42 return CheckQuery<T> (startingNode, false);
45 internal static QueryOptions CheckQuery<T> (QueryBaseNode<T> startingNode, bool blocking)
47 return CheckQuery (startingNode, GetBestWorkerNumber (blocking));
50 internal static QueryOptions CheckQuery<T> (QueryBaseNode<T> startingNode, int partitionCount)
52 QueryCheckerVisitor visitor = new QueryCheckerVisitor (partitionCount);
53 startingNode.Visit (visitor);
55 return visitor.Options;
58 internal static CancellationToken Chain (this CancellationToken self, CancellationTokenSource other)
60 CancellationTokenSource linked = CancellationTokenSource.CreateLinkedTokenSource (self, other.Token);
64 internal static bool IsOrdered<TSource> (this QueryBaseNode<TSource> source)
66 QueryIsOrderedVisitor visitor = new QueryIsOrderedVisitor ();
67 source.Visit (visitor);
69 return visitor.BehindOrderGuard;
72 internal static int GetBestWorkerNumber ()
74 return GetBestWorkerNumber (false);
77 internal static int GetBestWorkerNumber (bool blocking)
79 return blocking && Task.CurrentId == null ? Environment.ProcessorCount + 1 : Environment.ProcessorCount;
82 internal static Task[] Process<TSource, TElement> (QueryBaseNode<TSource> node,
83 Action<TElement, CancellationToken> call,
84 Func<QueryBaseNode<TSource>, QueryOptions, IList<IEnumerable<TElement>>> acquisitionFunc,
87 return Process<TSource, TElement> (node, call, acquisitionFunc, null, options);
90 internal static Task[] Process<TSource, TElement> (QueryBaseNode<TSource> node,
91 Action<TElement, CancellationToken> call,
92 Func<QueryBaseNode<TSource>, QueryOptions, IList<IEnumerable<TElement>>> acquisitionFunc,
96 CancellationTokenSource src
97 = CancellationTokenSource.CreateLinkedTokenSource (options.ImplementerToken, options.Token);
99 IList<IEnumerable<TElement>> enumerables = acquisitionFunc (node, options);
101 Task[] tasks = new Task[enumerables.Count];
103 for (int i = 0; i < tasks.Length; i++) {
105 tasks[i] = Task.Factory.StartNew (() => {
107 foreach (TElement item in enumerables[index]) {
108 // This is from specific operators
109 if (options.ImplementerToken.IsCancellationRequested)
111 if (options.Token.IsCancellationRequested)
112 throw new OperationCanceledException (options.Token);
114 call (item, src.Token);
117 if (endAction != null)
120 }, options.Token, TaskCreationOptions.AttachedToParent, TaskScheduler.Default);
126 internal static void ProcessAndBlock<T> (QueryBaseNode<T> node, Action<T, CancellationToken> call)
128 QueryOptions options = CheckQuery (node, true);
130 Task[] tasks = Process (node, call, (n, o) => n.GetEnumerables (o), options);
131 Task.WaitAll (tasks, options.Token);
134 internal static Action ProcessAndCallback<T> (QueryBaseNode<T> node, Action<T, CancellationToken> call,
135 Action callback, QueryOptions options)
137 Task[] tasks = Process (node, call, (n, o) => n.GetEnumerables (o), options);
138 if (callback != null)
139 Task.Factory.ContinueWhenAll (tasks, (_) => callback ());
141 return () => Task.WaitAll (tasks, options.Token);
144 internal static Action ProcessAndCallback<T> (QueryBaseNode<T> node, Action<KeyValuePair<long, T>, CancellationToken> call,
146 Action callback, QueryOptions options)
148 Task[] tasks = Process (node, call, (n, o) => n.GetOrderedEnumerables (o), endAction, options);
149 if (callback != null)
150 Task.Factory.ContinueWhenAll (tasks, (_) => callback ());
152 return () => Task.WaitAll (tasks, options.Token);
155 internal static void ProcessAndAggregate<T, U> (QueryBaseNode<T> node,
157 Func<U, T, U> localCall,
158 Action<IList<U>> call)
160 QueryOptions options = CheckQuery (node, true);
162 IList<IEnumerable<T>> enumerables = node.GetEnumerables (options);
163 U[] locals = new U[enumerables.Count];
164 Task[] tasks = new Task[enumerables.Count];
166 if (seedFunc != null) {
167 for (int i = 0; i < locals.Length; i++)
168 locals[i] = seedFunc ();
171 for (int i = 0; i < tasks.Length; i++) {
172 var procSlot = new AggregateProcessSlot<T, U> (options,
174 enumerables[i].GetEnumerator (),
179 tasks[i] = Task.Factory.StartNew (procSlot.Process, options.Token);
182 Task.WaitAll (tasks, options.Token);
188 class AggregateProcessSlot<T, U>
190 readonly QueryOptions options;
192 readonly IEnumerator<T> enumerator;
194 readonly Func<U, T, U> localCall;
195 readonly Func<U> seedFunc;
197 public AggregateProcessSlot (QueryOptions options,
199 IEnumerator<T> enumerator,
201 Func<U, T, U> localCall,
204 this.options = options;
206 this.enumerator = enumerator;
207 this.locals = locals;
208 this.localCall = localCall;
209 this.seedFunc = seedFunc;
212 public void Process ()
214 var token = options.Token;
215 var implementerToken = options.ImplementerToken;
218 if (seedFunc == null) {
219 if (!enumerator.MoveNext ())
221 locals[index] = (U)(object)enumerator.Current;
224 while (enumerator.MoveNext ()) {
225 if (implementerToken.IsCancellationRequested)
227 token.ThrowIfCancellationRequested ();
228 locals[index] = localCall (locals[index], enumerator.Current);
231 enumerator.Dispose ();