+#if NET_4_0
+//
+// ParallelEnumerable.cs
+//
+// Author:
+// Jérémie "Garuma" Laval <jeremie.laval@gmail.com>
+//
+// Copyright (c) 2010 Jérémie "Garuma" Laval
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+using System;
+using System.Threading;
+using System.Collections;
+using System.Collections.Generic;
+using System.Collections.Concurrent;
+
+namespace System.Linq
+{
+ public static class ParallelEnumerable
+ {
+ #region Range & Repeat
+ public static ParallelQuery<int> Range (int start, int count)
+ {
+ if (int.MaxValue - start < count)
+ throw new ArgumentOutOfRangeException ("count", "start + count - 1 is larger than Int32.MaxValue");
+ if (count < 0)
+ throw new ArgumentOutOfRangeException ("count", "count is less than 0");
+
+ return (new RangeList (start, count)).AsParallel ();
+ }
+
+ public static ParallelQuery<TResult> Repeat<TResult> (TResult obj, int count)
+ {
+ if (count < 0)
+ throw new ArgumentOutOfRangeException ("count", "count is less than 0");
+
+ return (new RepeatList<TResult> (obj, count)).AsParallel ();
+ }
+ #endregion
+
+ #region Empty
+ public static ParallelQuery<TResult> Empty<TResult> ()
+ {
+ return Repeat<TResult> (default (TResult), 0);
+ }
+ #endregion
+
+ #region AsParallel
+ public static ParallelQuery<TSource> AsParallel<TSource> (this IEnumerable<TSource> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return new ParallelQuery<TSource> (new QueryStartNode<TSource> (source));
+ }
+
+ public static ParallelQuery<TSource> AsParallel<TSource> (this Partitioner<TSource> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return new ParallelQuery<TSource> (new QueryStartNode<TSource> (source));
+ }
+
+ public static ParallelQuery AsParallel (this IEnumerable source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return new ParallelQuery<object> (new QueryStartNode<object> (source.Cast<object> ()));
+ }
+
+ public static IEnumerable<TSource> AsEnumerable<TSource> (this ParallelQuery<TSource> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.AsSequential ();
+ }
+
+ public static IEnumerable<TSource> AsSequential<TSource> ( this ParallelQuery<TSource> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Node.GetSequential ();
+ }
+ #endregion
+
+ #region AsOrdered / AsUnordered
+ public static ParallelQuery<TSource> AsOrdered<TSource> (this ParallelQuery<TSource> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return new ParallelQuery<TSource> (new QueryAsOrderedNode<TSource> (source.Node));
+ }
+
+ public static ParallelQuery<TSource> AsUnordered<TSource> (this ParallelQuery<TSource> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return new ParallelQuery<TSource> (new QueryAsUnorderedNode<TSource> (source.Node));
+ }
+
+ public static ParallelQuery AsOrdered (ParallelQuery source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.TypedQuery.AsOrdered ();
+ }
+ #endregion
+
+ #region With*
+ public static ParallelQuery<TSource> WithExecutionMode<TSource> (this ParallelQuery<TSource> source,
+ ParallelExecutionMode executionMode)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return new ParallelQuery<TSource> (new ParallelExecutionModeNode<TSource> (executionMode, source.Node));
+ }
+
+ public static ParallelQuery<TSource> WithCancellation<TSource> (this ParallelQuery<TSource> source,
+ CancellationToken cancellationToken)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return new ParallelQuery<TSource> (new CancellationTokenNode<TSource> (cancellationToken, source.Node));
+ }
+
+ public static ParallelQuery<TSource> WithMergeOptions<TSource> (this ParallelQuery<TSource> source,
+ ParallelMergeOptions mergeOptions)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return new ParallelQuery<TSource> (new ParallelMergeOptionsNode<TSource> (mergeOptions, source.Node));
+ }
+
+ public static ParallelQuery<TSource> WithDegreeOfParallelism<TSource> (this ParallelQuery<TSource> source,
+ int degreeParallelism)
+ {
+ if (degreeParallelism < 1 || degreeParallelism > 63)
+ throw new ArgumentException ("degreeOfParallelism is less than 1 or greater than 63", "degreeParallelism");
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return new ParallelQuery<TSource> (new DegreeOfParallelismNode<TSource> (degreeParallelism, source.Node));
+ }
+
+ internal static ParallelQuery<TSource> WithImplementerToken<TSource> (this ParallelQuery<TSource> source,
+ CancellationTokenSource token)
+ {
+ return new ParallelQuery<TSource> (new ImplementerTokenNode<TSource> (token, source.Node));
+ }
+ #endregion
+
+ #region Select
+ public static ParallelQuery<TResult> Select<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> selector)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (selector == null)
+ throw new ArgumentNullException ("selector");
+
+ return new ParallelQuery<TResult> (new QuerySelectNode<TResult, TSource> (source.Node, selector));
+ }
+
+ public static ParallelQuery<TResult> Select<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, int, TResult> selector)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (selector == null)
+ throw new ArgumentNullException ("selector");
+
+ return new ParallelQuery<TResult> (new QuerySelectNode<TResult, TSource> (source.Node, selector));
+ }
+ #endregion
+
+ #region SelectMany
+ public static ParallelQuery<TResult> SelectMany<TSource, TResult> (this ParallelQuery<TSource> source,
+ Func<TSource, IEnumerable<TResult>> selector)
+ {
+ return source.SelectMany (selector, (s, e) => e);
+ }
+
+ public static ParallelQuery<TResult> SelectMany<TSource, TResult> (this ParallelQuery<TSource> source,
+ Func<TSource, int, IEnumerable<TResult>> selector)
+ {
+ return source.SelectMany (selector, (s, e) => e);
+ }
+
+ public static ParallelQuery<TResult> SelectMany<TSource, TCollection, TResult> (this ParallelQuery<TSource> source,
+ Func<TSource, IEnumerable<TCollection>> collectionSelector,
+ Func<TSource, TCollection, TResult> resultSelector)
+ {
+ return new ParallelQuery<TResult> (new QuerySelectManyNode<TSource, TCollection, TResult> (source.Node,
+ collectionSelector,
+ resultSelector));
+ }
+
+ public static ParallelQuery<TResult> SelectMany<TSource, TCollection, TResult> (this ParallelQuery<TSource> source,
+ Func<TSource, int, IEnumerable<TCollection>> collectionSelector,
+ Func<TSource, TCollection, TResult> resultSelector)
+ {
+ return new ParallelQuery<TResult> (new QuerySelectManyNode<TSource, TCollection, TResult> (source.Node,
+ collectionSelector,
+ resultSelector));
+ }
+ #endregion
+
+ #region Where
+ public static ParallelQuery<TSource> Where<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (predicate == null)
+ throw new ArgumentNullException ("predicate");
+
+ return new ParallelQuery<TSource> (new QueryWhereNode<TSource> (source.Node, predicate));
+ }
+
+ public static ParallelQuery<TSource> Where<TSource> (this ParallelQuery<TSource> source, Func<TSource, int, bool> predicate)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (predicate == null)
+ throw new ArgumentNullException ("predicate");
+
+ return new ParallelQuery<TSource> (new QueryWhereNode<TSource> (source.Node, predicate));
+ }
+ #endregion
+
+ #region Aggregate
+ public static TSource Aggregate<TSource> (this ParallelQuery<TSource> source, Func<TSource, TSource, TSource> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Aggregate<TSource, TSource, TSource> ((Func<TSource>)null,
+ func,
+ func,
+ (e) => e);
+ }
+
+ public static TAccumulate Aggregate<TSource, TAccumulate> (this ParallelQuery<TSource> source,
+ TAccumulate seed,
+ Func<TAccumulate, TSource, TAccumulate> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Aggregate (seed, func, (e) => e);
+ }
+
+ public static TResult Aggregate<TSource, TAccumulate, TResult> (this ParallelQuery<TSource> source,
+ TAccumulate seed,
+ Func<TAccumulate, TSource, TAccumulate> func,
+ Func<TAccumulate, TResult> resultSelector)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+ if (resultSelector == null)
+ throw new ArgumentNullException ("resultSelector");
+
+ TAccumulate accumulator = seed;
+
+ foreach (TSource value in source)
+ accumulator = func (accumulator, value);
+
+ return resultSelector (accumulator);
+ }
+
+ public static TResult Aggregate<TSource, TAccumulate, TResult> (this ParallelQuery<TSource> source,
+ TAccumulate seed,
+ Func<TAccumulate, TSource, TAccumulate> updateAccumulatorFunc,
+ Func<TAccumulate, TAccumulate, TAccumulate> combineAccumulatorsFunc,
+ Func<TAccumulate, TResult> resultSelector)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (updateAccumulatorFunc == null)
+ throw new ArgumentNullException ("updateAccumulatorFunc");
+ if (combineAccumulatorsFunc == null)
+ throw new ArgumentNullException ("combineAccumulatorsFunc");
+ if (resultSelector == null)
+ throw new ArgumentNullException ("resultSelector");
+
+ return source.Aggregate (() => seed, updateAccumulatorFunc, combineAccumulatorsFunc, resultSelector);
+ }
+
+ public static TResult Aggregate<TSource, TAccumulate, TResult> (this ParallelQuery<TSource> source,
+ Func<TAccumulate> seedFunc,
+ Func<TAccumulate, TSource, TAccumulate> updateAccumulatorFunc,
+ Func<TAccumulate, TAccumulate, TAccumulate> combineAccumulatorsFunc,
+ Func<TAccumulate, TResult> resultSelector)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (seedFunc == null)
+ throw new ArgumentNullException ("seedFunc");
+ if (updateAccumulatorFunc == null)
+ throw new ArgumentNullException ("updateAccumulatorFunc");
+ if (combineAccumulatorsFunc == null)
+ throw new ArgumentNullException ("combineAccumulatorsFunc");
+ if (resultSelector == null)
+ throw new ArgumentNullException ("resultSelector");
+
+ TAccumulate accumulator = default (TAccumulate);
+
+ ParallelExecuter.ProcessAndAggregate<TSource, TAccumulate> (source.Node, seedFunc, updateAccumulatorFunc, (list) => {
+ accumulator = list [0];
+ for (int i = 1; i < list.Count; i++)
+ accumulator = combineAccumulatorsFunc (accumulator, list[i]);
+ });
+
+ return resultSelector (accumulator);;
+ }
+ #endregion
+
+ #region ForAll
+ public static void ForAll<TSource> (this ParallelQuery<TSource> source, Action<TSource> action)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (action == null)
+ throw new ArgumentNullException ("action");
+
+ ParallelExecuter.ProcessAndBlock (source.Node, action);
+ }
+ #endregion
+
+ #region OrderBy
+ public static OrderedParallelQuery<TSource> OrderByDescending<TSource, TKey> (this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ IComparer<TKey> comparer)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (keySelector == null)
+ throw new ArgumentNullException ("keySelector");
+ if (comparer == null)
+ throw new ArgumentNullException ("comparer");
+
+ Comparison<TSource> comparison = (e1, e2) => -comparer.Compare (keySelector (e1), keySelector (e2));
+
+ return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
+ }
+
+ public static OrderedParallelQuery<TSource> OrderByDescending<TSource, TKey> (this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector)
+ {
+ return OrderByDescending (source, keySelector, Comparer<TKey>.Default);
+ }
+
+ public static OrderedParallelQuery<TSource> OrderBy<TSource, TKey> (this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector)
+ {
+ return OrderBy (source, keySelector, Comparer<TKey>.Default);
+ }
+
+ public static OrderedParallelQuery<TSource> OrderBy<TSource, TKey> (this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ IComparer<TKey> comparer)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (keySelector == null)
+ throw new ArgumentNullException ("keySelector");
+ if (comparer == null)
+ throw new ArgumentNullException ("comparer");
+
+ Comparison<TSource> comparison = (e1, e2) => comparer.Compare (keySelector (e1), keySelector (e2));
+
+ return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
+ }
+ #endregion
+
+ #region ThenBy
+ public static OrderedParallelQuery<TSource> ThenBy<TSource, TKey> (this OrderedParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector)
+ {
+ return ThenBy (source, keySelector, Comparer<TKey>.Default);
+ }
+
+ public static OrderedParallelQuery<TSource> ThenBy<TSource, TKey> (this OrderedParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ IComparer<TKey> comparer)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (keySelector == null)
+ throw new ArgumentNullException ("keySelector");
+ if (comparer == null)
+ throw new ArgumentNullException ("comparer");
+
+ Comparison<TSource> comparison = (e1, e2) => comparer.Compare (keySelector (e1), keySelector (e2));
+
+ return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
+ }
+
+ public static OrderedParallelQuery<TSource> ThenByDescending<TSource, TKey> (this OrderedParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector)
+ {
+ return ThenByDescending (source, keySelector, Comparer<TKey>.Default);
+ }
+
+ public static OrderedParallelQuery<TSource> ThenByDescending<TSource, TKey> (this OrderedParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ IComparer<TKey> comparer)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (keySelector == null)
+ throw new ArgumentNullException ("keySelector");
+ if (comparer == null)
+ throw new ArgumentNullException ("comparer");
+
+ Comparison<TSource> comparison = (e1, e2) => -comparer.Compare (keySelector (e1), keySelector (e2));
+
+ return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
+ }
+ #endregion
+
+ #region All
+ public static bool All<TSource> (ParallelQuery<TSource> source, Func<TSource, bool> predicate)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (predicate == null)
+ throw new ArgumentNullException ("predicate");
+
+ CancellationTokenSource src = new CancellationTokenSource ();
+ ParallelQuery<TSource> innerQuery = source.WithImplementerToken (src);
+
+ bool result = true;
+ innerQuery.ForAll ((e) => {
+ if (!predicate (e)) {
+ result = false;
+ src.Cancel ();
+ }
+ });
+
+ return result;
+ }
+ #endregion
+
+ #region Any
+ public static bool Any<TSource> (this ParallelQuery<TSource> source)
+ {
+ return Any<TSource> (source, (_) => true);
+ }
+
+ public static bool Any<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (predicate == null)
+ throw new ArgumentNullException ("predicate");
+
+ return !source.All ((e) => !predicate (e));
+ }
+ #endregion
+
+ #region Contains
+ public static bool Contains<TSource> (this ParallelQuery<TSource> source, TSource value)
+ {
+ return Contains<TSource> (source, value, EqualityComparer<TSource>.Default);
+ }
+
+ public static bool Contains<TSource> (this ParallelQuery<TSource> source, TSource value, IEqualityComparer<TSource> comparer)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (comparer == null)
+ throw new ArgumentNullException ("comparer");
+
+ return Any<TSource> (source, (e) => comparer.Equals (value));
+ }
+ #endregion
+
+ #region SequenceEqual
+ public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first,
+ ParallelQuery<TSource> second)
+ {
+ if (first == null)
+ throw new ArgumentNullException ("first");
+ if (second == null)
+ throw new ArgumentNullException ("second");
+
+ return first.SequenceEqual (second, EqualityComparer<TSource>.Default);
+ }
+
+ public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first,
+ ParallelQuery<TSource> second,
+ IEqualityComparer<TSource> comparer)
+ {
+ if (first == null)
+ throw new ArgumentNullException ("first");
+ if (second == null)
+ throw new ArgumentNullException ("second");
+ if (comparer == null)
+ throw new ArgumentNullException ("comparer");
+
+ CancellationTokenSource source = new CancellationTokenSource ();
+ ParallelQuery<bool> innerQuery
+ = first.Zip (second, (e1, e2) => comparer.Equals (e1, e2)).Where ((e) => !e).WithImplementerToken (source);
+
+ bool result = true;
+
+ innerQuery.ForAll ((value) => {
+ result = false;
+ source.Cancel ();
+ });
+
+ return result;
+ }
+
+ [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
+ + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
+ + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
+ public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first, IEnumerable<TSource> second)
+ {
+ throw new NotSupportedException ();
+ }
+
+ [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
+ + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
+ + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
+ public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first,
+ IEnumerable<TSource> second,
+ IEqualityComparer<TSource> comparer)
+ {
+ throw new NotSupportedException ();
+ }
+
+ #endregion
+
+ #region GroupBy
+ public static ParallelQuery<IGrouping<TKey, TSource>> GroupBy<TSource, TKey> ( this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector)
+ {
+ return source.GroupBy (keySelector, EqualityComparer<TKey>.Default);
+ }
+
+ public static ParallelQuery<IGrouping<TKey, TSource>> GroupBy<TSource, TKey> ( this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ IEqualityComparer<TKey> comparer)
+ {
+ return source.GroupBy (keySelector, (e) => e, comparer);
+ }
+
+ public static ParallelQuery<IGrouping<TKey, TElement>> GroupBy<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ Func<TSource, TElement> elementSelector)
+ {
+ return source.GroupBy (keySelector, elementSelector, EqualityComparer<TKey>.Default);
+ }
+
+ public static ParallelQuery<TResult> GroupBy<TSource, TKey, TResult> (this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ Func<TKey, IEnumerable<TSource>, TResult> resultSelector)
+ {
+ return source.GroupBy (keySelector)
+ .Select ((g) => resultSelector (g.Key, (IEnumerable<TSource>)g));
+ }
+
+ public static ParallelQuery<IGrouping<TKey, TElement>> GroupBy<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ Func<TSource, TElement> elementSelector,
+ IEqualityComparer<TKey> comparer)
+ {
+ throw new NotImplementedException ();
+ }
+
+ public static ParallelQuery<TResult> GroupBy<TSource, TKey, TElement, TResult> (this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ Func<TSource, TElement> elementSelector,
+ Func<TKey, IEnumerable<TElement>, TResult> resultSelector)
+ {
+ return source.GroupBy (keySelector, elementSelector)
+ .Select ((g) => resultSelector (g.Key, (IEnumerable<TElement>)g));
+ }
+
+ public static ParallelQuery<TResult> GroupBy<TSource, TKey, TResult> ( this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ Func<TKey, IEnumerable<TSource>, TResult> resultSelector,
+ IEqualityComparer<TKey> comparer)
+ {
+ return source.GroupBy (keySelector, comparer)
+ .Select ((g) => resultSelector (g.Key, (IEnumerable<TSource>)g));
+ }
+
+ public static ParallelQuery<TResult> GroupBy<TSource, TKey, TElement, TResult> (this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ Func<TSource, TElement> elementSelector,
+ Func<TKey, IEnumerable<TElement>, TResult> resultSelector,
+ IEqualityComparer<TKey> comparer)
+ {
+ return source.GroupBy (keySelector, elementSelector, comparer)
+ .Select ((g) => resultSelector (g.Key, (IEnumerable<TElement>)g));
+ }
+ #endregion
+
+ #region GroupJoin
+ public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
+ ParallelQuery<TInner> inner,
+ Func<TOuter, TKey> outerKeySelector,
+ Func<TInner, TKey> innerKeySelector,
+ Func<TOuter, IEnumerable<TInner>, TResult> resultSelector)
+ {
+ return outer.GroupJoin (inner, outerKeySelector, innerKeySelector, resultSelector, EqualityComparer<TKey>.Default);
+ }
+
+ public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
+ ParallelQuery<TInner> inner,
+ Func<TOuter, TKey> outerKeySelector,
+ Func<TInner, TKey> innerKeySelector,
+ Func<TOuter, IEnumerable<TInner>, TResult> resultSelector,
+ IEqualityComparer<TKey> comparer)
+ {
+ throw new NotImplementedException ();
+ }
+
+ [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
+ + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
+ + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
+ public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
+ IEnumerable<TInner> inner,
+ Func<TOuter, TKey> outerKeySelector,
+ Func<TInner, TKey> innerKeySelector,
+ Func<TOuter, IEnumerable<TInner>, TResult> resultSelector)
+ {
+ throw new NotSupportedException ();
+ }
+
+ [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
+ + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
+ + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
+ public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
+ IEnumerable<TInner> inner,
+ Func<TOuter, TKey> outerKeySelector,
+ Func<TInner, TKey> innerKeySelector,
+ Func<TOuter, IEnumerable<TInner>, TResult> resultSelector,
+ IEqualityComparer<TKey> comparer)
+ {
+ throw new NotImplementedException ();
+ }
+ #endregion
+
+ #region ElementAt
+ public static TSource ElementAt<TSource> ( this ParallelQuery<TSource> source, int index)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (index < 0)
+ throw new ArgumentOutOfRangeException ("index");
+ if (index == 0) {
+ try {
+ return source.First ();
+ } catch (InvalidOperationException) {
+ throw new ArgumentOutOfRangeException ("index");
+ }
+ }
+
+ TSource result = default (TSource);
+
+ ParallelQuery<TSource> innerQuery = source.Where ((e, i) => i == index);
+
+ try {
+ result = innerQuery.First ();
+ } catch (InvalidOperationException) {
+ throw new ArgumentOutOfRangeException ("index");
+ }
+
+ return result;
+ }
+
+ public static TSource ElementAtOrDefault<TSource> (this ParallelQuery<TSource> source, int index)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ try {
+ return source.ElementAt (index);
+ } catch (ArgumentOutOfRangeException) {
+ return default (TSource);
+ }
+ }
+ #endregion
+
+ #region Intersect
+ public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first,
+ ParallelQuery<TSource> second)
+ {
+ return Intersect<TSource> (first, second, EqualityComparer<TSource>.Default);
+ }
+
+ public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first,
+ ParallelQuery<TSource> second,
+ IEqualityComparer<TSource> comparer)
+ {
+ if (first == null)
+ throw new ArgumentNullException ("first");
+ if (second == null)
+ throw new ArgumentNullException ("second");
+ if (comparer == null)
+ throw new ArgumentNullException ("comparer");
+
+ return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Intersect, comparer, first.Node, second.Node));
+ }
+
+ [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
+ + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
+ + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
+ public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first, IEnumerable<TSource> second)
+ {
+ throw new NotSupportedException ();
+ }
+
+ [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
+ + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
+ + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
+ public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first,
+ IEnumerable<TSource> second,
+ IEqualityComparer<TSource> comparer)
+ {
+ throw new NotSupportedException ();
+ }
+ #endregion
+
+ #region Join
+ public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> ( this ParallelQuery<TOuter> outer,
+ ParallelQuery<TInner> inner,
+ Func<TOuter, TKey> outerKeySelector,
+ Func<TInner, TKey> innerKeySelector,
+ Func<TOuter, TInner, TResult> resultSelector)
+ {
+ return outer.Join (inner, outerKeySelector, innerKeySelector, resultSelector, EqualityComparer<TKey>.Default);
+ }
+
+ public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> ( this ParallelQuery<TOuter> outer,
+ ParallelQuery<TInner> inner,
+ Func<TOuter, TKey> outerKeySelector,
+ Func<TInner, TKey> innerKeySelector,
+ Func<TOuter, TInner, TResult> resultSelector,
+ IEqualityComparer<TKey> comparer)
+ {
+ throw new NotImplementedException ();
+ }
+
+ [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
+ + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
+ + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
+ public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> ( this ParallelQuery<TOuter> outer,
+ IEnumerable<TInner> inner,
+ Func<TOuter, TKey> outerKeySelector,
+ Func<TInner, TKey> innerKeySelector,
+ Func<TOuter, TInner, TResult> resultSelector)
+ {
+ throw new NotSupportedException ();
+ }
+
+ [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
+ + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
+ + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
+ public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
+ IEnumerable<TInner> inner,
+ Func<TOuter, TKey> outerKeySelector,
+ Func<TInner, TKey> innerKeySelector,
+ Func<TOuter, TInner, TResult> resultSelector,
+ IEqualityComparer<TKey> comparer)
+ {
+ throw new NotSupportedException ();
+ }
+ #endregion
+
+ #region Except
+ public static ParallelQuery<TSource> Except<TSource> ( this ParallelQuery<TSource> first,
+ ParallelQuery<TSource> second)
+ {
+ return Except<TSource> (first, second, EqualityComparer<TSource>.Default);
+ }
+
+ public static ParallelQuery<TSource> Except<TSource> ( this ParallelQuery<TSource> first,
+ ParallelQuery<TSource> second,
+ IEqualityComparer<TSource> comparer)
+ {
+ if (first == null)
+ throw new ArgumentNullException ("first");
+ if (second == null)
+ throw new ArgumentNullException ("second");
+ if (comparer == null)
+ throw new ArgumentNullException ("comparer");
+
+ return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Except,
+ comparer, first.Node, second.Node));
+ }
+
+ [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
+ + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
+ + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
+ public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
+ IEnumerable<TSource> second)
+ {
+ throw new NotSupportedException ();
+ }
+
+ [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
+ + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
+ + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
+ public static ParallelQuery<TSource> Except<TSource> ( this ParallelQuery<TSource> first,
+ IEnumerable<TSource> second,
+ IEqualityComparer<TSource> comparer)
+ {
+ throw new NotSupportedException ();
+ }
+ #endregion
+
+ #region Distinct
+ public static ParallelQuery<TSource> Distinct<TSource> (this ParallelQuery<TSource> source)
+ {
+ return Distinct<TSource> (source, EqualityComparer<TSource>.Default);
+ }
+
+ public static ParallelQuery<TSource> Distinct<TSource> (this ParallelQuery<TSource> source, IEqualityComparer<TSource> comparer)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (comparer == null)
+ throw new ArgumentNullException ("comparer");
+
+ return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Distinct, comparer,
+ source.Node, null));
+ }
+ #endregion
+
+ #region Union
+ [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
+ + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
+ + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
+ public static ParallelQuery<TSource> Union<TSource> (this ParallelQuery<TSource> first,
+ IEnumerable<TSource> second)
+ {
+ throw new NotSupportedException ();
+ }
+
+ [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
+ + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
+ + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
+ public static ParallelQuery<TSource> Union<TSource>(this ParallelQuery<TSource> first,
+ IEnumerable<TSource> second,
+ IEqualityComparer<TSource> comparer)
+ {
+ throw new NotSupportedException ();
+ }
+
+ public static ParallelQuery<TSource> Union<TSource> (this ParallelQuery<TSource> first,
+ ParallelQuery<TSource> second)
+ {
+ return first.Union (second, EqualityComparer<TSource>.Default);
+ }
+
+ public static ParallelQuery<TSource> Union<TSource> (this ParallelQuery<TSource> first,
+ ParallelQuery<TSource> second,
+ IEqualityComparer<TSource> comparer)
+ {
+ if (first == null)
+ throw new ArgumentNullException ("first");
+ if (second == null)
+ throw new ArgumentNullException ("second");
+ if (comparer == null)
+ throw new ArgumentNullException ("comparer");
+
+ return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Union, comparer, first.Node, second.Node));
+ }
+ #endregion
+
+ #region Take
+ // TODO : introduce some early break up here, use ImplementerToken
+ public static ParallelQuery<TSource> Take<TSource> (this ParallelQuery<TSource> source, int count)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Where ((e, i) => i < count);
+ }
+
+ public static ParallelQuery<TSource> TakeWhile<TSource> (this ParallelQuery<TSource> source,
+ Func<TSource, bool> predicate)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (predicate == null)
+ throw new ArgumentNullException ("predicate");
+
+ return source.Where ((e) => predicate (e));
+ }
+
+ public static ParallelQuery<TSource> TakeWhile<TSource> (this ParallelQuery<TSource> source,
+ Func<TSource, int, bool> predicate)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (predicate == null)
+ throw new ArgumentNullException ("predicate");
+
+ return source.Where ((e, i) => predicate (e, i));
+ }
+ #endregion
+
+ #region Skip
+ public static ParallelQuery<TSource> Skip<TSource> (this ParallelQuery<TSource> source, int count)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Where ((e, i) => i >= count);
+ }
+
+ public static ParallelQuery<TSource> SkipWhile<TSource> (this ParallelQuery<TSource> source,
+ Func<TSource, bool> predicate)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (predicate == null)
+ throw new ArgumentNullException ("predicate");
+
+ return source.Where ((e) => !predicate (e));
+ }
+
+ public static ParallelQuery<TSource> SkipWhile<TSource> (this ParallelQuery<TSource> source,
+ Func<TSource, int, bool> predicate)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (predicate == null)
+ throw new ArgumentNullException ("predicate");
+
+ return source.Where ((e, i) => !predicate (e, i));
+ }
+ #endregion
+
+ #region Single
+ static TSource SingleInternal<TSource> (this ParallelQuery<TSource> source, params TSource[] init)
+ {
+ TSource result = default(TSource);
+ bool hasValue = false;
+
+ foreach (TSource element in source) {
+ if (hasValue)
+ throw new InvalidOperationException ("The input sequence contains more than one element.");
+
+ result = element;
+ hasValue = true;
+ }
+
+ if (!hasValue && init.Length != 0) {
+ result = init[0];
+ hasValue = true;
+ }
+
+ if (!hasValue)
+ throw new InvalidOperationException ("The input sequence is empty.");
+
+ return result;
+ }
+
+ public static TSource Single<TSource> (this ParallelQuery<TSource> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return SingleInternal<TSource> (source);
+ }
+
+ public static TSource Single<TSource> (this ParallelQuery<TSource> source,
+ Func<TSource, bool> predicate)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (predicate == null)
+ throw new ArgumentNullException ("predicate");
+
+ return source.Where (predicate).Single ();
+ }
+
+ public static TSource SingleOrDefault<TSource> (this ParallelQuery<TSource> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return SingleInternal<TSource> (source, default (TSource));
+ }
+
+ public static TSource SingleOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (predicate == null)
+ throw new ArgumentNullException ("predicate");
+
+ return source.Where (predicate).SingleOrDefault ();
+ }
+ #endregion
+
+ #region Count
+ public static int Count<TSource> (this ParallelQuery<TSource> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Aggregate<TSource, int, int> (() => 0,
+ (acc, e) => acc + 1,
+ (acc1, acc2) => acc1 + acc2,
+ (result) => result);
+ }
+
+ public static int Count<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (predicate == null)
+ throw new ArgumentNullException ("predicate");
+
+ return source.Where (predicate).Count ();
+ }
+
+ public static long LongCount<TSource> (this ParallelQuery<TSource> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Aggregate<TSource, long, long> (() => 0,
+ (acc, e) => acc + 1,
+ (acc1, acc2) => acc1 + acc2,
+ (result) => result);
+ }
+
+ public static long LongCount<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (predicate == null)
+ throw new ArgumentNullException ("predicate");
+
+ return source.Where (predicate).LongCount ();
+ }
+ #endregion
+
+ #region Average
+ public static double Average (this ParallelQuery<int> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Aggregate (() => new int[2],
+ (acc, e) => { acc[0] += e; acc[1]++; return acc; },
+ (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
+ (acc) => acc[0] / ((double)acc[1]));
+ }
+
+ public static double Average (this ParallelQuery<long> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Aggregate (() => new long[2],
+ (acc, e) => { acc[0] += e; acc[1]++; return acc; },
+ (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
+ (acc) => acc[0] / ((double)acc[1]));
+ }
+
+ public static decimal Average (this ParallelQuery<decimal> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Aggregate (() => new decimal[2],
+ (acc, e) => { acc[0] += e; acc[1]++; return acc; },
+ (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
+ (acc) => acc[0] / acc[1]);
+ }
+
+ public static double Average (this ParallelQuery<double> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Aggregate (() => new double[2],
+ (acc, e) => { acc[0] += e; acc[1]++; return acc; },
+ (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
+ (acc) => acc[0] / ((double)acc[1]));
+ }
+
+ public static float Average (this ParallelQuery<float> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Aggregate (() => new float[2],
+ (acc, e) => { acc[0] += e; acc[1]++; return acc; },
+ (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
+ (acc) => acc[0] / acc[1]);
+ }
+ #endregion
+
+ #region More Average
+ public static double? Average (this ParallelQuery<int?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();;
+ }
+
+ public static double? Average (this ParallelQuery<long?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
+ }
+
+ public static decimal? Average (this ParallelQuery<decimal?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
+ }
+
+ public static double? Average (this ParallelQuery<double?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
+ }
+
+ public static float? Average (this ParallelQuery<float?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
+ }
+
+ public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Average ();
+ }
+
+ public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Average ();
+ }
+
+ public static float Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Average ();
+ }
+
+ public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Average ();
+ }
+
+ public static decimal Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Average ();
+ }
+
+ public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Average ();
+ }
+
+ public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Average ();
+ }
+
+ public static float? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Average ();
+ }
+
+ public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Average ();
+ }
+
+ public static decimal? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Average ();
+ }
+ #endregion
+
+ #region Sum
+ public static int Sum (ParallelQuery<int> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Aggregate (0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
+ }
+
+ public static long Sum (ParallelQuery<long> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Aggregate ((long)0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
+ }
+
+ public static float Sum (ParallelQuery<float> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Aggregate (0.0f, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
+ }
+
+ public static double Sum (ParallelQuery<double> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Aggregate (0.0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
+ }
+
+ public static decimal Sum (ParallelQuery<decimal> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Aggregate ((decimal)0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
+ }
+
+ public static int? Sum (ParallelQuery<int?> source)
+ {
+ return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
+ }
+
+ public static long? Sum (ParallelQuery<long?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
+ }
+
+ public static float? Sum (ParallelQuery<float?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
+ }
+
+ public static double? Sum (ParallelQuery<double?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
+ }
+
+ public static decimal? Sum (ParallelQuery<decimal?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
+ }
+
+ public static int Sum<TSource> (ParallelQuery<TSource> source, Func<TSource, int> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Sum ();
+ }
+
+ public static long Sum<TSource> (ParallelQuery<TSource> source, Func<TSource, long> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Sum ();
+ }
+
+ public static decimal Sum<TSource> (ParallelQuery<TSource> source, Func<TSource, decimal> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Sum ();
+ }
+
+ public static float Sum<TSource> (ParallelQuery<TSource> source, Func<TSource, float> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Sum ();
+ }
+
+ public static double Sum<TSource> (ParallelQuery<TSource> source, Func<TSource, double> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Sum ();
+ }
+
+ public static int? Sum<TSource> (ParallelQuery<TSource> source, Func<TSource, int?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Sum ();
+ }
+
+ public static long? Sum<TSource> (ParallelQuery<TSource> source, Func<TSource, long?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Sum ();
+ }
+
+ public static decimal? Sum<TSource> (ParallelQuery<TSource> source, Func<TSource, decimal?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Sum ();
+ }
+
+ public static float? Sum<TSource> (ParallelQuery<TSource> source, Func<TSource, float?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Sum ();
+ }
+
+ public static double? Sum<TSource> (ParallelQuery<TSource> source, Func<TSource, double?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Sum ();
+ }
+ #endregion
+
+ #region Min-Max
+ static T BestOrder<T> (ParallelQuery<T> source, Func<T, T, bool> bestSelector, T seed)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ T best = seed;
+
+ best = source.Aggregate (() => seed,
+ (first, second) => (bestSelector(first, second)) ? first : second,
+ (first, second) => (bestSelector(first, second)) ? first : second,
+ (e) => e);
+ return best;
+ }
+
+ public static int Min (this ParallelQuery<int> source)
+ {
+ return BestOrder (source, (first, second) => first < second, int.MaxValue);
+ }
+
+ public static long Min (this ParallelQuery<long> source)
+ {
+ return BestOrder (source, (first, second) => first < second, long.MaxValue);
+ }
+
+ public static float Min (this ParallelQuery<float> source)
+ {
+ return BestOrder (source, (first, second) => first < second, float.MaxValue);
+ }
+
+ public static double Min (this ParallelQuery<double> source)
+ {
+ return BestOrder (source, (first, second) => first < second, double.MaxValue);
+ }
+
+ public static decimal Min (this ParallelQuery<decimal> source)
+ {
+ return BestOrder (source, (first, second) => first < second, decimal.MaxValue);
+ }
+
+ public static TSource Min<TSource> (this ParallelQuery<TSource> source)
+ {
+ IComparer<TSource> comparer = Comparer<TSource>.Default;
+
+ return BestOrder (source, (first, second) => comparer.Compare (first, second) < 0, default (TSource));
+ }
+
+ public static TResult Min<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Min ();
+ }
+
+ public static int? Min (this ParallelQuery<int?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : int.MaxValue).Min ();
+ }
+
+ public static long? Min (this ParallelQuery<long?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : long.MaxValue).Min ();
+ }
+
+ public static float? Min (this ParallelQuery<float?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : float.MaxValue).Min ();
+ }
+
+ public static double? Min (this ParallelQuery<double?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : double.MaxValue).Min ();
+ }
+
+ public static decimal? Min (this ParallelQuery<decimal?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : decimal.MaxValue).Min ();
+ }
+
+ public static int Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Min ();
+ }
+
+ public static long Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Min ();
+ }
+
+ public static float Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Min ();
+ }
+
+ public static double Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Min ();
+ }
+
+ public static decimal Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Min ();
+ }
+
+ public static int? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Min ();
+ }
+
+ public static long? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Min ();
+ }
+
+ public static float? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Min ();
+ }
+
+ public static double? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Min ();
+ }
+
+ public static decimal? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Min ();
+ }
+
+ public static int Max (this ParallelQuery<int> source)
+ {
+ return BestOrder (source, (first, second) => first > second, int.MinValue);
+ }
+
+ public static long Max(this ParallelQuery<long> source)
+ {
+ return BestOrder(source, (first, second) => first > second, long.MinValue);
+ }
+
+ public static float Max (this ParallelQuery<float> source)
+ {
+ return BestOrder(source, (first, second) => first > second, float.MinValue);
+ }
+
+ public static double Max (this ParallelQuery<double> source)
+ {
+ return BestOrder(source, (first, second) => first > second, double.MinValue);
+ }
+
+ public static decimal Max (this ParallelQuery<decimal> source)
+ {
+ return BestOrder(source, (first, second) => first > second, decimal.MinValue);
+ }
+
+ public static TSource Max<TSource> (this ParallelQuery<TSource> source)
+ {
+ IComparer<TSource> comparer = Comparer<TSource>.Default;
+
+ return BestOrder (source, (first, second) => comparer.Compare (first, second) > 0, default (TSource));
+ }
+
+ public static TResult Max<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Max ();
+ }
+
+ public static int? Max (this ParallelQuery<int?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : int.MinValue).Max ();
+ }
+
+ public static long? Max (this ParallelQuery<long?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : long.MinValue).Max ();
+ }
+
+ public static float? Max (this ParallelQuery<float?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : float.MinValue).Max ();
+ }
+
+ public static double? Max (this ParallelQuery<double?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : double.MinValue).Max ();
+ }
+
+ public static decimal? Max (this ParallelQuery<decimal?> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.Select ((e) => e.HasValue ? e.Value : decimal.MinValue).Max ();
+ }
+
+ public static int Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Max ();
+ }
+
+ public static long Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Max ();
+ }
+
+ public static float Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Max ();
+ }
+
+ public static double Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Max ();
+ }
+
+ public static decimal Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Max ();
+ }
+
+ public static int? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Max ();
+ }
+
+ public static long? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Max ();
+ }
+
+ public static float? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Max ();
+ }
+
+ public static double? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Max ();
+ }
+
+ public static decimal? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> func)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (func == null)
+ throw new ArgumentNullException ("func");
+
+ return source.Select (func).Max ();
+ }
+ #endregion
+
+ #region Cast / OfType
+ public static ParallelQuery<TResult> Cast<TResult> (this ParallelQuery source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.TypedQuery.Select ((e) => (TResult)e);
+ }
+
+ public static ParallelQuery<TResult> OfType<TResult> (ParallelQuery source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return source.TypedQuery.Where ((e) => e is TResult).Cast<TResult> ();
+ }
+ #endregion
+
+ #region Reverse
+ public static ParallelQuery<TSource> Reverse<TSource> (this ParallelQuery<TSource> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ return new ParallelQuery<TSource> (new QueryReverseNode<TSource> (source));
+ }
+ #endregion
+
+ #region ToArray - ToList - ToDictionary - ToLookup
+ public static List<TSource> ToList<TSource> (this ParallelQuery<TSource> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ OrderedParallelQuery<TSource> ordered = null;
+ if ((ordered = source as OrderedParallelQuery<TSource>) != null)
+ return ToListOrdered (ordered);
+
+ List<TSource> temp = source.Aggregate (() => new List<TSource>(50),
+ (list, e) => { list.Add (e); return list; },
+ (list, list2) => { list.AddRange (list2); return list; },
+ (list) => list);
+ return temp;
+ }
+
+ static List<TSource> ToListOrdered<TSource> (this OrderedParallelQuery<TSource> source)
+ {
+ List<TSource> result = new List<TSource> ();
+
+ foreach (TSource element in source)
+ result.Add (element);
+
+ return result;
+ }
+
+ public static TSource[] ToArray<TSource> (this ParallelQuery<TSource> source)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+
+ TSource[] result = null;
+
+ Func<List<TSource>, TSource, List<TSource>> intermediate = (list, e) => {
+ list.Add (e); return list;
+ };
+
+ Action<IList<List<TSource>>> final = (list) => {
+ int count = 0;
+
+ for (int i = 0; i < list.Count; i++)
+ count += list[i].Count;
+
+ result = new TSource[count];
+ int insertIndex = -1;
+
+ for (int i = 0; i < list.Count; i++)
+ for (int j = 0; j < list[i].Count; j++)
+ result [++insertIndex] = list[i][j];
+ };
+
+ ParallelExecuter.ProcessAndAggregate<TSource, List<TSource>> (source.Node,
+ () => new List<TSource> (),
+ intermediate,
+ final);
+
+ return result;
+ }
+
+ public static Dictionary<TKey, TSource> ToDictionary<TSource, TKey> (this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ IEqualityComparer<TKey> comparer)
+ {
+ return ToDictionary<TSource, TKey, TSource> (source, keySelector, (e) => e, comparer);
+ }
+
+ public static Dictionary<TKey, TSource> ToDictionary<TSource, TKey> (this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector)
+ {
+ return ToDictionary<TSource, TKey, TSource> (source, keySelector, (e) => e, EqualityComparer<TKey>.Default);
+ }
+
+ public static Dictionary<TKey, TElement> ToDictionary<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ Func<TSource, TElement> elementSelector)
+ {
+ return ToDictionary<TSource, TKey, TElement> (source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
+ }
+
+ public static Dictionary<TKey, TElement> ToDictionary<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ Func<TSource, TElement> elementSelector,
+ IEqualityComparer<TKey> comparer)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (keySelector == null)
+ throw new ArgumentNullException ("keySelector");
+ if (comparer == null)
+ throw new ArgumentNullException ("comparer");
+ if (elementSelector == null)
+ throw new ArgumentNullException ("elementSelector");
+
+ return source.Aggregate (() => new Dictionary<TKey, TElement> (comparer),
+ (d, e) => { d.Add (keySelector (e), elementSelector (e)); return d; },
+ (d1, d2) => { foreach (var couple in d2) d1.Add (couple.Key, couple.Value); return d1; },
+ (d) => d);
+ }
+
+ public static ILookup<TKey, TSource> ToLookup<TSource, TKey> ( this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector)
+ {
+ return ToLookup<TSource, TKey, TSource> (source, keySelector, (e) => e, EqualityComparer<TKey>.Default);
+ }
+
+ public static ILookup<TKey, TSource> ToLookup<TSource, TKey> ( this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ IEqualityComparer<TKey> comparer)
+ {
+ return ToLookup<TSource, TKey, TSource> (source, keySelector, (e) => e, comparer);
+ }
+
+ public static ILookup<TKey, TElement> ToLookup<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ Func<TSource, TElement> elementSelector)
+ {
+ return ToLookup<TSource, TKey, TElement> (source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
+ }
+
+ public static ILookup<TKey, TElement> ToLookup<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
+ Func<TSource, TKey> keySelector,
+ Func<TSource, TElement> elementSelector,
+ IEqualityComparer<TKey> comparer)
+ {
+ if (source == null)
+ throw new ArgumentNullException ("source");
+ if (keySelector == null)
+ throw new ArgumentNullException ("keySelector");
+ if (comparer == null)
+ throw new ArgumentNullException ("comparer");
+ if (elementSelector == null)
+ throw new ArgumentNullException ("elementSelector");
+
+ ConcurrentLookup<TKey, TElement> lookup = new ConcurrentLookup<TKey, TElement> (comparer);
+ source.ForAll ((e) => lookup.Add (keySelector (e), elementSelector (e)));
+
+ return lookup;
+ }
+ #endregion
+
+ #region Concat
+ [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather than "
+ + "System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() extension method "
+ + "to convert the right data source to System.Linq.ParallelQuery<T>.")]
+ public static ParallelQuery<TSource> Concat<TSource>(this ParallelQuery<TSource> first,
+ IEnumerable<TSource> second)
+ {
+ throw new NotSupportedException ();
+ }
+
+ public static ParallelQuery<TSource> Concat<TSource> ( this ParallelQuery<TSource> first, ParallelQuery<TSource> second)
+ {
+ return new ParallelQuery<TSource> (new QueryConcatNode<TSource> (first.Node, second.Node));
+ }
+ #endregion
+
+ #region DefaultIfEmpty
+ public static ParallelQuery<TSource> DefaultIfEmpty<TSource> ( this ParallelQuery<TSource> source)
+ {
+ return source.DefaultIfEmpty (default (TSource));
+ }
+
+ public static ParallelQuery<TSource> DefaultIfEmpty<TSource> ( this ParallelQuery<TSource> source, TSource defaultValue)
+ {
+ return new ParallelQuery<TSource> (new QueryDefaultEmptyNode<TSource> (source.Node, defaultValue));
+ }
+ #endregion
+
+ #region First
+ public static TSource First<TSource> ( this ParallelQuery<TSource> source)
+ {
+ CancellationTokenSource src = new CancellationTokenSource ();
+ IEnumerator<TSource> enumerator = source.WithImplementerToken (src).GetEnumerator ();
+
+ if (enumerator == null || !enumerator.MoveNext ())
+ throw new InvalidOperationException ("source contains no element");
+
+ TSource result = enumerator.Current;
+ src.Cancel ();
+
+ return result;
+ }
+
+ public static TSource First<TSource> ( this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
+ {
+ return source.Where (predicate).First ();
+ }
+
+ public static TSource FirstOrDefault<TSource> (this ParallelQuery<TSource> source)
+ {
+ return source.DefaultIfEmpty ().First ();
+ }
+
+ public static TSource FirstOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
+ {
+ return source.Where (predicate).FirstOrDefault ();
+ }
+ #endregion
+
+ #region Last
+ public static TSource Last<TSource> (this ParallelQuery<TSource> source)
+ {
+ return source.Reverse ().First ();
+ }
+
+ public static TSource Last<TSource> ( this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
+ {
+ return source.Reverse ().First (predicate);
+ }
+
+ public static TSource LastOrDefault<TSource> (this ParallelQuery<TSource> source)
+ {
+ return source.Reverse ().FirstOrDefault ();
+ }
+
+ public static TSource LastOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
+ {
+ return source.Reverse ().FirstOrDefault (predicate);
+ }
+ #endregion
+
+ #region Zip
+ public static ParallelQuery<TResult> Zip<TFirst, TSecond, TResult> (this ParallelQuery<TFirst> first,
+ ParallelQuery<TSecond> second,
+ Func<TFirst, TSecond, TResult> resultSelector)
+ {
+ if (first == null)
+ throw new ArgumentNullException ("first");
+ if (second == null)
+ throw new ArgumentNullException ("second");
+ if (resultSelector == null)
+ throw new ArgumentNullException ("resultSelector");
+
+ return new ParallelQuery<TResult> (new QueryZipNode<TFirst, TSecond, TResult> (resultSelector, first.Node, second.Node));
+ }
+
+ [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
+ + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
+ + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
+ public static ParallelQuery<TResult> Zip<TFirst, TSecond, TResult> (this ParallelQuery<TFirst> first,
+ IEnumerable<TSecond> second,
+ Func<TFirst, TSecond, TResult> resultSelector)
+ {
+ throw new NotSupportedException ();
+ }
+ #endregion
+ }
+}
+#endif