#if NET_4_0
//
// ParallelEnumerable.cs
//
// Author:
// Jrmie "Garuma" Laval <jeremie.laval@gmail.com>
//
// Copyright (c) 2010 Jrmie "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 (this 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> (this 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 (this 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 (this 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 (this 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 (this 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 (this 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 (this ParallelQuery<int?> source)
{
return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
}
public static long? Sum (this ParallelQuery<long?> source)
{
if (source == null)
throw new ArgumentNullException ("source");
return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
}
public static float? Sum (this ParallelQuery<float?> source)
{
if (source == null)
throw new ArgumentNullException ("source");
return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
}
public static double? Sum (this ParallelQuery<double?> source)
{
if (source == null)
throw new ArgumentNullException ("source");
return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
}
public static decimal? Sum (this 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> (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).Sum ();
}
public static long Sum<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).Sum ();
}
public static decimal Sum<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).Sum ();
}
public static float Sum<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).Sum ();
}
public static double Sum<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).Sum ();
}
public static int? Sum<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).Sum ();
}
public static long? Sum<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).Sum ();
}
public static decimal? Sum<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).Sum ();
}
public static float? Sum<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).Sum ();
}
public static double? Sum<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).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> (this 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
|