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                         var source = new CancellationTokenSource ();
 555                         var zip = new QueryZipNode<TSource, TSource, bool> (comparer.Equals, first.Node, second.Node) { Strict = true };
 556                         var innerQuery = new ParallelQuery<bool> (zip).WithImplementerToken (source);
 557
 558                         bool result = true;
 559
 560                         try {
 561                                 innerQuery.ForAll (value => {
 562                                         if (!value) {
 563                                                 result = false;
 564                                                 source.Cancel ();
 565                                         }
 566                                 });
 567                         } catch (AggregateException ex) {
 568                                 if (ex.InnerException is QueryZipException)
 569                                         return false;
 570                                 else
 571                                         throw ex;
 572                         } catch (OperationCanceledException e) {
 573                                 if (e.CancellationToken != source.Token)
 574                                         throw e;
 575                         }
 576
 577                         return result;
 578                 }
 579
 580                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 581                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 582                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 583                 public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first, IEnumerable<TSource> second)
 584                 {
 585                         throw new NotSupportedException ();
 586                 }
 587
 588                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 589                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 590                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 591                 public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first,
 592                                                               IEnumerable<TSource> second,
 593                                                               IEqualityComparer<TSource> comparer)
 594                 {
 595                         throw new NotSupportedException ();
 596                 }
 597
 598                 #endregion
 599
 600                 #region GroupBy
 601                 public static ParallelQuery<IGrouping<TKey, TSource>> GroupBy<TSource, TKey> (this ParallelQuery<TSource> source,
 602                                                                                               Func<TSource, TKey> keySelector)
 603                 {
 604                         return source.GroupBy (keySelector, EqualityComparer<TKey>.Default);
 605                 }
 606
 607                 public static ParallelQuery<IGrouping<TKey, TSource>> GroupBy<TSource, TKey> (this ParallelQuery<TSource> source,
 608                                                                                               Func<TSource, TKey> keySelector,
 609                                                                                               IEqualityComparer<TKey> comparer)
 610                 {
 611                         return source.GroupBy (keySelector, new Identity<TSource> ().Apply, comparer);
 612                 }
 613
 614                 public static ParallelQuery<IGrouping<TKey, TElement>> GroupBy<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
 615                                                                                                          Func<TSource, TKey> keySelector,
 616                                                                                                          Func<TSource, TElement> elementSelector)
 617                 {
 618                         return source.GroupBy (keySelector, elementSelector, EqualityComparer<TKey>.Default);
 619                 }
 620
 621                 public static ParallelQuery<TResult> GroupBy<TSource, TKey, TResult> (this ParallelQuery<TSource> source,
 622                                                                                       Func<TSource, TKey> keySelector,
 623                                                                                       Func<TKey, IEnumerable<TSource>, TResult> resultSelector)
 624                 {
 625                         return source.GroupBy (keySelector)
 626                                 .Select ((g) => resultSelector (g.Key, (IEnumerable<TSource>)g));
 627                 }
 628
 629                 public static ParallelQuery<IGrouping<TKey, TElement>> GroupBy<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
 630                                                                                                          Func<TSource, TKey> keySelector,
 631                                                                                                          Func<TSource, TElement> elementSelector,
 632                                                                                                          IEqualityComparer<TKey> comparer)
 633                 {
 634                         if (source == null)
 635                                 throw new ArgumentNullException ("source");
 636                         if (keySelector == null)
 637                                 throw new ArgumentNullException ("keySelector");
 638                         if (elementSelector == null)
 639                                 throw new ArgumentNullException ("elementSelector");
 640                         if (comparer == null)
 641                                 comparer = EqualityComparer<TKey>.Default;
 642
 643                         return new ParallelQuery<IGrouping<TKey, TElement>> (new QueryGroupByNode<TSource, TKey, TElement> (source.Node, keySelector, elementSelector, comparer));
 644                 }
 645
 646                 public static ParallelQuery<TResult> GroupBy<TSource, TKey, TElement, TResult> (this ParallelQuery<TSource> source,
 647                                                                                                 Func<TSource, TKey> keySelector,
 648                                                                                                 Func<TSource, TElement> elementSelector,
 649                                                                                                 Func<TKey, IEnumerable<TElement>, TResult> resultSelector)
 650                 {
 651                         return source.GroupBy (keySelector, elementSelector)
 652                                 .Select ((g) => resultSelector (g.Key, (IEnumerable<TElement>)g));
 653                 }
 654
 655                 public static ParallelQuery<TResult> GroupBy<TSource, TKey, TResult> (this ParallelQuery<TSource> source,
 656                                                                                       Func<TSource, TKey> keySelector,
 657                                                                                       Func<TKey, IEnumerable<TSource>, TResult> resultSelector,
 658                                                                                       IEqualityComparer<TKey> comparer)
 659                 {
 660                         return source.GroupBy (keySelector, comparer)
 661                                 .Select ((g) => resultSelector (g.Key, (IEnumerable<TSource>)g));
 662                 }
 663
 664                 public static ParallelQuery<TResult> GroupBy<TSource, TKey, TElement, TResult> (this ParallelQuery<TSource> source,
 665                                                                                                 Func<TSource, TKey> keySelector,
 666                                                                                                 Func<TSource, TElement> elementSelector,
 667                                                                                                 Func<TKey, IEnumerable<TElement>, TResult> resultSelector,
 668                                                                                                 IEqualityComparer<TKey> comparer)
 669                 {
 670                         return source.GroupBy (keySelector, elementSelector, comparer)
 671                                 .Select ((g) => resultSelector (g.Key, (IEnumerable<TElement>)g));
 672                 }
 673                 #endregion
 674
 675                 #region GroupJoin
 676                 public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 677                                                                                                ParallelQuery<TInner> inner,
 678                                                                                                Func<TOuter, TKey> outerKeySelector,
 679                                                                                                Func<TInner, TKey> innerKeySelector,
 680                                                                                                Func<TOuter, IEnumerable<TInner>, TResult> resultSelector)
 681                 {
 682                         return outer.GroupJoin (inner, outerKeySelector, innerKeySelector, resultSelector, EqualityComparer<TKey>.Default);
 683                 }
 684
 685                 public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 686                                                                                                ParallelQuery<TInner> inner,
 687                                                                                                Func<TOuter, TKey> outerKeySelector,
 688                                                                                                Func<TInner, TKey> innerKeySelector,
 689                                                                                                Func<TOuter, IEnumerable<TInner>, TResult> resultSelector,
 690                                                                                                IEqualityComparer<TKey> comparer)
 691                 {
 692                         return outer.Join (inner.GroupBy (innerKeySelector, (e) => e), outerKeySelector, (e) => e.Key, (e1, e2) => resultSelector (e1, e2), comparer);
 693                 }
 694
 695                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 696                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 697                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 698                 public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 699                                                                                                IEnumerable<TInner> inner,
 700                                                                                                Func<TOuter, TKey> outerKeySelector,
 701                                                                                                Func<TInner, TKey> innerKeySelector,
 702                                                                                                Func<TOuter, IEnumerable<TInner>, TResult> resultSelector)
 703                 {
 704                         throw new NotSupportedException ();
 705                 }
 706
 707                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 708                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 709                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 710                 public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 711                                                                                                IEnumerable<TInner> inner,
 712                                                                                                Func<TOuter, TKey> outerKeySelector,
 713                                                                                                Func<TInner, TKey> innerKeySelector,
 714                                                                                                Func<TOuter, IEnumerable<TInner>, TResult> resultSelector,
 715                                                                                                IEqualityComparer<TKey> comparer)
 716                 {
 717                         throw new NotSupportedException ();
 718                 }
 719                 #endregion
 720
 721                 #region ElementAt
 722                 public static TSource ElementAt<TSource> (this ParallelQuery<TSource> source, int index)
 723                 {
 724                         if (source == null)
 725                                 throw new ArgumentNullException ("source");
 726                         if (index < 0)
 727                                 throw new ArgumentOutOfRangeException ("index");
 728                         if (index == 0) {
 729                                 try {
 730                                         return source.First ();
 731                                 } catch (InvalidOperationException) {
 732                                         throw new ArgumentOutOfRangeException ("index");
 733                                 }
 734                         }
 735
 736                         TSource result = default (TSource);
 737
 738                         ParallelQuery<TSource> innerQuery = source.Where ((e, i) => i == index);
 739
 740                         try {
 741                                 result = innerQuery.First ();
 742                         } catch (InvalidOperationException) {
 743                                 throw new ArgumentOutOfRangeException ("index");
 744                         }
 745
 746                         return result;
 747                 }
 748
 749                 public static TSource ElementAtOrDefault<TSource> (this ParallelQuery<TSource> source, int index)
 750                 {
 751                         if (source == null)
 752                                 throw new ArgumentNullException ("source");
 753
 754                         try {
 755                                 return source.ElementAt (index);
 756                         } catch (ArgumentOutOfRangeException) {
 757                                 return default (TSource);
 758                         }
 759                 }
 760                 #endregion
 761
 762                 #region Intersect
 763                 public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first,
 764                                                                          ParallelQuery<TSource> second)
 765                 {
 766                         return Intersect<TSource> (first, second, EqualityComparer<TSource>.Default);
 767                 }
 768
 769                 public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first,
 770                                                                          ParallelQuery<TSource> second,
 771                                                                          IEqualityComparer<TSource> comparer)
 772                 {
 773                         if (first == null)
 774                                 throw new ArgumentNullException ("first");
 775                         if (second == null)
 776                                 throw new ArgumentNullException ("second");
 777                         if (comparer == null)
 778                                 comparer = EqualityComparer<TSource>.Default;
 779
 780                         return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Intersect, comparer, first.Node, second.Node));
 781                 }
 782
 783                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 784                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 785                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 786                 public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first, IEnumerable<TSource> second)
 787                 {
 788                         throw new NotSupportedException ();
 789                 }
 790
 791                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 792                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 793                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 794                 public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first,
 795                                                                          IEnumerable<TSource> second,
 796                                                                          IEqualityComparer<TSource> comparer)
 797                 {
 798                         throw new NotSupportedException ();
 799                 }
 800                 #endregion
 801
 802                 #region Join
 803                 public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 804                                                                                           ParallelQuery<TInner> inner,
 805                                                                                           Func<TOuter, TKey> outerKeySelector,
 806                                                                                           Func<TInner, TKey> innerKeySelector,
 807                                                                                           Func<TOuter, TInner, TResult> resultSelector)
 808                 {
 809                         return outer.Join (inner, outerKeySelector, innerKeySelector, resultSelector, EqualityComparer<TKey>.Default);
 810                 }
 811
 812                 public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 813                                                                                           ParallelQuery<TInner> inner,
 814                                                                                           Func<TOuter, TKey> outerKeySelector,
 815                                                                                           Func<TInner, TKey> innerKeySelector,
 816                                                                                           Func<TOuter, TInner, TResult> resultSelector,
 817                                                                                           IEqualityComparer<TKey> comparer)
 818                 {
 819                         return new ParallelQuery<TResult> (new QueryJoinNode<TOuter, TInner, TKey, TResult> (outer.Node, inner.Node, outerKeySelector, innerKeySelector, resultSelector, comparer));
 820                 }
 821
 822                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 823                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 824                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 825                 public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 826                                                                                           IEnumerable<TInner> inner,
 827                                                                                           Func<TOuter, TKey> outerKeySelector,
 828                                                                                           Func<TInner, TKey> innerKeySelector,
 829                                                                                           Func<TOuter, TInner, TResult> resultSelector)
 830                 {
 831                         throw new NotSupportedException ();
 832                 }
 833
 834                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 835                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 836                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 837                 public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
 838                                                                                           IEnumerable<TInner> inner,
 839                                                                                           Func<TOuter, TKey> outerKeySelector,
 840                                                                                           Func<TInner, TKey> innerKeySelector,
 841                                                                                           Func<TOuter, TInner, TResult> resultSelector,
 842                                                                                           IEqualityComparer<TKey> comparer)
 843                 {
 844                         throw new NotSupportedException ();
 845                 }
 846                 #endregion
 847
 848                 #region Except
 849                 public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
 850                                                                       ParallelQuery<TSource> second)
 851                 {
 852                         return Except<TSource> (first, second, EqualityComparer<TSource>.Default);
 853                 }
 854
 855                 public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
 856                                                                       ParallelQuery<TSource> second,
 857                                                                       IEqualityComparer<TSource> comparer)
 858                 {
 859                         if (first == null)
 860                                 throw new ArgumentNullException ("first");
 861                         if (second == null)
 862                                 throw new ArgumentNullException ("second");
 863                         if (comparer == null)
 864                                 comparer = EqualityComparer<TSource>.Default;
 865
 866                         return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Except,
 867                                                                                       comparer, first.Node, second.Node));
 868                 }
 869
 870                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 871                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 872                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 873                 public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
 874                                                                       IEnumerable<TSource> second)
 875                 {
 876                         throw new NotSupportedException ();
 877                 }
 878
 879                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 880                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 881                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 882                 public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
 883                                                                       IEnumerable<TSource> second,
 884                                                                       IEqualityComparer<TSource> comparer)
 885                 {
 886                         throw new NotSupportedException ();
 887                 }
 888                 #endregion
 889
 890                 #region Distinct
 891                 public static ParallelQuery<TSource> Distinct<TSource> (this ParallelQuery<TSource> source)
 892                 {
 893                         return Distinct<TSource> (source, EqualityComparer<TSource>.Default);
 894                 }
 895
 896                 public static ParallelQuery<TSource> Distinct<TSource> (this ParallelQuery<TSource> source, IEqualityComparer<TSource> comparer)
 897                 {
 898                         if (source == null)
 899                                 throw new ArgumentNullException ("source");
 900                         if (comparer == null)
 901                                 comparer = EqualityComparer<TSource>.Default;
 902
 903                         return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Distinct, comparer,
 904                                                                                       source.Node, null));
 905                 }
 906                 #endregion
 907
 908                 #region Union
 909                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 910                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 911                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 912                 public static ParallelQuery<TSource> Union<TSource> (this ParallelQuery<TSource> first,
 913                                                                      IEnumerable<TSource> second)
 914                 {
 915                         throw new NotSupportedException ();
 916                 }
 917
 918                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
 919                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
 920                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
 921                 public static ParallelQuery<TSource> Union<TSource>(this ParallelQuery<TSource> first,
 922                                                                     IEnumerable<TSource> second,
 923                                                                     IEqualityComparer<TSource> comparer)
 924                 {
 925                         throw new NotSupportedException ();
 926                 }
 927
 928                 public static ParallelQuery<TSource> Union<TSource> (this ParallelQuery<TSource> first,
 929                                                                      ParallelQuery<TSource> second)
 930                 {
 931                         return first.Union (second, EqualityComparer<TSource>.Default);
 932                 }
 933
 934                 public static ParallelQuery<TSource> Union<TSource> (this ParallelQuery<TSource> first,
 935                                                                      ParallelQuery<TSource> second,
 936                                                                      IEqualityComparer<TSource> comparer)
 937                 {
 938                         if (first == null)
 939                                 throw new ArgumentNullException ("first");
 940                         if (second == null)
 941                                 throw new ArgumentNullException ("second");
 942                         if (comparer == null)
 943                                 comparer = EqualityComparer<TSource>.Default;
 944
 945                         return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Union, comparer, first.Node, second.Node));
 946                 }
 947                 #endregion
 948
 949                 #region Take
 950                 public static ParallelQuery<TSource> Take<TSource> (this ParallelQuery<TSource> source, int count)
 951                 {
 952                         if (source == null)
 953                                 throw new ArgumentNullException ("source");
 954
 955                         return new ParallelQuery<TSource> (new QueryHeadWorkerNode<TSource> (source.Node, count));
 956                 }
 957
 958                 public static ParallelQuery<TSource> TakeWhile<TSource> (this ParallelQuery<TSource> source,
 959                                                                          Func<TSource, bool> predicate)
 960                 {
 961                         if (source == null)
 962                                 throw new ArgumentNullException ("source");
 963                         if (predicate == null)
 964                                 throw new ArgumentNullException ("predicate");
 965
 966                         return new ParallelQuery<TSource> (new QueryHeadWorkerNode<TSource> (source.Node, (e, _) => predicate (e), false));
 967                 }
 968
 969                 public static ParallelQuery<TSource> TakeWhile<TSource> (this ParallelQuery<TSource> source,
 970                                                                          Func<TSource, int, bool> predicate)
 971                 {
 972                         if (source == null)
 973                                 throw new ArgumentNullException ("source");
 974                         if (predicate == null)
 975                                 throw new ArgumentNullException ("predicate");
 976
 977                         return new ParallelQuery<TSource> (new QueryHeadWorkerNode<TSource> (source.Node, predicate, true));
 978                 }
 979                 #endregion
 980
 981                 #region Skip
 982                 public static ParallelQuery<TSource> Skip<TSource> (this ParallelQuery<TSource> source, int count)
 983                 {
 984                         if (source == null)
 985                                 throw new ArgumentNullException ("source");
 986
 987                         return source.Node.IsOrdered () ?
 988                                 source.Where ((e, i) => i >= count) :
 989                                 source.Where ((e) => count < 0 || Interlocked.Decrement (ref count) < 0);
 990
 991                 }
 992
 993                 public static ParallelQuery<TSource> SkipWhile<TSource> (this ParallelQuery<TSource> source,
 994                                                                          Func<TSource, bool> predicate)
 995                 {
 996                         if (source == null)
 997                                 throw new ArgumentNullException ("source");
 998                         if (predicate == null)
 999                                 throw new ArgumentNullException ("predicate");
1000
1001                         return source.Node.IsOrdered () ?
1002                                 source.SkipWhile ((e, i) => predicate (e)) :
1003                                 source.Where ((e) => !predicate (e));
1004                 }
1005
1006                 public static ParallelQuery<TSource> SkipWhile<TSource> (this ParallelQuery<TSource> source,
1007                                                                          Func<TSource, int, bool> predicate)
1008                 {
1009                         if (source == null)
1010                                 throw new ArgumentNullException ("source");
1011                         if (predicate == null)
1012                                 throw new ArgumentNullException ("predicate");
1013
1014                         int indexCache = int.MaxValue;
1015
1016                         return source.Where ((e, i) => i >= indexCache || (!predicate (e, i) && (indexCache = i) == i));
1017                 }
1018                 #endregion
1019
1020                 #region Single
1021                 static TSource SingleInternal<TSource> (this ParallelQuery<TSource> source, params TSource[] init)
1022                 {
1023                         TSource result = default(TSource);
1024                         bool hasValue = false;
1025
1026                         foreach (TSource element in source) {
1027                                 if (hasValue)
1028                                         throw new InvalidOperationException ("The input sequence contains more than one element.");
1029
1030                                 result = element;
1031                                 hasValue = true;
1032                         }
1033
1034                         if (!hasValue && init.Length != 0) {
1035                                 result = init[0];
1036                                 hasValue = true;
1037                         }
1038
1039                         if (!hasValue)
1040                                 throw new InvalidOperationException ("The input sequence is empty.");
1041
1042                         return result;
1043                 }
1044
1045                 public static TSource Single<TSource> (this ParallelQuery<TSource> source)
1046                 {
1047                         if (source == null)
1048                                 throw new ArgumentNullException ("source");
1049
1050                         return SingleInternal<TSource> (source);
1051                 }
1052
1053                 public static TSource Single<TSource> (this ParallelQuery<TSource> source,
1054                                                        Func<TSource, bool> predicate)
1055                 {
1056                         if (source == null)
1057                                 throw new ArgumentNullException ("source");
1058                         if (predicate == null)
1059                                 throw new ArgumentNullException ("predicate");
1060
1061                         return source.Where (predicate).Single ();
1062                 }
1063
1064                 public static TSource SingleOrDefault<TSource> (this ParallelQuery<TSource> source)
1065                 {
1066                         if (source == null)
1067                                 throw new ArgumentNullException ("source");
1068
1069                         return SingleInternal<TSource> (source, default (TSource));
1070                 }
1071
1072                 public static TSource SingleOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
1073                 {
1074                         if (source == null)
1075                                 throw new ArgumentNullException ("source");
1076                         if (predicate == null)
1077                                 throw new ArgumentNullException ("predicate");
1078
1079                         return source.Where (predicate).SingleOrDefault ();
1080                 }
1081                 #endregion
1082
1083                 #region Count
1084                 public static int Count<TSource> (this ParallelQuery<TSource> source)
1085                 {
1086                         if (source == null)
1087                                 throw new ArgumentNullException ("source");
1088
1089                         return source.Aggregate<TSource, int, int> (() => 0,
1090                                                                    (acc, e) => acc + 1,
1091                                                                    (acc1, acc2) => acc1 + acc2,
1092                                                                    (result) => result);
1093                 }
1094
1095                 public static int Count<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
1096                 {
1097                         if (source == null)
1098                                 throw new ArgumentNullException ("source");
1099                         if (predicate == null)
1100                                 throw new ArgumentNullException ("predicate");
1101
1102                         return source.Where (predicate).Count ();
1103                 }
1104
1105                 public static long LongCount<TSource> (this ParallelQuery<TSource> source)
1106                 {
1107                         if (source == null)
1108                                 throw new ArgumentNullException ("source");
1109
1110                         return source.Aggregate<TSource, long, long> (() => 0,
1111                                                                       (acc, e) => acc + 1,
1112                                                                       (acc1, acc2) => acc1 + acc2,
1113                                                                       (result) => result);
1114                 }
1115
1116                 public static long LongCount<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
1117                 {
1118                         if (source == null)
1119                                 throw new ArgumentNullException ("source");
1120                         if (predicate == null)
1121                                 throw new ArgumentNullException ("predicate");
1122
1123                         return source.Where (predicate).LongCount ();
1124                 }
1125                 #endregion
1126
1127                 #region Average
1128                 public static double Average (this ParallelQuery<int> source)
1129                 {
1130                         if (source == null)
1131                                 throw new ArgumentNullException ("source");
1132
1133                         return source.Aggregate (() => new int[2],
1134                                                 (acc, e) => { acc[0] += e; acc[1]++; return acc; },
1135                                                 (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
1136                                                 (acc) => acc[0] / ((double)acc[1]));
1137                 }
1138
1139                 public static double Average (this ParallelQuery<long> source)
1140                 {
1141                         if (source == null)
1142                                 throw new ArgumentNullException ("source");
1143
1144                         return source.Aggregate (() => new long[2],
1145                                                 (acc, e) => { acc[0] += e; acc[1]++; return acc; },
1146                                                 (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
1147                                                 (acc) => acc[0] / ((double)acc[1]));
1148                 }
1149
1150                 public static decimal Average (this ParallelQuery<decimal> source)
1151                 {
1152                         if (source == null)
1153                                 throw new ArgumentNullException ("source");
1154
1155                         return source.Aggregate (() => new decimal[2],
1156                                                 (acc, e) => { acc[0] += e; acc[1]++; return acc; },
1157                                                 (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
1158                                                 (acc) => acc[0] / acc[1]);
1159                 }
1160
1161                 public static double Average (this ParallelQuery<double> source)
1162                 {
1163                         if (source == null)
1164                                 throw new ArgumentNullException ("source");
1165
1166                         return source.Aggregate (() => new double[2],
1167                                                 (acc, e) => { acc[0] += e; acc[1]++; return acc; },
1168                                                 (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
1169                                                 (acc) => acc[0] / ((double)acc[1]));
1170                 }
1171
1172                 public static float Average (this ParallelQuery<float> source)
1173                 {
1174                         if (source == null)
1175                                 throw new ArgumentNullException ("source");
1176
1177                         return source.Aggregate (() => new float[2],
1178                                                 (acc, e) => { acc[0] += e; acc[1]++; return acc; },
1179                                                 (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
1180                                                 (acc) => acc[0] / acc[1]);
1181                 }
1182                 #endregion
1183
1184                 #region More Average
1185                 public static double? Average (this ParallelQuery<int?> source)
1186                 {
1187                         if (source == null)
1188                                 throw new ArgumentNullException ("source");
1189
1190                         return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();;
1191                 }
1192
1193                 public static double? Average (this ParallelQuery<long?> source)
1194                 {
1195                         if (source == null)
1196                                 throw new ArgumentNullException ("source");
1197
1198                         return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
1199                 }
1200
1201                 public static decimal? Average (this ParallelQuery<decimal?> source)
1202                 {
1203                         if (source == null)
1204                                 throw new ArgumentNullException ("source");
1205
1206                         return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
1207                 }
1208
1209                 public static double? Average (this ParallelQuery<double?> source)
1210                 {
1211                         if (source == null)
1212                                 throw new ArgumentNullException ("source");
1213
1214                         return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
1215                 }
1216
1217                 public static float? Average (this ParallelQuery<float?> source)
1218                 {
1219                         if (source == null)
1220                                 throw new ArgumentNullException ("source");
1221
1222                         return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
1223                 }
1224
1225                 public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
1226                 {
1227                         if (source == null)
1228                                 throw new ArgumentNullException ("source");
1229                         if (selector == null)
1230                                 throw new ArgumentNullException ("selector");
1231
1232                         return source.Select (selector).Average ();
1233                 }
1234
1235                 public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
1236                 {
1237                         if (source == null)
1238                                 throw new ArgumentNullException ("source");
1239                         if (selector == null)
1240                                 throw new ArgumentNullException ("selector");
1241
1242                         return source.Select (selector).Average ();
1243                 }
1244
1245                 public static float Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
1246                 {
1247                         if (source == null)
1248                                 throw new ArgumentNullException ("source");
1249                         if (selector == null)
1250                                 throw new ArgumentNullException ("selector");
1251
1252                         return source.Select (selector).Average ();
1253                 }
1254
1255                 public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
1256                 {
1257                         if (source == null)
1258                                 throw new ArgumentNullException ("source");
1259                         if (selector == null)
1260                                 throw new ArgumentNullException ("selector");
1261
1262                         return source.Select (selector).Average ();
1263                 }
1264
1265                 public static decimal Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
1266                 {
1267                         if (source == null)
1268                                 throw new ArgumentNullException ("source");
1269                         if (selector == null)
1270                                 throw new ArgumentNullException ("selector");
1271
1272                         return source.Select (selector).Average ();
1273                 }
1274
1275                 public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
1276                 {
1277                         if (source == null)
1278                                 throw new ArgumentNullException ("source");
1279                         if (selector == null)
1280                                 throw new ArgumentNullException ("selector");
1281
1282                         return source.Select (selector).Average ();
1283                 }
1284
1285                 public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
1286                 {
1287                         if (source == null)
1288                                 throw new ArgumentNullException ("source");
1289                         if (selector == null)
1290                                 throw new ArgumentNullException ("selector");
1291
1292                         return source.Select (selector).Average ();
1293                 }
1294
1295                 public static float? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
1296                 {
1297                         if (source == null)
1298                                 throw new ArgumentNullException ("source");
1299                         if (selector == null)
1300                                 throw new ArgumentNullException ("selector");
1301
1302                         return source.Select (selector).Average ();
1303                 }
1304
1305                 public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
1306                 {
1307                         if (source == null)
1308                                 throw new ArgumentNullException ("source");
1309                         if (selector == null)
1310                                 throw new ArgumentNullException ("selector");
1311
1312                         return source.Select (selector).Average ();
1313                 }
1314
1315                 public static decimal? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
1316                 {
1317                         if (source == null)
1318                                 throw new ArgumentNullException ("source");
1319                         if (selector == null)
1320                                 throw new ArgumentNullException ("selector");
1321
1322                         return source.Select (selector).Average ();
1323                 }
1324                 #endregion
1325
1326                 #region Sum
1327                 public static int Sum (this ParallelQuery<int> source)
1328                 {
1329                         if (source == null)
1330                                 throw new ArgumentNullException ("source");
1331
1332                         return source.Aggregate (0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
1333                 }
1334
1335                 public static long Sum (this ParallelQuery<long> source)
1336                 {
1337                         if (source == null)
1338                                 throw new ArgumentNullException ("source");
1339
1340                         return source.Aggregate ((long)0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
1341                 }
1342
1343                 public static float Sum (this ParallelQuery<float> source)
1344                 {
1345                         if (source == null)
1346                                 throw new ArgumentNullException ("source");
1347
1348                         return source.Aggregate (0.0f, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
1349                 }
1350
1351                 public static double Sum (this ParallelQuery<double> source)
1352                 {
1353                         if (source == null)
1354                                 throw new ArgumentNullException ("source");
1355
1356                         return source.Aggregate (0.0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
1357                 }
1358
1359                 public static decimal Sum (this ParallelQuery<decimal> source)
1360                 {
1361                         if (source == null)
1362                                 throw new ArgumentNullException ("source");
1363
1364                         return source.Aggregate ((decimal)0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
1365                 }
1366
1367                 public static int? Sum (this ParallelQuery<int?> source)
1368                 {
1369                         return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
1370                 }
1371
1372                 public static long? Sum (this ParallelQuery<long?> source)
1373                 {
1374                         if (source == null)
1375                                 throw new ArgumentNullException ("source");
1376
1377                         return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
1378                 }
1379
1380                 public static float? Sum (this ParallelQuery<float?> source)
1381                 {
1382                         if (source == null)
1383                                 throw new ArgumentNullException ("source");
1384
1385                         return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
1386                 }
1387
1388                 public static double? Sum (this ParallelQuery<double?> source)
1389                 {
1390                         if (source == null)
1391                                 throw new ArgumentNullException ("source");
1392
1393                         return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
1394                 }
1395
1396                 public static decimal? Sum (this ParallelQuery<decimal?> source)
1397                 {
1398                         if (source == null)
1399                                 throw new ArgumentNullException ("source");
1400
1401                         return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
1402                 }
1403
1404                 public static int Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
1405                 {
1406                         if (source == null)
1407                                 throw new ArgumentNullException ("source");
1408                         if (selector == null)
1409                                 throw new ArgumentNullException ("selector");
1410
1411                         return source.Select (selector).Sum ();
1412                 }
1413
1414                 public static long Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
1415                 {
1416                         if (source == null)
1417                                 throw new ArgumentNullException ("source");
1418                         if (selector == null)
1419                                 throw new ArgumentNullException ("selector");
1420
1421                         return source.Select (selector).Sum ();
1422                 }
1423
1424                 public static decimal Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
1425                 {
1426                         if (source == null)
1427                                 throw new ArgumentNullException ("source");
1428                         if (selector == null)
1429                                 throw new ArgumentNullException ("selector");
1430
1431                         return source.Select (selector).Sum ();
1432                 }
1433
1434                 public static float Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
1435                 {
1436                         if (source == null)
1437                                 throw new ArgumentNullException ("source");
1438                         if (selector == null)
1439                                 throw new ArgumentNullException ("selector");
1440
1441                         return source.Select (selector).Sum ();
1442                 }
1443
1444                 public static double Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
1445                 {
1446                         if (source == null)
1447                                 throw new ArgumentNullException ("source");
1448                         if (selector == null)
1449                                 throw new ArgumentNullException ("selector");
1450
1451                         return source.Select (selector).Sum ();
1452                 }
1453
1454                 public static int? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
1455                 {
1456                         if (source == null)
1457                                 throw new ArgumentNullException ("source");
1458                         if (selector == null)
1459                                 throw new ArgumentNullException ("selector");
1460
1461                         return source.Select (selector).Sum ();
1462                 }
1463
1464                 public static long? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
1465                 {
1466                         if (source == null)
1467                                 throw new ArgumentNullException ("source");
1468                         if (selector == null)
1469                                 throw new ArgumentNullException ("selector");
1470
1471                         return source.Select (selector).Sum ();
1472                 }
1473
1474                 public static decimal? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
1475                 {
1476                         if (source == null)
1477                                 throw new ArgumentNullException ("source");
1478                         if (selector == null)
1479                                 throw new ArgumentNullException ("selector");
1480
1481                         return source.Select (selector).Sum ();
1482                 }
1483
1484                 public static float? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
1485                 {
1486                         if (source == null)
1487                                 throw new ArgumentNullException ("source");
1488                         if (selector == null)
1489                                 throw new ArgumentNullException ("selector");
1490
1491                         return source.Select (selector).Sum ();
1492                 }
1493
1494                 public static double? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
1495                 {
1496                         if (source == null)
1497                                 throw new ArgumentNullException ("source");
1498                         if (selector == null)
1499                                 throw new ArgumentNullException ("selector");
1500
1501                         return source.Select (selector).Sum ();
1502                 }
1503                 #endregion
1504
1505                 #region Min-Max
1506                 static T BestOrder<T> (ParallelQuery<T> source, Func<T, T, bool> bestSelector, T seed)
1507                 {
1508                         if (source == null)
1509                                 throw new ArgumentNullException ("source");
1510
1511                         T best = seed;
1512
1513                         best = source.Aggregate (() => seed,
1514                                                 (first, second) => (bestSelector(first, second)) ? first : second,
1515                                                 (first, second) => (bestSelector(first, second)) ? first : second,
1516                                                 (e) => e);
1517                         return best;
1518                 }
1519
1520                 public static int Min (this ParallelQuery<int> source)
1521                 {
1522                         return BestOrder (source, (first, second) => first < second, int.MaxValue);
1523                 }
1524
1525                 public static long Min (this ParallelQuery<long> source)
1526                 {
1527                         return BestOrder (source, (first, second) => first < second, long.MaxValue);
1528                 }
1529
1530                 public static float Min (this ParallelQuery<float> source)
1531                 {
1532                         return BestOrder (source, (first, second) => first < second, float.MaxValue);
1533                 }
1534
1535                 public static double Min (this ParallelQuery<double> source)
1536                 {
1537                         return BestOrder (source, (first, second) => first < second, double.MaxValue);
1538                 }
1539
1540                 public static decimal Min (this ParallelQuery<decimal> source)
1541                 {
1542                         return BestOrder (source, (first, second) => first < second, decimal.MaxValue);
1543                 }
1544
1545                 public static TSource Min<TSource> (this ParallelQuery<TSource> source)
1546                 {
1547                         IComparer<TSource> comparer = Comparer<TSource>.Default;
1548
1549                         return BestOrder (source, (first, second) => comparer.Compare (first, second) < 0, default (TSource));
1550                 }
1551
1552                 public static TResult Min<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> selector)
1553                 {
1554                         if (source == null)
1555                                 throw new ArgumentNullException ("source");
1556                         if (selector == null)
1557                                 throw new ArgumentNullException ("selector");
1558
1559                         return source.Select (selector).Min ();
1560                 }
1561
1562                 public static int? Min (this ParallelQuery<int?> source)
1563                 {
1564                         if (source == null)
1565                                 throw new ArgumentNullException ("source");
1566
1567                         return source.Select ((e) => e.HasValue ? e.Value : int.MaxValue).Min ();
1568                 }
1569
1570                 public static long? Min (this ParallelQuery<long?> source)
1571                 {
1572                         if (source == null)
1573                                 throw new ArgumentNullException ("source");
1574
1575                         return source.Select ((e) => e.HasValue ? e.Value : long.MaxValue).Min ();
1576                 }
1577
1578                 public static float? Min (this ParallelQuery<float?> source)
1579                 {
1580                         if (source == null)
1581                                 throw new ArgumentNullException ("source");
1582
1583                         return source.Select ((e) => e.HasValue ? e.Value : float.MaxValue).Min ();
1584                 }
1585
1586                 public static double? Min (this ParallelQuery<double?> source)
1587                 {
1588                         if (source == null)
1589                                 throw new ArgumentNullException ("source");
1590
1591                         return source.Select ((e) => e.HasValue ? e.Value : double.MaxValue).Min ();
1592                 }
1593
1594                 public static decimal? Min (this ParallelQuery<decimal?> source)
1595                 {
1596                         if (source == null)
1597                                 throw new ArgumentNullException ("source");
1598
1599                         return source.Select ((e) => e.HasValue ? e.Value : decimal.MaxValue).Min ();
1600                 }
1601
1602                 public static int Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
1603                 {
1604                         if (source == null)
1605                                 throw new ArgumentNullException ("source");
1606                         if (selector == null)
1607                                 throw new ArgumentNullException ("selector");
1608
1609                         return source.Select (selector).Min ();
1610                 }
1611
1612                 public static long Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
1613                 {
1614                         if (source == null)
1615                                 throw new ArgumentNullException ("source");
1616                         if (selector == null)
1617                                 throw new ArgumentNullException ("selector");
1618
1619                         return source.Select (selector).Min ();
1620                 }
1621
1622                 public static float Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
1623                 {
1624                         if (source == null)
1625                                 throw new ArgumentNullException ("source");
1626                         if (selector == null)
1627                                 throw new ArgumentNullException ("selector");
1628
1629                         return source.Select (selector).Min ();
1630                 }
1631
1632                 public static double Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
1633                 {
1634                         if (source == null)
1635                                 throw new ArgumentNullException ("source");
1636                         if (selector == null)
1637                                 throw new ArgumentNullException ("selector");
1638
1639                         return source.Select (selector).Min ();
1640                 }
1641
1642                 public static decimal Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
1643                 {
1644                         if (source == null)
1645                                 throw new ArgumentNullException ("source");
1646                         if (selector == null)
1647                                 throw new ArgumentNullException ("selector");
1648
1649                         return source.Select (selector).Min ();
1650                 }
1651
1652                 public static int? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
1653                 {
1654                         if (source == null)
1655                                 throw new ArgumentNullException ("source");
1656                         if (selector == null)
1657                                 throw new ArgumentNullException ("selector");
1658
1659                         return source.Select (selector).Min ();
1660                 }
1661
1662                 public static long? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
1663                 {
1664                         if (source == null)
1665                                 throw new ArgumentNullException ("source");
1666                         if (selector == null)
1667                                 throw new ArgumentNullException ("selector");
1668
1669                         return source.Select (selector).Min ();
1670                 }
1671
1672                 public static float? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
1673                 {
1674                         if (source == null)
1675                                 throw new ArgumentNullException ("source");
1676                         if (selector == null)
1677                                 throw new ArgumentNullException ("selector");
1678
1679                         return source.Select (selector).Min ();
1680                 }
1681
1682                 public static double? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
1683                 {
1684                         if (source == null)
1685                                 throw new ArgumentNullException ("source");
1686                         if (selector == null)
1687                                 throw new ArgumentNullException ("selector");
1688
1689                         return source.Select (selector).Min ();
1690                 }
1691
1692                 public static decimal? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
1693                 {
1694                         if (source == null)
1695                                 throw new ArgumentNullException ("source");
1696                         if (selector == null)
1697                                 throw new ArgumentNullException ("selector");
1698
1699                         return source.Select (selector).Min ();
1700                 }
1701
1702                 public static int Max (this ParallelQuery<int> source)
1703                 {
1704                         return BestOrder (source, (first, second) => first > second, int.MinValue);
1705                 }
1706
1707                 public static long Max(this ParallelQuery<long> source)
1708                 {
1709                         return BestOrder(source, (first, second) => first > second, long.MinValue);
1710                 }
1711
1712                 public static float Max (this ParallelQuery<float> source)
1713                 {
1714                         return BestOrder(source, (first, second) => first > second, float.MinValue);
1715                 }
1716
1717                 public static double Max (this ParallelQuery<double> source)
1718                 {
1719                         return BestOrder(source, (first, second) => first > second, double.MinValue);
1720                 }
1721
1722                 public static decimal Max (this ParallelQuery<decimal> source)
1723                 {
1724                         return BestOrder(source, (first, second) => first > second, decimal.MinValue);
1725                 }
1726
1727                 public static TSource Max<TSource> (this ParallelQuery<TSource> source)
1728                 {
1729                         IComparer<TSource> comparer = Comparer<TSource>.Default;
1730
1731                         return BestOrder (source, (first, second) => comparer.Compare (first, second) > 0, default (TSource));
1732                 }
1733
1734                 public static TResult Max<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> selector)
1735                 {
1736                         if (source == null)
1737                                 throw new ArgumentNullException ("source");
1738                         if (selector == null)
1739                                 throw new ArgumentNullException ("selector");
1740
1741                         return source.Select (selector).Max ();
1742                 }
1743
1744                 public static int? Max (this ParallelQuery<int?> source)
1745                 {
1746                         if (source == null)
1747                                 throw new ArgumentNullException ("source");
1748
1749                         return source.Select ((e) => e.HasValue ? e.Value : int.MinValue).Max ();
1750                 }
1751
1752                 public static long? Max (this ParallelQuery<long?> source)
1753                 {
1754                         if (source == null)
1755                                 throw new ArgumentNullException ("source");
1756
1757                         return source.Select ((e) => e.HasValue ? e.Value : long.MinValue).Max ();
1758                 }
1759
1760                 public static float? Max (this ParallelQuery<float?> source)
1761                 {
1762                         if (source == null)
1763                                 throw new ArgumentNullException ("source");
1764
1765                         return source.Select ((e) => e.HasValue ? e.Value : float.MinValue).Max ();
1766                 }
1767
1768                 public static double? Max (this ParallelQuery<double?> source)
1769                 {
1770                         if (source == null)
1771                                 throw new ArgumentNullException ("source");
1772
1773                         return source.Select ((e) => e.HasValue ? e.Value : double.MinValue).Max ();
1774                 }
1775
1776                 public static decimal? Max (this ParallelQuery<decimal?> source)
1777                 {
1778                         if (source == null)
1779                                 throw new ArgumentNullException ("source");
1780
1781                         return source.Select ((e) => e.HasValue ? e.Value : decimal.MinValue).Max ();
1782                 }
1783
1784                 public static int Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
1785                 {
1786                         if (source == null)
1787                                 throw new ArgumentNullException ("source");
1788                         if (selector == null)
1789                                 throw new ArgumentNullException ("selector");
1790
1791                         return source.Select (selector).Max ();
1792                 }
1793
1794                 public static long Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
1795                 {
1796                         if (source == null)
1797                                 throw new ArgumentNullException ("source");
1798                         if (selector == null)
1799                                 throw new ArgumentNullException ("selector");
1800
1801                         return source.Select (selector).Max ();
1802                 }
1803
1804                 public static float Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
1805                 {
1806                         if (source == null)
1807                                 throw new ArgumentNullException ("source");
1808                         if (selector == null)
1809                                 throw new ArgumentNullException ("selector");
1810
1811                         return source.Select (selector).Max ();
1812                 }
1813
1814                 public static double Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
1815                 {
1816                         if (source == null)
1817                                 throw new ArgumentNullException ("source");
1818                         if (selector == null)
1819                                 throw new ArgumentNullException ("selector");
1820
1821                         return source.Select (selector).Max ();
1822                 }
1823
1824                 public static decimal Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
1825                 {
1826                         if (source == null)
1827                                 throw new ArgumentNullException ("source");
1828                         if (selector == null)
1829                                 throw new ArgumentNullException ("selector");
1830
1831                         return source.Select (selector).Max ();
1832                 }
1833
1834                 public static int? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
1835                 {
1836                         if (source == null)
1837                                 throw new ArgumentNullException ("source");
1838                         if (selector == null)
1839                                 throw new ArgumentNullException ("selector");
1840
1841                         return source.Select (selector).Max ();
1842                 }
1843
1844                 public static long? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
1845                 {
1846                         if (source == null)
1847                                 throw new ArgumentNullException ("source");
1848                         if (selector == null)
1849                                 throw new ArgumentNullException ("selector");
1850
1851                         return source.Select (selector).Max ();
1852                 }
1853
1854                 public static float? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
1855                 {
1856                         if (source == null)
1857                                 throw new ArgumentNullException ("source");
1858                         if (selector == null)
1859                                 throw new ArgumentNullException ("selector");
1860
1861                         return source.Select (selector).Max ();
1862                 }
1863
1864                 public static double? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
1865                 {
1866                         if (source == null)
1867                                 throw new ArgumentNullException ("source");
1868                         if (selector == null)
1869                                 throw new ArgumentNullException ("selector");
1870
1871                         return source.Select (selector).Max ();
1872                 }
1873
1874                 public static decimal? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
1875                 {
1876                         if (source == null)
1877                                 throw new ArgumentNullException ("source");
1878                         if (selector == null)
1879                                 throw new ArgumentNullException ("selector");
1880
1881                         return source.Select (selector).Max ();
1882                 }
1883                 #endregion
1884
1885                 #region Cast / OfType
1886                 public static ParallelQuery<TResult> Cast<TResult> (this ParallelQuery source)
1887                 {
1888                         if (source == null)
1889                                 throw new ArgumentNullException ("source");
1890
1891                         return source.TypedQuery.Select ((e) => (TResult)e);
1892                 }
1893
1894                 public static ParallelQuery<TResult> OfType<TResult> (this ParallelQuery source)
1895                 {
1896                         if (source == null)
1897                                 throw new ArgumentNullException ("source");
1898
1899                         return source.TypedQuery.Where ((e) => e is TResult).Cast<TResult> ();
1900                 }
1901                 #endregion
1902
1903                 #region Reverse
1904                 public static ParallelQuery<TSource> Reverse<TSource> (this ParallelQuery<TSource> source)
1905                 {
1906                         if (source == null)
1907                                 throw new ArgumentNullException ("source");
1908
1909                         return new ParallelQuery<TSource> (new QueryReverseNode<TSource> (source));
1910                 }
1911                 #endregion
1912
1913                 #region ToArray - ToList - ToDictionary - ToLookup
1914                 public static List<TSource> ToList<TSource> (this ParallelQuery<TSource> source)
1915                 {
1916                         if (source == null)
1917                                 throw new ArgumentNullException ("source");
1918
1919                         if (source.Node.IsOrdered ())
1920                                 return ToListOrdered (source);
1921
1922                         var helper = new ListAggregateHelper<TSource> ();
1923                         List<TSource> temp = source.Aggregate (helper.Seed,
1924                                                                helper.Intermediate,
1925                                                                helper.Reducer,
1926                                                                helper.Final);
1927                         return temp;
1928                 }
1929
1930                 class ListAggregateHelper<TSource>
1931                 {
1932                         public List<TSource> Seed ()
1933                         {
1934                                 return new List<TSource> (50);
1935                         }
1936
1937                         public List<TSource> Intermediate (List<TSource> list, TSource e)
1938                         {
1939                                 list.Add (e);
1940                                 return list;
1941                         }
1942
1943                         public List<TSource> Reducer (List<TSource> list, List<TSource> list2)
1944                         {
1945                                 list.AddRange (list2);
1946                                 return list;
1947                         }
1948
1949                         public List<TSource> Final (List<TSource> list)
1950                         {
1951                                 return list;
1952                         }
1953                 }
1954
1955                 internal static List<TSource> ToListOrdered<TSource> (this ParallelQuery<TSource> source)
1956                 {
1957                         List<TSource> result = new List<TSource> ();
1958
1959                         foreach (TSource element in source)
1960                                 result.Add (element);
1961
1962                         return result;
1963                 }
1964
1965                 public static TSource[] ToArray<TSource> (this ParallelQuery<TSource> source)
1966                 {
1967                         if (source == null)
1968                                 throw new ArgumentNullException ("source");
1969
1970                         if (source.Node.IsOrdered ())
1971                                 return ToListOrdered (source).ToArray ();
1972
1973                         var helper = new ArrayAggregateHelper<TSource> ();
1974                         ParallelExecuter.ProcessAndAggregate<TSource, List<TSource>> (source.Node,
1975                                                                                       helper.Seed,
1976                                                                                       helper.Intermediate,
1977                                                                                       helper.Final);
1978
1979                         return helper.Result;
1980                 }
1981
1982                 class ArrayAggregateHelper<TSource>
1983                 {
1984                         TSource[] result;
1985
1986                         public TSource[] Result {
1987                                 get {
1988                                         return result;
1989                                 }
1990                         }
1991
1992                         internal List<TSource> Seed ()
1993                         {
1994                                 return new List<TSource> ();
1995                         }
1996
1997                         internal List<TSource> Intermediate (List<TSource> list, TSource e)
1998                         {
1999                                 list.Add (e);
2000                                 return list;
2001                         }
2002
2003                         internal void Final (IList<List<TSource>> list)
2004                         {
2005                                 int count = 0;
2006
2007                                 for (int i = 0; i < list.Count; i++)
2008                                         count += list[i].Count;
2009
2010                                 result = new TSource[count];
2011                                 int insertIndex = -1;
2012
2013                                 for (int i = 0; i < list.Count; i++)
2014                                         for (int j = 0; j < list[i].Count; j++)
2015                                                 result [++insertIndex] = list[i][j];
2016                         }
2017                 }
2018
2019                 public static Dictionary<TKey, TSource> ToDictionary<TSource, TKey> (this ParallelQuery<TSource> source,
2020                                                                                      Func<TSource, TKey> keySelector,
2021                                                                                      IEqualityComparer<TKey> comparer)
2022                 {
2023                         return ToDictionary<TSource, TKey, TSource> (source, keySelector, (e) => e, comparer);
2024                 }
2025
2026                 public static Dictionary<TKey, TSource> ToDictionary<TSource, TKey> (this ParallelQuery<TSource> source,
2027                                                                                      Func<TSource, TKey> keySelector)
2028                 {
2029                         return ToDictionary<TSource, TKey, TSource> (source, keySelector, (e) => e, EqualityComparer<TKey>.Default);
2030                 }
2031
2032                 public static Dictionary<TKey, TElement> ToDictionary<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
2033                                                                                                   Func<TSource, TKey> keySelector,
2034                                                                                                   Func<TSource, TElement> elementSelector)
2035                 {
2036                         return ToDictionary<TSource, TKey, TElement> (source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
2037                 }
2038
2039                 public static Dictionary<TKey, TElement> ToDictionary<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
2040                                                                                                   Func<TSource, TKey> keySelector,
2041                                                                                                   Func<TSource, TElement> elementSelector,
2042                                                                                                   IEqualityComparer<TKey> comparer)
2043                 {
2044                         if (source == null)
2045                                 throw new ArgumentNullException ("source");
2046                         if (keySelector == null)
2047                                 throw new ArgumentNullException ("keySelector");
2048                         if (comparer == null)
2049                                 comparer = EqualityComparer<TKey>.Default;
2050                         if (elementSelector == null)
2051                                 throw new ArgumentNullException ("elementSelector");
2052
2053                         var helper = new DictionaryAggregateHelper<TSource, TKey, TElement> (comparer, keySelector, elementSelector);
2054                         return source.Aggregate (helper.Seed,
2055                                                  helper.Intermediate,
2056                                                  helper.Reducer,
2057                                                  helper.Final);
2058                 }
2059
2060                 class DictionaryAggregateHelper<TSource, TKey, TElement>
2061                 {
2062                         IEqualityComparer<TKey> comparer;
2063                         Func<TSource, TKey> keySelector;
2064                         Func<TSource, TElement> elementSelector;
2065
2066                         public DictionaryAggregateHelper (IEqualityComparer<TKey> comparer,
2067                                                           Func<TSource, TKey> keySelector,
2068                                                           Func<TSource, TElement> elementSelector)
2069                         {
2070                                 this.comparer = comparer;
2071                                 this.keySelector = keySelector;
2072                                 this.elementSelector = elementSelector;
2073                         }
2074
2075                         public Dictionary<TKey, TElement> Seed ()
2076                         {
2077                                 return new Dictionary<TKey, TElement> (comparer);
2078                         }
2079
2080                         public Dictionary<TKey, TElement> Intermediate (Dictionary<TKey, TElement> d, TSource e)
2081                         {
2082                                 d.Add (keySelector (e), elementSelector (e));
2083                                 return d;
2084                         }
2085
2086                         public Dictionary<TKey, TElement> Reducer (Dictionary<TKey, TElement> d1, Dictionary<TKey, TElement> d2)
2087                         {
2088                                 foreach (var couple in d2) d1.Add (couple.Key, couple.Value);
2089                                 return d1;
2090                         }
2091
2092                         public Dictionary<TKey, TElement> Final (Dictionary<TKey, TElement> d)
2093                         {
2094                                 return d;
2095                         }
2096                 }
2097
2098                 public static ILookup<TKey, TSource> ToLookup<TSource, TKey> (this ParallelQuery<TSource> source,
2099                                                                               Func<TSource, TKey> keySelector)
2100                 {
2101                         return ToLookup<TSource, TKey, TSource> (source, keySelector, (e) => e, EqualityComparer<TKey>.Default);
2102                 }
2103
2104                 public static ILookup<TKey, TSource> ToLookup<TSource, TKey> (this ParallelQuery<TSource> source,
2105                                                                               Func<TSource, TKey> keySelector,
2106                                                                               IEqualityComparer<TKey> comparer)
2107                 {
2108                         return ToLookup<TSource, TKey, TSource> (source, keySelector, (e) => e, comparer);
2109                 }
2110
2111                 public static ILookup<TKey, TElement> ToLookup<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
2112                                                                                          Func<TSource, TKey> keySelector,
2113                                                                                          Func<TSource, TElement> elementSelector)
2114                 {
2115                         return ToLookup<TSource, TKey, TElement> (source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
2116                 }
2117
2118                 public static ILookup<TKey, TElement> ToLookup<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
2119                                                                                          Func<TSource, TKey> keySelector,
2120                                                                                          Func<TSource, TElement> elementSelector,
2121                                                                                          IEqualityComparer<TKey> comparer)
2122                 {
2123                         if (source == null)
2124                                 throw new ArgumentNullException ("source");
2125                         if (keySelector == null)
2126                                 throw new ArgumentNullException ("keySelector");
2127                         if (comparer == null)
2128                                 comparer = EqualityComparer<TKey>.Default;
2129                         if (elementSelector == null)
2130                                 throw new ArgumentNullException ("elementSelector");
2131
2132                         ConcurrentLookup<TKey, TElement> lookup = new ConcurrentLookup<TKey, TElement> (comparer);
2133                         source.ForAll ((e) => lookup.Add (keySelector (e), elementSelector (e)));
2134
2135                         return lookup;
2136                 }
2137                 #endregion
2138
2139                 #region Concat
2140                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather than "
2141                                    + "System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() extension method "
2142                                    + "to convert the right data source to System.Linq.ParallelQuery<T>.")]
2143                 public static ParallelQuery<TSource> Concat<TSource>(this ParallelQuery<TSource> first,
2144                                                                      IEnumerable<TSource> second)
2145                 {
2146                         throw new NotSupportedException ();
2147                 }
2148
2149                 public static ParallelQuery<TSource> Concat<TSource> (this ParallelQuery<TSource> first, ParallelQuery<TSource> second)
2150                 {
2151                         return new ParallelQuery<TSource> (new QueryConcatNode<TSource> (first.Node, second.Node));
2152                 }
2153                 #endregion
2154
2155                 #region DefaultIfEmpty
2156                 public static ParallelQuery<TSource> DefaultIfEmpty<TSource> (this ParallelQuery<TSource> source)
2157                 {
2158                         return source.DefaultIfEmpty (default (TSource));
2159                 }
2160
2161                 public static ParallelQuery<TSource> DefaultIfEmpty<TSource> (this ParallelQuery<TSource> source, TSource defaultValue)
2162                 {
2163                         return new ParallelQuery<TSource> (new QueryDefaultEmptyNode<TSource> (source.Node, defaultValue));
2164                 }
2165                 #endregion
2166
2167                 #region First
2168                 public static TSource First<TSource> (this ParallelQuery<TSource> source)
2169                 {
2170                         CancellationTokenSource src = new CancellationTokenSource ();
2171                         IEnumerator<TSource> enumerator = source.WithImplementerToken (src).GetEnumerator ();
2172
2173                         if (enumerator == null || !enumerator.MoveNext ())
2174                                 throw new InvalidOperationException ("source contains no element");
2175
2176                         TSource result = enumerator.Current;
2177                         src.Cancel ();
2178                         enumerator.Dispose ();
2179
2180                         return result;
2181                 }
2182
2183                 public static TSource First<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
2184                 {
2185                         return source.Where (predicate).First ();
2186                 }
2187
2188                 public static TSource FirstOrDefault<TSource> (this ParallelQuery<TSource> source)
2189                 {
2190                         return source.DefaultIfEmpty ().First ();
2191                 }
2192
2193                 public static TSource FirstOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
2194                 {
2195                         return source.Where (predicate).FirstOrDefault ();
2196                 }
2197                 #endregion
2198
2199                 #region Last
2200                 public static TSource Last<TSource> (this ParallelQuery<TSource> source)
2201                 {
2202                         return source.Reverse ().First ();
2203                 }
2204
2205                 public static TSource Last<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
2206                 {
2207                         return source.Reverse ().First (predicate);
2208                 }
2209
2210                 public static TSource LastOrDefault<TSource> (this ParallelQuery<TSource> source)
2211                 {
2212                         return source.Reverse ().FirstOrDefault ();
2213                 }
2214
2215                 public static TSource LastOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
2216                 {
2217                         return source.Reverse ().FirstOrDefault (predicate);
2218                 }
2219                 #endregion
2220
2221                 #region Zip
2222                 public static ParallelQuery<TResult> Zip<TFirst, TSecond, TResult> (this ParallelQuery<TFirst> first,
2223                                                                                       ParallelQuery<TSecond> second,
2224                                                                                       Func<TFirst, TSecond, TResult> resultSelector)
2225                 {
2226                         if (first == null)
2227                                 throw new ArgumentNullException ("first");
2228                         if (second == null)
2229                                 throw new ArgumentNullException ("second");
2230                         if (resultSelector == null)
2231                                 throw new ArgumentNullException ("resultSelector");
2232
2233                         return new ParallelQuery<TResult> (new QueryZipNode<TFirst, TSecond, TResult> (resultSelector, first.Node, second.Node));
2234                 }
2235
2236                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
2237                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
2238                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
2239                 public static ParallelQuery<TResult> Zip<TFirst, TSecond, TResult> (this ParallelQuery<TFirst> first,
2240                                                                                       IEnumerable<TSecond> second,
2241                                                                                       Func<TFirst, TSecond, TResult> resultSelector)
2242                 {
2243                         throw new NotSupportedException ();
2244                 }
2245                 #endregion
2246
2247                 #region Helpers
2248
2249                 class Identity<T>
2250                 {
2251                         public T Apply (T input)
2252                         {
2253                                 return input;
2254                         }
2255                 }
2256                 #endregion
2257         }
2258 }
2259 #endif