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 ? 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 (() => {
106 foreach (TElement item in enumerables[index]) {
107 // This is from specific operators
108 if (options.ImplementerToken.IsCancellationRequested)
110 if (options.Token.IsCancellationRequested)
111 throw new OperationCanceledException (options.Token);
113 call (item, src.Token);
115 if (endAction != null)
123 internal static void ProcessAndBlock<T> (QueryBaseNode<T> node, Action<T, CancellationToken> call)
125 QueryOptions options = CheckQuery (node, true);
127 Task[] tasks = Process (node, call, (n, o) => n.GetEnumerables (o), options);
128 Task.WaitAll (tasks, options.Token);
131 internal static Action ProcessAndCallback<T> (QueryBaseNode<T> node, Action<T, CancellationToken> call,
132 Action callback, QueryOptions options)
134 Task[] tasks = Process (node, call, (n, o) => n.GetEnumerables (o), options);
135 if (callback != null)
136 Task.Factory.ContinueWhenAll (tasks, (_) => callback ());
138 return () => Task.WaitAll (tasks, options.Token);
141 internal static Action ProcessAndCallback<T> (QueryBaseNode<T> node, Action<KeyValuePair<long, T>, CancellationToken> call,
143 Action callback, QueryOptions options)
145 Task[] tasks = Process (node, call, (n, o) => n.GetOrderedEnumerables (o), endAction, options);
146 if (callback != null)
147 Task.Factory.ContinueWhenAll (tasks, (_) => callback ());
149 return () => Task.WaitAll (tasks, options.Token);
152 internal static void ProcessAndAggregate<T, U> (QueryBaseNode<T> node,
154 Func<U, T, U> localCall,
155 Action<IList<U>> call)
157 QueryOptions options = CheckQuery (node, true);
159 IList<IEnumerable<T>> enumerables = node.GetEnumerables (options);
160 U[] locals = new U[enumerables.Count];
161 Task[] tasks = new Task[enumerables.Count];
163 if (seedFunc != null) {
164 for (int i = 0; i < locals.Length; i++)
165 locals[i] = seedFunc ();
168 for (int i = 0; i < tasks.Length; i++) {
169 var procSlot = new AggregateProcessSlot<T, U> (options,
171 enumerables[i].GetEnumerator (),
176 tasks[i] = Task.Factory.StartNew (procSlot.Process, options.Token);
179 Task.WaitAll (tasks, options.Token);
185 class AggregateProcessSlot<T, U>
187 readonly QueryOptions options;
189 readonly IEnumerator<T> enumerator;
191 readonly Func<U, T, U> localCall;
192 readonly Func<U> seedFunc;
194 public AggregateProcessSlot (QueryOptions options,
196 IEnumerator<T> enumerator,
198 Func<U, T, U> localCall,
201 this.options = options;
203 this.enumerator = enumerator;
204 this.locals = locals;
205 this.localCall = localCall;
206 this.seedFunc = seedFunc;
209 public void Process ()
211 var token = options.Token;
212 var implementerToken = options.ImplementerToken;
215 if (seedFunc == null) {
216 if (!enumerator.MoveNext ())
218 locals[index] = (U)(object)enumerator.Current;
221 while (enumerator.MoveNext ()) {
222 if (implementerToken.IsCancellationRequested)
224 token.ThrowIfCancellationRequested ();
225 locals[index] = localCall (locals[index], enumerator.Current);
228 enumerator.Dispose ();