mcs/class/System.Core/System.Linq/ParallelEnumerable.cs

   1 //
   2 // ParallelEnumerable.cs
   3 //
   4 // Author:
   5 //       Jérémie "Garuma" Laval <jeremie.laval@gmail.com>
   6 //
   7 // Copyright (c) 2010 Jérémie "Garuma" Laval
   8 //
   9 // Permission is hereby granted, free of charge, to any person obtaining a copy
  10 // of this software and associated documentation files (the "Software"), to deal
  11 // in the Software without restriction, including without limitation the rights
  12 // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  13 // copies of the Software, and to permit persons to whom the Software is
  14 // furnished to do so, subject to the following conditions:
  15 //
  16 // The above copyright notice and this permission notice shall be included in
  17 // all copies or substantial portions of the Software.
  18 //
  19 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  20 // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  21 // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  22 // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  23 // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  24 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  25 // THE SOFTWARE.
  26
  27 #if NET_4_0
  28 using System;
  29 using System.Threading;
  30 using System.Collections;
  31 using System.Collections.Generic;
  32 using System.Collections.Concurrent;
  33 using System.Linq.Parallel;
  34 using System.Linq.Parallel.QueryNodes;
  35
  36 namespace System.Linq
  37 {
  38         public static class ParallelEnumerable
  39         {
  40                 #region Range & Repeat
  41                 public static ParallelQuery<int> Range (int start, int count)
  42                 {
  43                         if (int.MaxValue - start < count - 1)
  44                                 throw new ArgumentOutOfRangeException ("count", "start + count - 1 is larger than Int32.MaxValue");
  45                         if (count < 0)
  46                                 throw new ArgumentOutOfRangeException ("count", "count is less than 0");
  47
  48                         return (new RangeList (start, count)).AsParallel ();
  49                 }
  50
  51                 public static ParallelQuery<TResult> Repeat<TResult> (TResult element, int count)
  52                 {
  53                         if (count < 0)
  54                                 throw new ArgumentOutOfRangeException ("count", "count is less than 0");
  55
  56                         return (new RepeatList<TResult> (element, count)).AsParallel ();
  57                 }
  58                 #endregion
  59
  60                 #region Empty
  61                 public static ParallelQuery<TResult> Empty<TResult> ()
  62                 {
  63                         return Repeat<TResult> (default (TResult), 0);
  64                 }
  65                 #endregion
  66
  67                 #region AsParallel
  68                 public static ParallelQuery<TSource> AsParallel<TSource> (this IEnumerable<TSource> source)
  69                 {
  70                         if (source == null)
  71                                 throw new ArgumentNullException ("source");
  72
  73                         /* Someone might be trying to use AsParallel a bit too much, if the query was in fact
  74                          * already a ParallelQuery, just cast it
  75                          */
  76                         ParallelQuery<TSource> query = source as ParallelQuery<TSource>;
  77
  78                         return query ?? new ParallelQuery<TSource> (new QueryStartNode<TSource> (source));
  79                 }
  80
  81                 public static ParallelQuery<TSource> AsParallel<TSource> (this Partitioner<TSource> source)
  82                 {
  83                         if (source == null)
  84                                 throw new ArgumentNullException ("source");
  85
  86                         /* Someone might be trying to use AsParallel a bit too much, if the query was in fact
  87                          * already a ParallelQuery, just cast it
  88                          */
  89                         ParallelQuery<TSource> query = source as ParallelQuery<TSource>;
  90
  91                         return query ?? new ParallelQuery<TSource> (new QueryStartNode<TSource> (source));
  92                 }
  93
  94                 public static ParallelQuery AsParallel (this IEnumerable source)
  95                 {
  96                         if (source == null)
  97                                 throw new ArgumentNullException ("source");
  98
  99                         /* Someone might be trying to use AsParallel a bit too much, if the query was in fact
 100                          * already a ParallelQuery, just cast it
 101                          */
 102                         ParallelQuery query = source as ParallelQuery;
 103
 104                         return query ?? new ParallelQuery<object> (new QueryStartNode<object> (source.Cast<object> ()));
 105                 }
 106
 107                 public static IEnumerable<TSource> AsEnumerable<TSource> (this ParallelQuery<TSource> source)
 108                 {
 109                         if (source == null)
 110                                 throw new ArgumentNullException ("source");
 111
 112                         return source.AsSequential ();
 113                 }
 114
 115                 public static IEnumerable<TSource> AsSequential<TSource> (this ParallelQuery<TSource> source)
 116                 {
 117                         if (source == null)
 118                                 throw new ArgumentNullException ("source");
 119
 120                         return source.Node.GetSequential ();
 121                 }
 122                 #endregion
 123
 124                 #region AsOrdered / AsUnordered
 125                 public static ParallelQuery<TSource> AsOrdered<TSource> (this ParallelQuery<TSource> source)
 126                 {
 127                         if (source == null)
 128                                 throw new ArgumentNullException ("source");
 129
 130                         return new ParallelQuery<TSource> (new QueryAsOrderedNode<TSource> (source.Node));
 131                 }
 132
 133                 public static ParallelQuery<TSource> AsUnordered<TSource> (this ParallelQuery<TSource> source)
 134                 {
 135                         if (source == null)
 136                                 throw new ArgumentNullException ("source");
 137
 138                         return new ParallelQuery<TSource> (new QueryAsUnorderedNode<TSource> (source.Node));
 139                 }
 140
 141                 public static ParallelQuery AsOrdered (this ParallelQuery source)
 142                 {
 143                         if (source == null)
 144                                 throw new ArgumentNullException ("source");
 145
 146                         return source.TypedQuery.AsOrdered ();
 147                 }
 148                 #endregion
 149
 150                 #region With*
 151                 public static ParallelQuery<TSource> WithExecutionMode<TSource> (this ParallelQuery<TSource> source,
 152                                                                                  ParallelExecutionMode executionMode)
 153                 {
 154                         if (source == null)
 155                                 throw new ArgumentNullException ("source");
 156
 157                         return new ParallelQuery<TSource> (new ParallelExecutionModeNode<TSource> (executionMode, source.Node));
 158                 }
 159
 160                 public static ParallelQuery<TSource> WithCancellation<TSource> (this ParallelQuery<TSource> source,
 161                                                                                 CancellationToken cancellationToken)
 162                 {
 163                         if (source == null)
 164                                 throw new ArgumentNullException ("source");
 165
 166                         return new ParallelQuery<TSource> (new CancellationTokenNode<TSource> (cancellationToken, source.Node));
 167                 }
 168
 169                 public static ParallelQuery<TSource> WithMergeOptions<TSource> (this ParallelQuery<TSource> source,
 170                                                                                 ParallelMergeOptions mergeOptions)
 171                 {
 172                         if (source == null)
 173                                 throw new ArgumentNullException ("source");
 174
 175                         return new ParallelQuery<TSource> (new ParallelMergeOptionsNode<TSource> (mergeOptions, source.Node));
 176                 }
 177
 178                 public static ParallelQuery<TSource> WithDegreeOfParallelism<TSource> (this ParallelQuery<TSource> source,
 179                                                                                        int degreeOfParallelism)
 180                 {
 181                         if (degreeOfParallelism < 1 || degreeOfParallelism > 63)
 182                                 throw new ArgumentException ("degreeOfParallelism is less than 1 or greater than 63", "degreeOfParallelism");
 183                         if (source == null)
 184                                 throw new ArgumentNullException ("source");
 185
 186                         return new ParallelQuery<TSource> (new DegreeOfParallelismNode<TSource> (degreeOfParallelism, source.Node));
 187                 }
 188
 189                 internal static ParallelQuery<TSource> WithImplementerToken<TSource> (this ParallelQuery<TSource> source,
 190                                                                                       CancellationTokenSource token)
 191                 {
 192                         return new ParallelQuery<TSource> (new ImplementerTokenNode<TSource> (token, source.Node));
 193                 }
 194                 #endregion
 195
 196                 #region Select
 197                 public static ParallelQuery<TResult> Select<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> selector)
 198                 {
 199                         if (source == null)
 200                                 throw new ArgumentNullException ("source");
 201                         if (selector == null)
 202                                 throw new ArgumentNullException ("selector");
 203
 204                         return new ParallelQuery<TResult> (new QuerySelectNode<TResult, TSource> (source.Node, selector));
 205                 }
 206
 207                 public static ParallelQuery<TResult> Select<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, int, TResult> selector)
 208                 {
 209                         if (source == null)
 210                                 throw new ArgumentNullException ("source");
 211                         if (selector == null)
 212                                 throw new ArgumentNullException ("selector");
 213
 214                         return new ParallelQuery<TResult> (new QuerySelectNode<TResult, TSource> (source.Node, selector));
 215                 }
 216                 #endregion
 217
 218                 #region SelectMany
 219                 public static ParallelQuery<TResult> SelectMany<TSource, TResult> (this ParallelQuery<TSource> source,
 220                                                                                    Func<TSource, IEnumerable<TResult>> selector)
 221                 {
 222                         return source.SelectMany (selector, (s, e) => e);
 223                 }
 224
 225                 public static ParallelQuery<TResult> SelectMany<TSource, TResult> (this ParallelQuery<TSource> source,
 226                                                                                    Func<TSource, int, IEnumerable<TResult>> selector)
 227                 {
 228                         return source.SelectMany (selector, (s, e) => e);
 229                 }
 230
 231                 public static ParallelQuery<TResult> SelectMany<TSource, TCollection, TResult> (this ParallelQuery<TSource> source,
 232                                                                                                 Func<TSource, IEnumerable<TCollection>> collectionSelector,
 233                                                                                                 Func<TSource, TCollection, TResult> resultSelector)
 234                 {
 235                         return new ParallelQuery<TResult> (new QuerySelectManyNode<TSource, TCollection, TResult> (source.Node,
 236                                                                                                                    collectionSelector,
 237                                                                                                                    resultSelector));
 238                 }
 239
 240                 public static ParallelQuery<TResult> SelectMany<TSource, TCollection, TResult> (this ParallelQuery<TSource> source,
 241                                                                                                 Func<TSource, int, IEnumerable<TCollection>> collectionSelector,
 242                                                                                                 Func<TSource, TCollection, TResult> resultSelector)
 243                 {
 244                         return new ParallelQuery<TResult> (new QuerySelectManyNode<TSource, TCollection, TResult> (source.Node,
 245                                                                                                                    collectionSelector,
 246                                                                                                                    resultSelector));
 247                 }
 248                 #endregion
 249
 250                 #region Where
 251                 public static ParallelQuery<TSource> Where<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
 252                 {
 253                         if (source == null)
 254                                 throw new ArgumentNullException ("source");
 255                         if (predicate == null)
 256                                 throw new ArgumentNullException ("predicate");
 257
 258                         return new ParallelQuery<TSource> (new QueryWhereNode<TSource> (source.Node, predicate));
 259                 }
 260
 261                 public static ParallelQuery<TSource> Where<TSource> (this ParallelQuery<TSource> source, Func<TSource, int, bool> predicate)
 262                 {
 263                         if (source == null)
 264                                 throw new ArgumentNullException ("source");
 265                         if (predicate == null)
 266                                 throw new ArgumentNullException ("predicate");
 267
 268                         return new ParallelQuery<TSource> (new QueryWhereNode<TSource> (source.Node, predicate));
 269                 }
 270                 #endregion
 271
 272                 #region Aggregate
 273                 public static TSource Aggregate<TSource> (this ParallelQuery<TSource> source, Func<TSource, TSource, TSource> func)
 274                 {
 275                         if (source == null)
 276                                 throw new ArgumentNullException ("source");
 277                         if (func == null)
 278                                 throw new ArgumentNullException ("func");
 279
 280                         return source.Aggregate<TSource, TSource, TSource> ((Func<TSource>)null,
 281                                                                             func,
 282                                                                             func,
 283                                                                             (e) => e);
 284                 }
 285
 286                 public static TAccumulate Aggregate<TSource, TAccumulate> (this ParallelQuery<TSource> source,
 287                                                                              TAccumulate seed,
 288                                                                              Func<TAccumulate, TSource, TAccumulate> func)
 289                 {
 290                         if (source == null)
 291                                 throw new ArgumentNullException ("source");
 292                         if (func == null)
 293                                 throw new ArgumentNullException ("func");
 294
 295                         return source.Aggregate (seed, func, (e) => e);
 296                 }
 297
 298                 public static TResult Aggregate<TSource, TAccumulate, TResult> (this ParallelQuery<TSource> source,
 299                                                                                   TAccumulate seed,
 300                                                                                   Func<TAccumulate, TSource, TAccumulate> func,
 301                                                                                   Func<TAccumulate, TResult> resultSelector)
 302                 {
 303                         if (source == null)
 304                                 throw new ArgumentNullException ("source");
 305                         if (func == null)
 306                                 throw new ArgumentNullException ("func");
 307                         if (resultSelector == null)
 308                                 throw new ArgumentNullException ("resultSelector");
 309
 310                         TAccumulate accumulator = seed;
 311
 312                         foreach (TSource value in source)
 313                                 accumulator = func (accumulator, value);
 314
 315                         return resultSelector (accumulator);
 316                 }
 317
 318                 public static TResult Aggregate<TSource, TAccumulate, TResult> (this ParallelQuery<TSource> source,
 319                                                                                   TAccumulate seed,
 320                                                                                   Func<TAccumulate, TSource, TAccumulate> updateAccumulatorFunc,
 321                                                                                   Func<TAccumulate, TAccumulate, TAccumulate> combineAccumulatorsFunc,
 322                                                                                   Func<TAccumulate, TResult> resultSelector)
 323                 {
 324                         if (source == null)
 325                                 throw new ArgumentNullException ("source");
 326                         if (updateAccumulatorFunc == null)
 327                                 throw new ArgumentNullException ("updateAccumulatorFunc");
 328                         if (combineAccumulatorsFunc == null)
 329                                 throw new ArgumentNullException ("combineAccumulatorsFunc");
 330                         if (resultSelector == null)
 331                                 throw new ArgumentNullException ("resultSelector");
 332
 333                         return source.Aggregate (() => seed, updateAccumulatorFunc, combineAccumulatorsFunc, resultSelector);
 334                 }
 335
 336                 public static TResult Aggregate<TSource, TAccumulate, TResult> (this ParallelQuery<TSource> source,
 337                                                                                   Func<TAccumulate> seedFactory,
 338                                                                                   Func<TAccumulate, TSource, TAccumulate> updateAccumulatorFunc,
 339                                                                                   Func<TAccumulate, TAccumulate, TAccumulate> combineAccumulatorsFunc,
 340                                                                                   Func<TAccumulate, TResult> resultSelector)
 341                 {
 342                         if (source == null)
 343                                 throw new ArgumentNullException ("source");
 344                         if (seedFactory == null)
 345                                 throw new ArgumentNullException ("seedFactory");
 346                         if (updateAccumulatorFunc == null)
 347                                 throw new ArgumentNullException ("updateAccumulatorFunc");
 348                         if (combineAccumulatorsFunc == null)
 349                                 throw new ArgumentNullException ("combineAccumulatorsFunc");
 350                         if (resultSelector == null)
 351                                 throw new ArgumentNullException ("resultSelector");
 352
 353                         TAccumulate accumulator = default (TAccumulate);
 354
 355                         ParallelExecuter.ProcessAndAggregate<TSource, TAccumulate> (source.Node, seedFactory, updateAccumulatorFunc, (list) => {
 356                                 accumulator = list [0];
 357                                 for (int i = 1; i < list.Count; i++)
 358                                         accumulator = combineAccumulatorsFunc (accumulator, list[i]);
 359                         });
 360
 361                         return resultSelector (accumulator);;
 362                 }
 363                 #endregion
 364
 365                 #region ForAll
 366                 public static void ForAll<TSource> (this ParallelQuery<TSource> source, Action<TSource> action)
 367                 {
 368                         if (source == null)
 369                                 throw new ArgumentNullException ("source");
 370                         if (action == null)
 371                                 throw new ArgumentNullException ("action");
 372
 373                         ParallelExecuter.ProcessAndBlock (source.Node, (e, c) => action (e));
 374                 }
 375                 #endregion
 376
 377                 #region OrderBy
 378                 public static OrderedParallelQuery<TSource> OrderByDescending<TSource, TKey> (this ParallelQuery<TSource> source,
 379                                                                                               Func<TSource, TKey> keySelector,
 380                                                                                               IComparer<TKey> comparer)
 381                 {
 382                         if (source == null)
 383                                 throw new ArgumentNullException ("source");
 384                         if (keySelector == null)
 385                                 throw new ArgumentNullException ("keySelector");
 386                         if (comparer == null)
 387                                 comparer = Comparer<TKey>.Default;
 388
 389                         Comparison<TSource> comparison = (e1, e2) => -comparer.Compare (keySelector (e1), keySelector (e2));
 390
 391                         return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
 392                 }
 393
 394                 public static OrderedParallelQuery<TSource> OrderByDescending<TSource, TKey> (this ParallelQuery<TSource> source,
 395                                                                                               Func<TSource, TKey> keySelector)
 396                 {
 397                         return OrderByDescending (source, keySelector, Comparer<TKey>.Default);
 398                 }
 399
 400                 public static OrderedParallelQuery<TSource> OrderBy<TSource, TKey> (this ParallelQuery<TSource> source,
 401                                                                                     Func<TSource, TKey> keySelector)
 402                 {
 403                         return OrderBy (source, keySelector, Comparer<TKey>.Default);
 404                 }
 405
 406                 public static OrderedParallelQuery<TSource> OrderBy<TSource, TKey> (this ParallelQuery<TSource> source,
 407                                                                                     Func<TSource, TKey> keySelector,
 408                                                                                     IComparer<TKey> comparer)
 409                 {
 410                         if (source == null)
 411                                 throw new ArgumentNullException ("source");
 412                         if (keySelector == null)
 413                                 throw new ArgumentNullException ("keySelector");
 414                         if (comparer == null)
 415                                 comparer = Comparer<TKey>.Default;
 416
 417                         Comparison<TSource> comparison = (e1, e2) => comparer.Compare (keySelector (e1), keySelector (e2));
 418
 419                         return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
 420                 }
 421                 #endregion
 422
 423                 #region ThenBy
 424                 public static OrderedParallelQuery<TSource> ThenBy<TSource, TKey> (this OrderedParallelQuery<TSource> source,
 425                                                                                    Func<TSource, TKey> keySelector)
 426                 {
 427                         return ThenBy (source, keySelector, Comparer<TKey>.Default);
 428                 }
 429
 430                 public static OrderedParallelQuery<TSource> ThenBy<TSource, TKey> (this OrderedParallelQuery<TSource> source,
 431                                                                                    Func<TSource, TKey> keySelector,
 432                                                                                    IComparer<TKey> comparer)
 433                 {
 434                         if (source == null)
 435                                 throw new ArgumentNullException ("source");
 436                         if (keySelector == null)
 437                                 throw new ArgumentNullException ("keySelector");
 438                         if (comparer == null)
 439                                 comparer = Comparer<TKey>.Default;
 440
 441                         Comparison<TSource> comparison = (e1, e2) => comparer.Compare (keySelector (e1), keySelector (e2));
 442
 443                         return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
 444                 }
 445
 446                 public static OrderedParallelQuery<TSource> ThenByDescending<TSource, TKey> (this OrderedParallelQuery<TSource> source,
 447                                                                                              Func<TSource, TKey> keySelector)
 448                 {
 449                         return ThenByDescending (source, keySelector, Comparer<TKey>.Default);
 450                 }
 451
 452                 public static OrderedParallelQuery<TSource> ThenByDescending<TSource, TKey> (this OrderedParallelQuery<TSource> source,
 453                                                                                              Func<TSource, TKey> keySelector,
 454                                                                                              IComparer<TKey> comparer)
 455                 {
 456                         if (source == null)
 457                                 throw new ArgumentNullException ("source");
 458                         if (keySelector == null)
 459                                 throw new ArgumentNullException ("keySelector");
 460                         if (comparer == null)
 461                                 comparer = Comparer<TKey>.Default;
 462
 463                         Comparison<TSource> comparison = (e1, e2) => -comparer.Compare (keySelector (e1), keySelector (e2));
 464
 465                         return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
 466                 }
 467                 #endregion
 468
 469                 #region All
 470                 public static bool All<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
 471                 {
 472                         if (source == null)
 473                                 throw new ArgumentNullException ("source");
 474                         if (predicate == null)
 475                                 throw new ArgumentNullException ("predicate");
 476
 477                         CancellationTokenSource src = new CancellationTokenSource ();
 478                         ParallelQuery<TSource> innerQuery = source.WithImplementerToken (src);
 479
 480                         bool result = true;
 481                         try {
 482                                 innerQuery.ForAll ((e) => {
 483                                                 if (!predicate (e)) {
 484                                                         result = false;
 485                                                         src.Cancel ();
 486                                                 }
 487                                         });
 488                         } catch (OperationCanceledException e) {
 489                                 if (e.CancellationToken != src.Token)
 490                                         throw e;
 491                         }
 492
 493                         return result;
 494                 }
 495                 #endregion
 496
 497                 #region Any
 498                 public static bool Any<TSource> (this ParallelQuery<TSource> source)
 499                 {
 500                         return Any<TSource> (source, (_) => true);
 501                 }
 502
 503                 public static bool Any<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
 504                 {
 505                         if (source == null)
 506                                 throw new ArgumentNullException ("source");
 507                         if (predicate == null)
 508                                 throw new ArgumentNullException ("predicate");
 509
 510                         return !source.All ((e) => !predicate (e));
 511                 }
 512                 #endregion
 513
 514                 #region Contains
 515                 public static bool Contains<TSource> (this ParallelQuery<TSource> source, TSource value)
 516                 {
 517                         return Contains<TSource> (source, value, EqualityComparer<TSource>.Default);
 518                 }
 519
 520                 public static bool Contains<TSource> (this ParallelQuery<TSource> source, TSource value, IEqualityComparer<TSource> comparer)
 521                 {
 522                         if (source == null)
 523                                 throw new ArgumentNullException ("source");
 524                         if (comparer == null)
 525                                 comparer = EqualityComparer<TSource>.Default;
 526
 527                         return Any<TSource> (source, (e) => comparer.Equals (value));
 528                 }
 529                 #endregion
 530
 531                 #region SequenceEqual
 532                 public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first,
 533                                                            ParallelQuery<TSource> second)
 534                 {
 535                         if (first == null)
 536                                 throw new ArgumentNullException ("first");
 537                         if (second == null)
 538                                 throw new ArgumentNullException ("second");
 539
 540                         return first.SequenceEqual (second, EqualityComparer<TSource>.Default);
 541                 }
 542
 543                 public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first,
 544                                                            ParallelQuery<TSource> second,
 545                                                            IEqualityComparer<TSource> comparer)
 546                 {
 547                         if (first == null)
 548                                 throw new ArgumentNullException ("first");
 549                         if (second == null)
 550                                 throw new ArgumentNullException ("second");
 551                         if (comparer == null)
 552                                 comparer = EqualityComparer<TSource>.Default;
 553
 554                         CancellationTokenSource source = new CancellationTokenSource ();
 555                         ParallelQuery<bool> innerQuery
 556                                 = first.Zip (second, (e1, e2) => comparer.Equals (e1, e2)).Where ((e) => !e).WithImplementerToken (source);
 557
 558                         bool result = true;
 559
 560                         try {
 561                                 innerQuery.ForAll ((value) => {
 562                                                 result = false;
 563                                                 source.Cancel ();
 564                                         });
 565                         } catch (OperationCanceledException e) {
 566                                 if (e.CancellationToken != source.Token)
 567                                         throw e;
 568                         }
 569
 570                         return result;
 571                 }
 572
 573                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 574                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 575                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 576                 public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first, IEnumerable<TSource> second)
 577                 {
 578                         throw new NotSupportedException ();
 579                 }
 580
 581                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 582                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 583                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 584                 public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first,
 585                                                               IEnumerable<TSource> second,
 586                                                               IEqualityComparer<TSource> comparer)
 587                 {
 588                         throw new NotSupportedException ();
 589                 }
 590
 591                 #endregion
 592
 593                 #region GroupBy
 594                 public static ParallelQuery<IGrouping<TKey, TSource>> GroupBy<TSource, TKey> (this ParallelQuery<TSource> source,
 595                                                                                               Func<TSource, TKey> keySelector)
 596                 {
 597                         return source.GroupBy (keySelector, EqualityComparer<TKey>.Default);
 598                 }
 599
 600                 public static ParallelQuery<IGrouping<TKey, TSource>> GroupBy<TSource, TKey> (this ParallelQuery<TSource> source,
 601                                                                                               Func<TSource, TKey> keySelector,
 602                                                                                               IEqualityComparer<TKey> comparer)
 603                 {
 604                         return source.GroupBy (keySelector, (e) => e, comparer);
 605                 }
 606
 607                 public static ParallelQuery<IGrouping<TKey, TElement>> GroupBy<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
 608                                                                                                          Func<TSource, TKey> keySelector,
 609                                                                                                          Func<TSource, TElement> elementSelector)
 610                 {
 611                         return source.GroupBy (keySelector, elementSelector, EqualityComparer<TKey>.Default);
 612                 }
 613
 614                 public static ParallelQuery<TResult> GroupBy<TSource, TKey, TResult> (this ParallelQuery<TSource> source,
 615                                                                                       Func<TSource, TKey> keySelector,
 616                                                                                       Func<TKey, IEnumerable<TSource>, TResult> resultSelector)
 617                 {
 618                         return source.GroupBy (keySelector)
 619                                 .Select ((g) => resultSelector (g.Key, (IEnumerable<TSource>)g));
 620                 }
 621
 622                 public static ParallelQuery<IGrouping<TKey, TElement>> GroupBy<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
 623                                                                                                          Func<TSource, TKey> keySelector,
 624                                                                                                          Func<TSource, TElement> elementSelector,
 625                                                                                                          IEqualityComparer<TKey> comparer)
 626                 {
 627                         if (source == null)
 628                                 throw new ArgumentNullException ("source");
 629                         if (keySelector == null)
 630                                 throw new ArgumentNullException ("keySelector");
 631                         if (elementSelector == null)
 632                                 throw new ArgumentNullException ("elementSelector");
 633                         if (comparer == null)
 634                                 comparer = EqualityComparer<TKey>.Default;
 635
 636                         return new ParallelQuery<IGrouping<TKey, TElement>> (new QueryGroupByNode<TSource, TKey, TElement> (source.Node, keySelector, elementSelector, comparer));
 637                 }
 638
 639                 public static ParallelQuery<TResult> GroupBy<TSource, TKey, TElement, TResult> (this ParallelQuery<TSource> source,
 640                                                                                                 Func<TSource, TKey> keySelector,
 641                                                                                                 Func<TSource, TElement> elementSelector,
 642                                                                                                 Func<TKey, IEnumerable<TElement>, TResult> resultSelector)
 643                 {
 644                         return source.GroupBy (keySelector, elementSelector)
 645                                 .Select ((g) => resultSelector (g.Key, (IEnumerable<TElement>)g));
 646                 }
 647
 648                 public static ParallelQuery<TResult> GroupBy<TSource, TKey, TResult> (this ParallelQuery<TSource> source,
 649                                                                                       Func<TSource, TKey> keySelector,
 650                                                                                       Func<TKey, IEnumerable<TSource>, TResult> resultSelector,
 651                                                                                       IEqualityComparer<TKey> comparer)
 652                 {
 653                         return source.GroupBy (keySelector, comparer)
 654                                 .Select ((g) => resultSelector (g.Key, (IEnumerable<TSource>)g));
 655                 }
 656
 657                 public static ParallelQuery<TResult> GroupBy<TSource, TKey, TElement, TResult> (this ParallelQuery<TSource> source,
 658                                                                                                 Func<TSource, TKey> keySelector,
 659                                                                                                 Func<TSource, TElement> elementSelector,
 660                                                                                                 Func<TKey, IEnumerable<TElement>, TResult> resultSelector,
 661                                                                                                 IEqualityComparer<TKey> comparer)
 662                 {
 663                         return source.GroupBy (keySelector, elementSelector, comparer)
 664                                 .Select ((g) => resultSelector (g.Key, (IEnumerable<TElement>)g));
 665                 }
 666                 #endregion
 667
 668                 #region GroupJoin
 669                 public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 670                                                                                                ParallelQuery<TInner> inner,
 671                                                                                                Func<TOuter, TKey> outerKeySelector,
 672                                                                                                Func<TInner, TKey> innerKeySelector,
 673                                                                                                Func<TOuter, IEnumerable<TInner>, TResult> resultSelector)
 674                 {
 675                         return outer.GroupJoin (inner, outerKeySelector, innerKeySelector, resultSelector, EqualityComparer<TKey>.Default);
 676                 }
 677
 678                 public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 679                                                                                                ParallelQuery<TInner> inner,
 680                                                                                                Func<TOuter, TKey> outerKeySelector,
 681                                                                                                Func<TInner, TKey> innerKeySelector,
 682                                                                                                Func<TOuter, IEnumerable<TInner>, TResult> resultSelector,
 683                                                                                                IEqualityComparer<TKey> comparer)
 684                 {
 685                         return outer.Join (inner.GroupBy (innerKeySelector, (e) => e), outerKeySelector, (e) => e.Key, (e1, e2) => resultSelector (e1, e2), comparer);
 686                 }
 687
 688                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 689                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 690                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 691                 public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 692                                                                                                IEnumerable<TInner> inner,
 693                                                                                                Func<TOuter, TKey> outerKeySelector,
 694                                                                                                Func<TInner, TKey> innerKeySelector,
 695                                                                                                Func<TOuter, IEnumerable<TInner>, TResult> resultSelector)
 696                 {
 697                         throw new NotSupportedException ();
 698                 }
 699
 700                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 701                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 702                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 703                 public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 704                                                                                                IEnumerable<TInner> inner,
 705                                                                                                Func<TOuter, TKey> outerKeySelector,
 706                                                                                                Func<TInner, TKey> innerKeySelector,
 707                                                                                                Func<TOuter, IEnumerable<TInner>, TResult> resultSelector,
 708                                                                                                IEqualityComparer<TKey> comparer)
 709                 {
 710                         throw new NotSupportedException ();
 711                 }
 712                 #endregion
 713
 714                 #region ElementAt
 715                 public static TSource ElementAt<TSource> (this ParallelQuery<TSource> source, int index)
 716                 {
 717                         if (source == null)
 718                                 throw new ArgumentNullException ("source");
 719                         if (index < 0)
 720                                 throw new ArgumentOutOfRangeException ("index");
 721                         if (index == 0) {
 722                                 try {
 723                                         return source.First ();
 724                                 } catch (InvalidOperationException) {
 725                                         throw new ArgumentOutOfRangeException ("index");
 726                                 }
 727                         }
 728
 729                         TSource result = default (TSource);
 730
 731                         ParallelQuery<TSource> innerQuery = source.Where ((e, i) => i == index);
 732
 733                         try {
 734                                 result = innerQuery.First ();
 735                         } catch (InvalidOperationException) {
 736                                 throw new ArgumentOutOfRangeException ("index");
 737                         }
 738
 739                         return result;
 740                 }
 741
 742                 public static TSource ElementAtOrDefault<TSource> (this ParallelQuery<TSource> source, int index)
 743                 {
 744                         if (source == null)
 745                                 throw new ArgumentNullException ("source");
 746
 747                         try {
 748                                 return source.ElementAt (index);
 749                         } catch (ArgumentOutOfRangeException) {
 750                                 return default (TSource);
 751                         }
 752                 }
 753                 #endregion
 754
 755                 #region Intersect
 756                 public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first,
 757                                                                          ParallelQuery<TSource> second)
 758                 {
 759                         return Intersect<TSource> (first, second, EqualityComparer<TSource>.Default);
 760                 }
 761
 762                 public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first,
 763                                                                          ParallelQuery<TSource> second,
 764                                                                          IEqualityComparer<TSource> comparer)
 765                 {
 766                         if (first == null)
 767                                 throw new ArgumentNullException ("first");
 768                         if (second == null)
 769                                 throw new ArgumentNullException ("second");
 770                         if (comparer == null)
 771                                 comparer = EqualityComparer<TSource>.Default;
 772
 773                         return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Intersect, comparer, first.Node, second.Node));
 774                 }
 775
 776                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 777                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 778                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 779                 public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first, IEnumerable<TSource> second)
 780                 {
 781                         throw new NotSupportedException ();
 782                 }
 783
 784                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 785                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 786                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 787                 public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first,
 788                                                                          IEnumerable<TSource> second,
 789                                                                          IEqualityComparer<TSource> comparer)
 790                 {
 791                         throw new NotSupportedException ();
 792                 }
 793                 #endregion
 794
 795                 #region Join
 796                 public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 797                                                                                           ParallelQuery<TInner> inner,
 798                                                                                           Func<TOuter, TKey> outerKeySelector,
 799                                                                                           Func<TInner, TKey> innerKeySelector,
 800                                                                                           Func<TOuter, TInner, TResult> resultSelector)
 801                 {
 802                         return outer.Join (inner, outerKeySelector, innerKeySelector, resultSelector, EqualityComparer<TKey>.Default);
 803                 }
 804
 805                 public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 806                                                                                           ParallelQuery<TInner> inner,
 807                                                                                           Func<TOuter, TKey> outerKeySelector,
 808                                                                                           Func<TInner, TKey> innerKeySelector,
 809                                                                                           Func<TOuter, TInner, TResult> resultSelector,
 810                                                                                           IEqualityComparer<TKey> comparer)
 811                 {
 812                         return new ParallelQuery<TResult> (new QueryJoinNode<TOuter, TInner, TKey, TResult> (outer.Node, inner.Node, outerKeySelector, innerKeySelector, resultSelector, comparer));
 813                 }
 814
 815                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 816                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 817                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 818                 public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 819                                                                                           IEnumerable<TInner> inner,
 820                                                                                           Func<TOuter, TKey> outerKeySelector,
 821                                                                                           Func<TInner, TKey> innerKeySelector,
 822                                                                                           Func<TOuter, TInner, TResult> resultSelector)
 823                 {
 824                         throw new NotSupportedException ();
 825                 }
 826
 827                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 828                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 829                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 830                 public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 831                                                                                           IEnumerable<TInner> inner,
 832                                                                                           Func<TOuter, TKey> outerKeySelector,
 833                                                                                           Func<TInner, TKey> innerKeySelector,
 834                                                                                           Func<TOuter, TInner, TResult> resultSelector,
 835                                                                                           IEqualityComparer<TKey> comparer)
 836                 {
 837                         throw new NotSupportedException ();
 838                 }
 839                 #endregion
 840
 841                 #region Except
 842                 public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
 843                                                                       ParallelQuery<TSource> second)
 844                 {
 845                         return Except<TSource> (first, second, EqualityComparer<TSource>.Default);
 846                 }
 847
 848                 public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
 849                                                                       ParallelQuery<TSource> second,
 850                                                                       IEqualityComparer<TSource> comparer)
 851                 {
 852                         if (first == null)
 853                                 throw new ArgumentNullException ("first");
 854                         if (second == null)
 855                                 throw new ArgumentNullException ("second");
 856                         if (comparer == null)
 857                                 comparer = EqualityComparer<TSource>.Default;
 858
 859                         return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Except,
 860                                                                                       comparer, first.Node, second.Node));
 861                 }
 862
 863                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 864                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 865                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 866                 public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
 867                                                                       IEnumerable<TSource> second)
 868                 {
 869                         throw new NotSupportedException ();
 870                 }
 871
 872                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 873                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 874                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 875                 public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
 876                                                                       IEnumerable<TSource> second,
 877                                                                       IEqualityComparer<TSource> comparer)
 878                 {
 879                         throw new NotSupportedException ();
 880                 }
 881                 #endregion
 882
 883                 #region Distinct
 884                 public static ParallelQuery<TSource> Distinct<TSource> (this ParallelQuery<TSource> source)
 885                 {
 886                         return Distinct<TSource> (source, EqualityComparer<TSource>.Default);
 887                 }
 888
 889                 public static ParallelQuery<TSource> Distinct<TSource> (this ParallelQuery<TSource> source, IEqualityComparer<TSource> comparer)
 890                 {
 891                         if (source == null)
 892                                 throw new ArgumentNullException ("source");
 893                         if (comparer == null)
 894                                 comparer = EqualityComparer<TSource>.Default;
 895
 896                         return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Distinct, comparer,
 897                                                                                       source.Node, null));
 898                 }
 899                 #endregion
 900
 901                 #region Union
 902                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 903                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 904                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 905                 public static ParallelQuery<TSource> Union<TSource> (this ParallelQuery<TSource> first,
 906                                                                      IEnumerable<TSource> second)
 907                 {
 908                         throw new NotSupportedException ();
 909                 }
 910
 911                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 912                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 913                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 914                 public static ParallelQuery<TSource> Union<TSource>(this ParallelQuery<TSource> first,
 915                                                                     IEnumerable<TSource> second,
 916                                                                     IEqualityComparer<TSource> comparer)
 917                 {
 918                         throw new NotSupportedException ();
 919                 }
 920
 921                 public static ParallelQuery<TSource> Union<TSource> (this ParallelQuery<TSource> first,
 922                                                                      ParallelQuery<TSource> second)
 923                 {
 924                         return first.Union (second, EqualityComparer<TSource>.Default);
 925                 }
 926
 927                 public static ParallelQuery<TSource> Union<TSource> (this ParallelQuery<TSource> first,
 928                                                                      ParallelQuery<TSource> second,
 929                                                                      IEqualityComparer<TSource> comparer)
 930                 {
 931                         if (first == null)
 932                                 throw new ArgumentNullException ("first");
 933                         if (second == null)
 934                                 throw new ArgumentNullException ("second");
 935                         if (comparer == null)
 936                                 comparer = EqualityComparer<TSource>.Default;
 937
 938                         return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Union, comparer, first.Node, second.Node));
 939                 }
 940                 #endregion
 941
 942                 #region Take
 943                 public static ParallelQuery<TSource> Take<TSource> (this ParallelQuery<TSource> source, int count)
 944                 {
 945                         if (source == null)
 946                                 throw new ArgumentNullException ("source");
 947
 948                         return new ParallelQuery<TSource> (new QueryHeadWorkerNode<TSource> (source.Node, count));
 949                 }
 950
 951                 public static ParallelQuery<TSource> TakeWhile<TSource> (this ParallelQuery<TSource> source,
 952                                                                          Func<TSource, bool> predicate)
 953                 {
 954                         if (source == null)
 955                                 throw new ArgumentNullException ("source");
 956                         if (predicate == null)
 957                                 throw new ArgumentNullException ("predicate");
 958
 959                         return new ParallelQuery<TSource> (new QueryHeadWorkerNode<TSource> (source.Node, (e, _) => predicate (e), false));
 960                 }
 961
 962                 public static ParallelQuery<TSource> TakeWhile<TSource> (this ParallelQuery<TSource> source,
 963                                                                          Func<TSource, int, bool> predicate)
 964                 {
 965                         if (source == null)
 966                                 throw new ArgumentNullException ("source");
 967                         if (predicate == null)
 968                                 throw new ArgumentNullException ("predicate");
 969
 970                         return new ParallelQuery<TSource> (new QueryHeadWorkerNode<TSource> (source.Node, predicate, true));
 971                 }
 972                 #endregion
 973
 974                 #region Skip
 975                 public static ParallelQuery<TSource> Skip<TSource> (this ParallelQuery<TSource> source, int count)
 976                 {
 977                         if (source == null)
 978                                 throw new ArgumentNullException ("source");
 979
 980                         return source.Node.IsOrdered () ?
 981                                 source.Where ((e, i) => i >= count) :
 982                                 source.Where ((e) => count < 0 || Interlocked.Decrement (ref count) < 0);
 983
 984                 }
 985
 986                 public static ParallelQuery<TSource> SkipWhile<TSource> (this ParallelQuery<TSource> source,
 987                                                                          Func<TSource, bool> predicate)
 988                 {
 989                         if (source == null)
 990                                 throw new ArgumentNullException ("source");
 991                         if (predicate == null)
 992                                 throw new ArgumentNullException ("predicate");
 993
 994                         return source.Node.IsOrdered () ?
 995                                 source.SkipWhile ((e, i) => predicate (e)) :
 996                                 source.Where ((e) => !predicate (e));
 997                 }
 998
 999                 public static ParallelQuery<TSource> SkipWhile<TSource> (this ParallelQuery<TSource> source,
1000                                                                          Func<TSource, int, bool> predicate)
1001                 {
1002                         if (source == null)
1003                                 throw new ArgumentNullException ("source");
1004                         if (predicate == null)
1005                                 throw new ArgumentNullException ("predicate");
1006
1007                         int indexCache = int.MaxValue;
1008
1009                         return source.Where ((e, i) => i >= indexCache || (!predicate (e, i) && (indexCache = i) == i));
1010                 }
1011                 #endregion
1012
1013                 #region Single
1014                 static TSource SingleInternal<TSource> (this ParallelQuery<TSource> source, params TSource[] init)
1015                 {
1016                         TSource result = default(TSource);
1017                         bool hasValue = false;
1018
1019                         foreach (TSource element in source) {
1020                                 if (hasValue)
1021                                         throw new InvalidOperationException ("The input sequence contains more than one element.");
1022
1023                                 result = element;
1024                                 hasValue = true;
1025                         }
1026
1027                         if (!hasValue && init.Length != 0) {
1028                                 result = init[0];
1029                                 hasValue = true;
1030                         }
1031
1032                         if (!hasValue)
1033                                 throw new InvalidOperationException ("The input sequence is empty.");
1034
1035                         return result;
1036                 }
1037
1038                 public static TSource Single<TSource> (this ParallelQuery<TSource> source)
1039                 {
1040                         if (source == null)
1041                                 throw new ArgumentNullException ("source");
1042
1043                         return SingleInternal<TSource> (source);
1044                 }
1045
1046                 public static TSource Single<TSource> (this ParallelQuery<TSource> source,
1047                                                        Func<TSource, bool> predicate)
1048                 {
1049                         if (source == null)
1050                                 throw new ArgumentNullException ("source");
1051                         if (predicate == null)
1052                                 throw new ArgumentNullException ("predicate");
1053
1054                         return source.Where (predicate).Single ();
1055                 }
1056
1057                 public static TSource SingleOrDefault<TSource> (this ParallelQuery<TSource> source)
1058                 {
1059                         if (source == null)
1060                                 throw new ArgumentNullException ("source");
1061
1062                         return SingleInternal<TSource> (source, default (TSource));
1063                 }
1064
1065                 public static TSource SingleOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
1066                 {
1067                         if (source == null)
1068                                 throw new ArgumentNullException ("source");
1069                         if (predicate == null)
1070                                 throw new ArgumentNullException ("predicate");
1071
1072                         return source.Where (predicate).SingleOrDefault ();
1073                 }
1074                 #endregion
1075
1076                 #region Count
1077                 public static int Count<TSource> (this ParallelQuery<TSource> source)
1078                 {
1079                         if (source == null)
1080                                 throw new ArgumentNullException ("source");
1081
1082                         return source.Aggregate<TSource, int, int> (() => 0,
1083                                                                    (acc, e) => acc + 1,
1084                                                                    (acc1, acc2) => acc1 + acc2,
1085                                                                    (result) => result);
1086                 }
1087
1088                 public static int Count<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
1089                 {
1090                         if (source == null)
1091                                 throw new ArgumentNullException ("source");
1092                         if (predicate == null)
1093                                 throw new ArgumentNullException ("predicate");
1094
1095                         return source.Where (predicate).Count ();
1096                 }
1097
1098                 public static long LongCount<TSource> (this ParallelQuery<TSource> source)
1099                 {
1100                         if (source == null)
1101                                 throw new ArgumentNullException ("source");
1102
1103                         return source.Aggregate<TSource, long, long> (() => 0,
1104                                                                       (acc, e) => acc + 1,
1105                                                                       (acc1, acc2) => acc1 + acc2,
1106                                                                       (result) => result);
1107                 }
1108
1109                 public static long LongCount<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
1110                 {
1111                         if (source == null)
1112                                 throw new ArgumentNullException ("source");
1113                         if (predicate == null)
1114                                 throw new ArgumentNullException ("predicate");
1115
1116                         return source.Where (predicate).LongCount ();
1117                 }
1118                 #endregion
1119
1120                 #region Average
1121                 public static double Average (this ParallelQuery<int> source)
1122                 {
1123                         if (source == null)
1124                                 throw new ArgumentNullException ("source");
1125
1126                         return source.Aggregate (() => new int[2],
1127                                                 (acc, e) => { acc[0] += e; acc[1]++; return acc; },
1128                                                 (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
1129                                                 (acc) => acc[0] / ((double)acc[1]));
1130                 }
1131
1132                 public static double Average (this ParallelQuery<long> source)
1133                 {
1134                         if (source == null)
1135                                 throw new ArgumentNullException ("source");
1136
1137                         return source.Aggregate (() => new long[2],
1138                                                 (acc, e) => { acc[0] += e; acc[1]++; return acc; },
1139                                                 (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
1140                                                 (acc) => acc[0] / ((double)acc[1]));
1141                 }
1142
1143                 public static decimal Average (this ParallelQuery<decimal> source)
1144                 {
1145                         if (source == null)
1146                                 throw new ArgumentNullException ("source");
1147
1148                         return source.Aggregate (() => new decimal[2],
1149                                                 (acc, e) => { acc[0] += e; acc[1]++; return acc; },
1150                                                 (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
1151                                                 (acc) => acc[0] / acc[1]);
1152                 }
1153
1154                 public static double Average (this ParallelQuery<double> source)
1155                 {
1156                         if (source == null)
1157                                 throw new ArgumentNullException ("source");
1158
1159                         return source.Aggregate (() => new double[2],
1160                                                 (acc, e) => { acc[0] += e; acc[1]++; return acc; },
1161                                                 (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
1162                                                 (acc) => acc[0] / ((double)acc[1]));
1163                 }
1164
1165                 public static float Average (this ParallelQuery<float> source)
1166                 {
1167                         if (source == null)
1168                                 throw new ArgumentNullException ("source");
1169
1170                         return source.Aggregate (() => new float[2],
1171                                                 (acc, e) => { acc[0] += e; acc[1]++; return acc; },
1172                                                 (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
1173                                                 (acc) => acc[0] / acc[1]);
1174                 }
1175                 #endregion
1176
1177                 #region More Average
1178                 public static double? Average (this ParallelQuery<int?> source)
1179                 {
1180                         if (source == null)
1181                                 throw new ArgumentNullException ("source");
1182
1183                         return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();;
1184                 }
1185
1186                 public static double? Average (this ParallelQuery<long?> source)
1187                 {
1188                         if (source == null)
1189                                 throw new ArgumentNullException ("source");
1190
1191                         return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
1192                 }
1193
1194                 public static decimal? Average (this ParallelQuery<decimal?> source)
1195                 {
1196                         if (source == null)
1197                                 throw new ArgumentNullException ("source");
1198
1199                         return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
1200                 }
1201
1202                 public static double? Average (this ParallelQuery<double?> source)
1203                 {
1204                         if (source == null)
1205                                 throw new ArgumentNullException ("source");
1206
1207                         return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
1208                 }
1209
1210                 public static float? Average (this ParallelQuery<float?> source)
1211                 {
1212                         if (source == null)
1213                                 throw new ArgumentNullException ("source");
1214
1215                         return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
1216                 }
1217
1218                 public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
1219                 {
1220                         if (source == null)
1221                                 throw new ArgumentNullException ("source");
1222                         if (selector == null)
1223                                 throw new ArgumentNullException ("selector");
1224
1225                         return source.Select (selector).Average ();
1226                 }
1227
1228                 public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
1229                 {
1230                         if (source == null)
1231                                 throw new ArgumentNullException ("source");
1232                         if (selector == null)
1233                                 throw new ArgumentNullException ("selector");
1234
1235                         return source.Select (selector).Average ();
1236                 }
1237
1238                 public static float Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
1239                 {
1240                         if (source == null)
1241                                 throw new ArgumentNullException ("source");
1242                         if (selector == null)
1243                                 throw new ArgumentNullException ("selector");
1244
1245                         return source.Select (selector).Average ();
1246                 }
1247
1248                 public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
1249                 {
1250                         if (source == null)
1251                                 throw new ArgumentNullException ("source");
1252                         if (selector == null)
1253                                 throw new ArgumentNullException ("selector");
1254
1255                         return source.Select (selector).Average ();
1256                 }
1257
1258                 public static decimal Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
1259                 {
1260                         if (source == null)
1261                                 throw new ArgumentNullException ("source");
1262                         if (selector == null)
1263                                 throw new ArgumentNullException ("selector");
1264
1265                         return source.Select (selector).Average ();
1266                 }
1267
1268                 public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
1269                 {
1270                         if (source == null)
1271                                 throw new ArgumentNullException ("source");
1272                         if (selector == null)
1273                                 throw new ArgumentNullException ("selector");
1274
1275                         return source.Select (selector).Average ();
1276                 }
1277
1278                 public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
1279                 {
1280                         if (source == null)
1281                                 throw new ArgumentNullException ("source");
1282                         if (selector == null)
1283                                 throw new ArgumentNullException ("selector");
1284
1285                         return source.Select (selector).Average ();
1286                 }
1287
1288                 public static float? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
1289                 {
1290                         if (source == null)
1291                                 throw new ArgumentNullException ("source");
1292                         if (selector == null)
1293                                 throw new ArgumentNullException ("selector");
1294
1295                         return source.Select (selector).Average ();
1296                 }
1297
1298                 public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
1299                 {
1300                         if (source == null)
1301                                 throw new ArgumentNullException ("source");
1302                         if (selector == null)
1303                                 throw new ArgumentNullException ("selector");
1304
1305                         return source.Select (selector).Average ();
1306                 }
1307
1308                 public static decimal? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
1309                 {
1310                         if (source == null)
1311                                 throw new ArgumentNullException ("source");
1312                         if (selector == null)
1313                                 throw new ArgumentNullException ("selector");
1314
1315                         return source.Select (selector).Average ();
1316                 }
1317                 #endregion
1318
1319                 #region Sum
1320                 public static int Sum (this ParallelQuery<int> source)
1321                 {
1322                         if (source == null)
1323                                 throw new ArgumentNullException ("source");
1324
1325                         return source.Aggregate (0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
1326                 }
1327
1328                 public static long Sum (this ParallelQuery<long> source)
1329                 {
1330                         if (source == null)
1331                                 throw new ArgumentNullException ("source");
1332
1333                         return source.Aggregate ((long)0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
1334                 }
1335
1336                 public static float Sum (this ParallelQuery<float> source)
1337                 {
1338                         if (source == null)
1339                                 throw new ArgumentNullException ("source");
1340
1341                         return source.Aggregate (0.0f, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
1342                 }
1343
1344                 public static double Sum (this ParallelQuery<double> source)
1345                 {
1346                         if (source == null)
1347                                 throw new ArgumentNullException ("source");
1348
1349                         return source.Aggregate (0.0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
1350                 }
1351
1352                 public static decimal Sum (this ParallelQuery<decimal> source)
1353                 {
1354                         if (source == null)
1355                                 throw new ArgumentNullException ("source");
1356
1357                         return source.Aggregate ((decimal)0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
1358                 }
1359
1360                 public static int? Sum (this ParallelQuery<int?> source)
1361                 {
1362                         return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
1363                 }
1364
1365                 public static long? Sum (this ParallelQuery<long?> source)
1366                 {
1367                         if (source == null)
1368                                 throw new ArgumentNullException ("source");
1369
1370                         return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
1371                 }
1372
1373                 public static float? Sum (this ParallelQuery<float?> source)
1374                 {
1375                         if (source == null)
1376                                 throw new ArgumentNullException ("source");
1377
1378                         return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
1379                 }
1380
1381                 public static double? Sum (this ParallelQuery<double?> source)
1382                 {
1383                         if (source == null)
1384                                 throw new ArgumentNullException ("source");
1385
1386                         return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
1387                 }
1388
1389                 public static decimal? Sum (this ParallelQuery<decimal?> source)
1390                 {
1391                         if (source == null)
1392                                 throw new ArgumentNullException ("source");
1393
1394                         return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
1395                 }
1396
1397                 public static int Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
1398                 {
1399                         if (source == null)
1400                                 throw new ArgumentNullException ("source");
1401                         if (selector == null)
1402                                 throw new ArgumentNullException ("selector");
1403
1404                         return source.Select (selector).Sum ();
1405                 }
1406
1407                 public static long Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
1408                 {
1409                         if (source == null)
1410                                 throw new ArgumentNullException ("source");
1411                         if (selector == null)
1412                                 throw new ArgumentNullException ("selector");
1413
1414                         return source.Select (selector).Sum ();
1415                 }
1416
1417                 public static decimal Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
1418                 {
1419                         if (source == null)
1420                                 throw new ArgumentNullException ("source");
1421                         if (selector == null)
1422                                 throw new ArgumentNullException ("selector");
1423
1424                         return source.Select (selector).Sum ();
1425                 }
1426
1427                 public static float Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
1428                 {
1429                         if (source == null)
1430                                 throw new ArgumentNullException ("source");
1431                         if (selector == null)
1432                                 throw new ArgumentNullException ("selector");
1433
1434                         return source.Select (selector).Sum ();
1435                 }
1436
1437                 public static double Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
1438                 {
1439                         if (source == null)
1440                                 throw new ArgumentNullException ("source");
1441                         if (selector == null)
1442                                 throw new ArgumentNullException ("selector");
1443
1444                         return source.Select (selector).Sum ();
1445                 }
1446
1447                 public static int? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
1448                 {
1449                         if (source == null)
1450                                 throw new ArgumentNullException ("source");
1451                         if (selector == null)
1452                                 throw new ArgumentNullException ("selector");
1453
1454                         return source.Select (selector).Sum ();
1455                 }
1456
1457                 public static long? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
1458                 {
1459                         if (source == null)
1460                                 throw new ArgumentNullException ("source");
1461                         if (selector == null)
1462                                 throw new ArgumentNullException ("selector");
1463
1464                         return source.Select (selector).Sum ();
1465                 }
1466
1467                 public static decimal? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
1468                 {
1469                         if (source == null)
1470                                 throw new ArgumentNullException ("source");
1471                         if (selector == null)
1472                                 throw new ArgumentNullException ("selector");
1473
1474                         return source.Select (selector).Sum ();
1475                 }
1476
1477                 public static float? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
1478                 {
1479                         if (source == null)
1480                                 throw new ArgumentNullException ("source");
1481                         if (selector == null)
1482                                 throw new ArgumentNullException ("selector");
1483
1484                         return source.Select (selector).Sum ();
1485                 }
1486
1487                 public static double? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
1488                 {
1489                         if (source == null)
1490                                 throw new ArgumentNullException ("source");
1491                         if (selector == null)
1492                                 throw new ArgumentNullException ("selector");
1493
1494                         return source.Select (selector).Sum ();
1495                 }
1496                 #endregion
1497
1498                 #region Min-Max
1499                 static T BestOrder<T> (ParallelQuery<T> source, Func<T, T, bool> bestSelector, T seed)
1500                 {
1501                         if (source == null)
1502                                 throw new ArgumentNullException ("source");
1503
1504                         T best = seed;
1505
1506                         best = source.Aggregate (() => seed,
1507                                                 (first, second) => (bestSelector(first, second)) ? first : second,
1508                                                 (first, second) => (bestSelector(first, second)) ? first : second,
1509                                                 (e) => e);
1510                         return best;
1511                 }
1512
1513                 public static int Min (this ParallelQuery<int> source)
1514                 {
1515                         return BestOrder (source, (first, second) => first < second, int.MaxValue);
1516                 }
1517
1518                 public static long Min (this ParallelQuery<long> source)
1519                 {
1520                         return BestOrder (source, (first, second) => first < second, long.MaxValue);
1521                 }
1522
1523                 public static float Min (this ParallelQuery<float> source)
1524                 {
1525                         return BestOrder (source, (first, second) => first < second, float.MaxValue);
1526                 }
1527
1528                 public static double Min (this ParallelQuery<double> source)
1529                 {
1530                         return BestOrder (source, (first, second) => first < second, double.MaxValue);
1531                 }
1532
1533                 public static decimal Min (this ParallelQuery<decimal> source)
1534                 {
1535                         return BestOrder (source, (first, second) => first < second, decimal.MaxValue);
1536                 }
1537
1538                 public static TSource Min<TSource> (this ParallelQuery<TSource> source)
1539                 {
1540                         IComparer<TSource> comparer = Comparer<TSource>.Default;
1541
1542                         return BestOrder (source, (first, second) => comparer.Compare (first, second) < 0, default (TSource));
1543                 }
1544
1545                 public static TResult Min<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> selector)
1546                 {
1547                         if (source == null)
1548                                 throw new ArgumentNullException ("source");
1549                         if (selector == null)
1550                                 throw new ArgumentNullException ("selector");
1551
1552                         return source.Select (selector).Min ();
1553                 }
1554
1555                 public static int? Min (this ParallelQuery<int?> source)
1556                 {
1557                         if (source == null)
1558                                 throw new ArgumentNullException ("source");
1559
1560                         return source.Select ((e) => e.HasValue ? e.Value : int.MaxValue).Min ();
1561                 }
1562
1563                 public static long? Min (this ParallelQuery<long?> source)
1564                 {
1565                         if (source == null)
1566                                 throw new ArgumentNullException ("source");
1567
1568                         return source.Select ((e) => e.HasValue ? e.Value : long.MaxValue).Min ();
1569                 }
1570
1571                 public static float? Min (this ParallelQuery<float?> source)
1572                 {
1573                         if (source == null)
1574                                 throw new ArgumentNullException ("source");
1575
1576                         return source.Select ((e) => e.HasValue ? e.Value : float.MaxValue).Min ();
1577                 }
1578
1579                 public static double? Min (this ParallelQuery<double?> source)
1580                 {
1581                         if (source == null)
1582                                 throw new ArgumentNullException ("source");
1583
1584                         return source.Select ((e) => e.HasValue ? e.Value : double.MaxValue).Min ();
1585                 }
1586
1587                 public static decimal? Min (this ParallelQuery<decimal?> source)
1588                 {
1589                         if (source == null)
1590                                 throw new ArgumentNullException ("source");
1591
1592                         return source.Select ((e) => e.HasValue ? e.Value : decimal.MaxValue).Min ();
1593                 }
1594
1595                 public static int Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
1596                 {
1597                         if (source == null)
1598                                 throw new ArgumentNullException ("source");
1599                         if (selector == null)
1600                                 throw new ArgumentNullException ("selector");
1601
1602                         return source.Select (selector).Min ();
1603                 }
1604
1605                 public static long Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
1606                 {
1607                         if (source == null)
1608                                 throw new ArgumentNullException ("source");
1609                         if (selector == null)
1610                                 throw new ArgumentNullException ("selector");
1611
1612                         return source.Select (selector).Min ();
1613                 }
1614
1615                 public static float Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
1616                 {
1617                         if (source == null)
1618                                 throw new ArgumentNullException ("source");
1619                         if (selector == null)
1620                                 throw new ArgumentNullException ("selector");
1621
1622                         return source.Select (selector).Min ();
1623                 }
1624
1625                 public static double Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
1626                 {
1627                         if (source == null)
1628                                 throw new ArgumentNullException ("source");
1629                         if (selector == null)
1630                                 throw new ArgumentNullException ("selector");
1631
1632                         return source.Select (selector).Min ();
1633                 }
1634
1635                 public static decimal Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
1636                 {
1637                         if (source == null)
1638                                 throw new ArgumentNullException ("source");
1639                         if (selector == null)
1640                                 throw new ArgumentNullException ("selector");
1641
1642                         return source.Select (selector).Min ();
1643                 }
1644
1645                 public static int? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
1646                 {
1647                         if (source == null)
1648                                 throw new ArgumentNullException ("source");
1649                         if (selector == null)
1650                                 throw new ArgumentNullException ("selector");
1651
1652                         return source.Select (selector).Min ();
1653                 }
1654
1655                 public static long? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
1656                 {
1657                         if (source == null)
1658                                 throw new ArgumentNullException ("source");
1659                         if (selector == null)
1660                                 throw new ArgumentNullException ("selector");
1661
1662                         return source.Select (selector).Min ();
1663                 }
1664
1665                 public static float? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
1666                 {
1667                         if (source == null)
1668                                 throw new ArgumentNullException ("source");
1669                         if (selector == null)
1670                                 throw new ArgumentNullException ("selector");
1671
1672                         return source.Select (selector).Min ();
1673                 }
1674
1675                 public static double? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
1676                 {
1677                         if (source == null)
1678                                 throw new ArgumentNullException ("source");
1679                         if (selector == null)
1680                                 throw new ArgumentNullException ("selector");
1681
1682                         return source.Select (selector).Min ();
1683                 }
1684
1685                 public static decimal? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
1686                 {
1687                         if (source == null)
1688                                 throw new ArgumentNullException ("source");
1689                         if (selector == null)
1690                                 throw new ArgumentNullException ("selector");
1691
1692                         return source.Select (selector).Min ();
1693                 }
1694
1695                 public static int Max (this ParallelQuery<int> source)
1696                 {
1697                         return BestOrder (source, (first, second) => first > second, int.MinValue);
1698                 }
1699
1700                 public static long Max(this ParallelQuery<long> source)
1701                 {
1702                         return BestOrder(source, (first, second) => first > second, long.MinValue);
1703                 }
1704
1705                 public static float Max (this ParallelQuery<float> source)
1706                 {
1707                         return BestOrder(source, (first, second) => first > second, float.MinValue);
1708                 }
1709
1710                 public static double Max (this ParallelQuery<double> source)
1711                 {
1712                         return BestOrder(source, (first, second) => first > second, double.MinValue);
1713                 }
1714
1715                 public static decimal Max (this ParallelQuery<decimal> source)
1716                 {
1717                         return BestOrder(source, (first, second) => first > second, decimal.MinValue);
1718                 }
1719
1720                 public static TSource Max<TSource> (this ParallelQuery<TSource> source)
1721                 {
1722                         IComparer<TSource> comparer = Comparer<TSource>.Default;
1723
1724                         return BestOrder (source, (first, second) => comparer.Compare (first, second) > 0, default (TSource));
1725                 }
1726
1727                 public static TResult Max<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> selector)
1728                 {
1729                         if (source == null)
1730                                 throw new ArgumentNullException ("source");
1731                         if (selector == null)
1732                                 throw new ArgumentNullException ("selector");
1733
1734                         return source.Select (selector).Max ();
1735                 }
1736
1737                 public static int? Max (this ParallelQuery<int?> source)
1738                 {
1739                         if (source == null)
1740                                 throw new ArgumentNullException ("source");
1741
1742                         return source.Select ((e) => e.HasValue ? e.Value : int.MinValue).Max ();
1743                 }
1744
1745                 public static long? Max (this ParallelQuery<long?> source)
1746                 {
1747                         if (source == null)
1748                                 throw new ArgumentNullException ("source");
1749
1750                         return source.Select ((e) => e.HasValue ? e.Value : long.MinValue).Max ();
1751                 }
1752
1753                 public static float? Max (this ParallelQuery<float?> source)
1754                 {
1755                         if (source == null)
1756                                 throw new ArgumentNullException ("source");
1757
1758                         return source.Select ((e) => e.HasValue ? e.Value : float.MinValue).Max ();
1759                 }
1760
1761                 public static double? Max (this ParallelQuery<double?> source)
1762                 {
1763                         if (source == null)
1764                                 throw new ArgumentNullException ("source");
1765
1766                         return source.Select ((e) => e.HasValue ? e.Value : double.MinValue).Max ();
1767                 }
1768
1769                 public static decimal? Max (this ParallelQuery<decimal?> source)
1770                 {
1771                         if (source == null)
1772                                 throw new ArgumentNullException ("source");
1773
1774                         return source.Select ((e) => e.HasValue ? e.Value : decimal.MinValue).Max ();
1775                 }
1776
1777                 public static int Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
1778                 {
1779                         if (source == null)
1780                                 throw new ArgumentNullException ("source");
1781                         if (selector == null)
1782                                 throw new ArgumentNullException ("selector");
1783
1784                         return source.Select (selector).Max ();
1785                 }
1786
1787                 public static long Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
1788                 {
1789                         if (source == null)
1790                                 throw new ArgumentNullException ("source");
1791                         if (selector == null)
1792                                 throw new ArgumentNullException ("selector");
1793
1794                         return source.Select (selector).Max ();
1795                 }
1796
1797                 public static float Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
1798                 {
1799                         if (source == null)
1800                                 throw new ArgumentNullException ("source");
1801                         if (selector == null)
1802                                 throw new ArgumentNullException ("selector");
1803
1804                         return source.Select (selector).Max ();
1805                 }
1806
1807                 public static double Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
1808                 {
1809                         if (source == null)
1810                                 throw new ArgumentNullException ("source");
1811                         if (selector == null)
1812                                 throw new ArgumentNullException ("selector");
1813
1814                         return source.Select (selector).Max ();
1815                 }
1816
1817                 public static decimal Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
1818                 {
1819                         if (source == null)
1820                                 throw new ArgumentNullException ("source");
1821                         if (selector == null)
1822                                 throw new ArgumentNullException ("selector");
1823
1824                         return source.Select (selector).Max ();
1825                 }
1826
1827                 public static int? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
1828                 {
1829                         if (source == null)
1830                                 throw new ArgumentNullException ("source");
1831                         if (selector == null)
1832                                 throw new ArgumentNullException ("selector");
1833
1834                         return source.Select (selector).Max ();
1835                 }
1836
1837                 public static long? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
1838                 {
1839                         if (source == null)
1840                                 throw new ArgumentNullException ("source");
1841                         if (selector == null)
1842                                 throw new ArgumentNullException ("selector");
1843
1844                         return source.Select (selector).Max ();
1845                 }
1846
1847                 public static float? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
1848                 {
1849                         if (source == null)
1850                                 throw new ArgumentNullException ("source");
1851                         if (selector == null)
1852                                 throw new ArgumentNullException ("selector");
1853
1854                         return source.Select (selector).Max ();
1855                 }
1856
1857                 public static double? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
1858                 {
1859                         if (source == null)
1860                                 throw new ArgumentNullException ("source");
1861                         if (selector == null)
1862                                 throw new ArgumentNullException ("selector");
1863
1864                         return source.Select (selector).Max ();
1865                 }
1866
1867                 public static decimal? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
1868                 {
1869                         if (source == null)
1870                                 throw new ArgumentNullException ("source");
1871                         if (selector == null)
1872                                 throw new ArgumentNullException ("selector");
1873
1874                         return source.Select (selector).Max ();
1875                 }
1876                 #endregion
1877
1878                 #region Cast / OfType
1879                 public static ParallelQuery<TResult> Cast<TResult> (this ParallelQuery source)
1880                 {
1881                         if (source == null)
1882                                 throw new ArgumentNullException ("source");
1883
1884                         return source.TypedQuery.Select ((e) => (TResult)e);
1885                 }
1886
1887                 public static ParallelQuery<TResult> OfType<TResult> (this ParallelQuery source)
1888                 {
1889                         if (source == null)
1890                                 throw new ArgumentNullException ("source");
1891
1892                         return source.TypedQuery.Where ((e) => e is TResult).Cast<TResult> ();
1893                 }
1894                 #endregion
1895
1896                 #region Reverse
1897                 public static ParallelQuery<TSource> Reverse<TSource> (this ParallelQuery<TSource> source)
1898                 {
1899                         if (source == null)
1900                                 throw new ArgumentNullException ("source");
1901
1902                         return new ParallelQuery<TSource> (new QueryReverseNode<TSource> (source));
1903                 }
1904                 #endregion
1905
1906                 #region ToArray - ToList - ToDictionary - ToLookup
1907                 public static List<TSource> ToList<TSource> (this ParallelQuery<TSource> source)
1908                 {
1909                         if (source == null)
1910                                 throw new ArgumentNullException ("source");
1911
1912                         if (source.Node.IsOrdered ())
1913                                 return ToListOrdered (source);
1914
1915                         List<TSource> temp = source.Aggregate (() => new List<TSource>(50),
1916                                                                (list, e) => { list.Add (e); return list; },
1917                                                                (list, list2) => { list.AddRange (list2); return list; },
1918                                                                (list) => list);
1919                         return temp;
1920                 }
1921
1922                 internal static List<TSource> ToListOrdered<TSource> (this ParallelQuery<TSource> source)
1923                 {
1924                         List<TSource> result = new List<TSource> ();
1925
1926                         foreach (TSource element in source)
1927                                 result.Add (element);
1928
1929                         return result;
1930                 }
1931
1932                 public static TSource[] ToArray<TSource> (this ParallelQuery<TSource> source)
1933                 {
1934                         if (source == null)
1935                                 throw new ArgumentNullException ("source");
1936
1937                         if (source.Node.IsOrdered ())
1938                                 return ToListOrdered (source).ToArray ();
1939
1940                         TSource[] result = null;
1941
1942                         Func<List<TSource>, TSource, List<TSource>> intermediate = (list, e) => {
1943                                 list.Add (e); return list;
1944                         };
1945
1946                         Action<IList<List<TSource>>> final = (list) => {
1947                                 int count = 0;
1948
1949                                 for (int i = 0; i < list.Count; i++)
1950                                   count += list[i].Count;
1951
1952                                 result = new TSource[count];
1953                                 int insertIndex = -1;
1954
1955                                 for (int i = 0; i < list.Count; i++)
1956                                   for (int j = 0; j < list[i].Count; j++)
1957                                     result [++insertIndex] = list[i][j];
1958                         };
1959
1960                         ParallelExecuter.ProcessAndAggregate<TSource, List<TSource>> (source.Node,
1961                                                                                       () => new List<TSource> (),
1962                                                                                       intermediate,
1963                                                                                       final);
1964
1965                         return result;
1966                 }
1967
1968                 public static Dictionary<TKey, TSource> ToDictionary<TSource, TKey> (this ParallelQuery<TSource> source,
1969                                                                                      Func<TSource, TKey> keySelector,
1970                                                                                      IEqualityComparer<TKey> comparer)
1971                 {
1972                         return ToDictionary<TSource, TKey, TSource> (source, keySelector, (e) => e, comparer);
1973                 }
1974
1975                 public static Dictionary<TKey, TSource> ToDictionary<TSource, TKey> (this ParallelQuery<TSource> source,
1976                                                                                      Func<TSource, TKey> keySelector)
1977                 {
1978                         return ToDictionary<TSource, TKey, TSource> (source, keySelector, (e) => e, EqualityComparer<TKey>.Default);
1979                 }
1980
1981                 public static Dictionary<TKey, TElement> ToDictionary<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
1982                                                                                                   Func<TSource, TKey> keySelector,
1983                                                                                                   Func<TSource, TElement> elementSelector)
1984                 {
1985                         return ToDictionary<TSource, TKey, TElement> (source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
1986                 }
1987
1988                 public static Dictionary<TKey, TElement> ToDictionary<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
1989                                                                                                   Func<TSource, TKey> keySelector,
1990                                                                                                   Func<TSource, TElement> elementSelector,
1991                                                                                                   IEqualityComparer<TKey> comparer)
1992                 {
1993                         if (source == null)
1994                                 throw new ArgumentNullException ("source");
1995                         if (keySelector == null)
1996                                 throw new ArgumentNullException ("keySelector");
1997                         if (comparer == null)
1998                                 comparer = EqualityComparer<TKey>.Default;
1999                         if (elementSelector == null)
2000                                 throw new ArgumentNullException ("elementSelector");
2001
2002                         return source.Aggregate (() => new Dictionary<TKey, TElement> (comparer),
2003                                                   (d, e) => { d.Add (keySelector (e), elementSelector (e)); return d; },
2004                                                   (d1, d2) => { foreach (var couple in d2) d1.Add (couple.Key, couple.Value); return d1; },
2005                                                   (d) => d);
2006                 }
2007
2008                 public static ILookup<TKey, TSource> ToLookup<TSource, TKey> (this ParallelQuery<TSource> source,
2009                                                                               Func<TSource, TKey> keySelector)
2010                 {
2011                         return ToLookup<TSource, TKey, TSource> (source, keySelector, (e) => e, EqualityComparer<TKey>.Default);
2012                 }
2013
2014                 public static ILookup<TKey, TSource> ToLookup<TSource, TKey> (this ParallelQuery<TSource> source,
2015                                                                               Func<TSource, TKey> keySelector,
2016                                                                               IEqualityComparer<TKey> comparer)
2017                 {
2018                         return ToLookup<TSource, TKey, TSource> (source, keySelector, (e) => e, comparer);
2019                 }
2020
2021                 public static ILookup<TKey, TElement> ToLookup<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
2022                                                                                          Func<TSource, TKey> keySelector,
2023                                                                                          Func<TSource, TElement> elementSelector)
2024                 {
2025                         return ToLookup<TSource, TKey, TElement> (source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
2026                 }
2027
2028                 public static ILookup<TKey, TElement> ToLookup<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
2029                                                                                          Func<TSource, TKey> keySelector,
2030                                                                                          Func<TSource, TElement> elementSelector,
2031                                                                                          IEqualityComparer<TKey> comparer)
2032                 {
2033                         if (source == null)
2034                                 throw new ArgumentNullException ("source");
2035                         if (keySelector == null)
2036                                 throw new ArgumentNullException ("keySelector");
2037                         if (comparer == null)
2038                                 comparer = EqualityComparer<TKey>.Default;
2039                         if (elementSelector == null)
2040                                 throw new ArgumentNullException ("elementSelector");
2041
2042                         ConcurrentLookup<TKey, TElement> lookup = new ConcurrentLookup<TKey, TElement> (comparer);
2043                         source.ForAll ((e) => lookup.Add (keySelector (e), elementSelector (e)));
2044
2045                         return lookup;
2046                 }
2047                 #endregion
2048
2049                 #region Concat
2050                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather than "
2051                                    + "System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() extension method "
2052                                    + "to convert the right data source to System.Linq.ParallelQuery<T>.")]
2053                 public static ParallelQuery<TSource> Concat<TSource>(this ParallelQuery<TSource> first,
2054                                                                      IEnumerable<TSource> second)
2055                 {
2056                         throw new NotSupportedException ();
2057                 }
2058
2059                 public static ParallelQuery<TSource> Concat<TSource> (this ParallelQuery<TSource> first, ParallelQuery<TSource> second)
2060                 {
2061                         return new ParallelQuery<TSource> (new QueryConcatNode<TSource> (first.Node, second.Node));
2062                 }
2063                 #endregion
2064
2065                 #region DefaultIfEmpty
2066                 public static ParallelQuery<TSource> DefaultIfEmpty<TSource> (this ParallelQuery<TSource> source)
2067                 {
2068                         return source.DefaultIfEmpty (default (TSource));
2069                 }
2070
2071                 public static ParallelQuery<TSource> DefaultIfEmpty<TSource> (this ParallelQuery<TSource> source, TSource defaultValue)
2072                 {
2073                         return new ParallelQuery<TSource> (new QueryDefaultEmptyNode<TSource> (source.Node, defaultValue));
2074                 }
2075                 #endregion
2076
2077                 #region First
2078                 public static TSource First<TSource> (this ParallelQuery<TSource> source)
2079                 {
2080                         CancellationTokenSource src = new CancellationTokenSource ();
2081                         IEnumerator<TSource> enumerator = source.WithImplementerToken (src).GetEnumerator ();
2082
2083                         if (enumerator == null || !enumerator.MoveNext ())
2084                                 throw new InvalidOperationException ("source contains no element");
2085
2086                         TSource result = enumerator.Current;
2087                         src.Cancel ();
2088                         enumerator.Dispose ();
2089
2090                         return result;
2091                 }
2092
2093                 public static TSource First<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
2094                 {
2095                         return source.Where (predicate).First ();
2096                 }
2097
2098                 public static TSource FirstOrDefault<TSource> (this ParallelQuery<TSource> source)
2099                 {
2100                         return source.DefaultIfEmpty ().First ();
2101                 }
2102
2103                 public static TSource FirstOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
2104                 {
2105                         return source.Where (predicate).FirstOrDefault ();
2106                 }
2107                 #endregion
2108
2109                 #region Last
2110                 public static TSource Last<TSource> (this ParallelQuery<TSource> source)
2111                 {
2112                         return source.Reverse ().First ();
2113                 }
2114
2115                 public static TSource Last<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
2116                 {
2117                         return source.Reverse ().First (predicate);
2118                 }
2119
2120                 public static TSource LastOrDefault<TSource> (this ParallelQuery<TSource> source)
2121                 {
2122                         return source.Reverse ().FirstOrDefault ();
2123                 }
2124
2125                 public static TSource LastOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
2126                 {
2127                         return source.Reverse ().FirstOrDefault (predicate);
2128                 }
2129                 #endregion
2130
2131                 #region Zip
2132                 public static ParallelQuery<TResult> Zip<TFirst, TSecond, TResult> (this ParallelQuery<TFirst> first,
2133                                                                                       ParallelQuery<TSecond> second,
2134                                                                                       Func<TFirst, TSecond, TResult> resultSelector)
2135                 {
2136                         if (first == null)
2137                                 throw new ArgumentNullException ("first");
2138                         if (second == null)
2139                                 throw new ArgumentNullException ("second");
2140                         if (resultSelector == null)
2141                                 throw new ArgumentNullException ("resultSelector");
2142
2143                         return new ParallelQuery<TResult> (new QueryZipNode<TFirst, TSecond, TResult> (resultSelector, first.Node, second.Node));
2144                 }
2145
2146                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
2147                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
2148                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
2149                 public static ParallelQuery<TResult> Zip<TFirst, TSecond, TResult> (this ParallelQuery<TFirst> first,
2150                                                                                       IEnumerable<TSecond> second,
2151                                                                                       Func<TFirst, TSecond, TResult> resultSelector)
2152                 {
2153                         throw new NotSupportedException ();
2154                 }
2155                 #endregion
2156         }
2157 }
2158 #endif