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 || MOBILE
  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, e));
 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                         var helper = new CountAggregateHelper<TSource> ();
1090                         return source.Aggregate<TSource, int, int> (helper.Seed,
1091                                                                     helper.Intermediate,
1092                                                                     helper.Reducer,
1093                                                                     helper.Final);
1094                 }
1095
1096                 public static int Count<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
1097                 {
1098                         if (source == null)
1099                                 throw new ArgumentNullException ("source");
1100                         if (predicate == null)
1101                                 throw new ArgumentNullException ("predicate");
1102
1103                         return source.Where (predicate).Count ();
1104                 }
1105
1106                 class CountAggregateHelper<TSource>
1107                 {
1108                         public int Seed ()
1109                         {
1110                                 return 0;
1111                         }
1112
1113                         public int Intermediate (int acc, TSource e)
1114                         {
1115                                 return acc + 1;
1116                         }
1117
1118                         public int Reducer (int acc1, int acc2)
1119                         {
1120                                 return acc1 + acc2;
1121                         }
1122
1123                         public int Final (int acc)
1124                         {
1125                                 return acc;
1126                         }
1127                 }
1128
1129                 public static long LongCount<TSource> (this ParallelQuery<TSource> source)
1130                 {
1131                         if (source == null)
1132                                 throw new ArgumentNullException ("source");
1133
1134                         var helper = new LongCountAggregateHelper<TSource> ();
1135                         return source.Aggregate<TSource, long, long> (helper.Seed,
1136                                                                       helper.Intermediate,
1137                                                                       helper.Reducer,
1138                                                                       helper.Final);
1139                 }
1140
1141                 public static long LongCount<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
1142                 {
1143                         if (source == null)
1144                                 throw new ArgumentNullException ("source");
1145                         if (predicate == null)
1146                                 throw new ArgumentNullException ("predicate");
1147
1148                         return source.Where (predicate).LongCount ();
1149                 }
1150
1151                 class LongCountAggregateHelper<TSource>
1152                 {
1153                         public long Seed ()
1154                         {
1155                                 return 0;
1156                         }
1157
1158                         public long Intermediate (long acc, TSource e)
1159                         {
1160                                 return acc + 1;
1161                         }
1162
1163                         public long Reducer (long acc1, long acc2)
1164                         {
1165                                 return acc1 + acc2;
1166                         }
1167
1168                         public long Final (long acc)
1169                         {
1170                                 return acc;
1171                         }
1172                 }
1173                 #endregion
1174
1175                 #region Average
1176                 public static double Average (this ParallelQuery<int> source)
1177                 {
1178                         if (source == null)
1179                                 throw new ArgumentNullException ("source");
1180
1181                         return source.Aggregate (() => new int[2],
1182                                                 (acc, e) => { acc[0] += e; acc[1]++; return acc; },
1183                                                 (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
1184                                                 (acc) => acc[0] / ((double)acc[1]));
1185                 }
1186
1187                 public static double Average (this ParallelQuery<long> source)
1188                 {
1189                         if (source == null)
1190                                 throw new ArgumentNullException ("source");
1191
1192                         return source.Aggregate (() => new long[2],
1193                                                 (acc, e) => { acc[0] += e; acc[1]++; return acc; },
1194                                                 (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
1195                                                 (acc) => acc[0] / ((double)acc[1]));
1196                 }
1197
1198                 public static decimal Average (this ParallelQuery<decimal> source)
1199                 {
1200                         if (source == null)
1201                                 throw new ArgumentNullException ("source");
1202
1203                         return source.Aggregate (() => new decimal[2],
1204                                                 (acc, e) => { acc[0] += e; acc[1]++; return acc; },
1205                                                 (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
1206                                                 (acc) => acc[0] / acc[1]);
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.Aggregate (() => new double[2],
1215                                                 (acc, e) => { acc[0] += e; acc[1]++; return acc; },
1216                                                 (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
1217                                                 (acc) => acc[0] / ((double)acc[1]));
1218                 }
1219
1220                 public static float Average (this ParallelQuery<float> source)
1221                 {
1222                         if (source == null)
1223                                 throw new ArgumentNullException ("source");
1224
1225                         return source.Aggregate (() => new float[2],
1226                                                 (acc, e) => { acc[0] += e; acc[1]++; return acc; },
1227                                                 (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
1228                                                 (acc) => acc[0] / acc[1]);
1229                 }
1230                 #endregion
1231
1232                 #region More Average
1233                 public static double? Average (this ParallelQuery<int?> source)
1234                 {
1235                         if (source == null)
1236                                 throw new ArgumentNullException ("source");
1237
1238                         return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();;
1239                 }
1240
1241                 public static double? Average (this ParallelQuery<long?> source)
1242                 {
1243                         if (source == null)
1244                                 throw new ArgumentNullException ("source");
1245
1246                         return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
1247                 }
1248
1249                 public static decimal? Average (this ParallelQuery<decimal?> source)
1250                 {
1251                         if (source == null)
1252                                 throw new ArgumentNullException ("source");
1253
1254                         return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
1255                 }
1256
1257                 public static double? Average (this ParallelQuery<double?> source)
1258                 {
1259                         if (source == null)
1260                                 throw new ArgumentNullException ("source");
1261
1262                         return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
1263                 }
1264
1265                 public static float? Average (this ParallelQuery<float?> source)
1266                 {
1267                         if (source == null)
1268                                 throw new ArgumentNullException ("source");
1269
1270                         return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
1271                 }
1272
1273                 public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
1274                 {
1275                         if (source == null)
1276                                 throw new ArgumentNullException ("source");
1277                         if (selector == null)
1278                                 throw new ArgumentNullException ("selector");
1279
1280                         return source.Select (selector).Average ();
1281                 }
1282
1283                 public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
1284                 {
1285                         if (source == null)
1286                                 throw new ArgumentNullException ("source");
1287                         if (selector == null)
1288                                 throw new ArgumentNullException ("selector");
1289
1290                         return source.Select (selector).Average ();
1291                 }
1292
1293                 public static float Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
1294                 {
1295                         if (source == null)
1296                                 throw new ArgumentNullException ("source");
1297                         if (selector == null)
1298                                 throw new ArgumentNullException ("selector");
1299
1300                         return source.Select (selector).Average ();
1301                 }
1302
1303                 public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
1304                 {
1305                         if (source == null)
1306                                 throw new ArgumentNullException ("source");
1307                         if (selector == null)
1308                                 throw new ArgumentNullException ("selector");
1309
1310                         return source.Select (selector).Average ();
1311                 }
1312
1313                 public static decimal Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
1314                 {
1315                         if (source == null)
1316                                 throw new ArgumentNullException ("source");
1317                         if (selector == null)
1318                                 throw new ArgumentNullException ("selector");
1319
1320                         return source.Select (selector).Average ();
1321                 }
1322
1323                 public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
1324                 {
1325                         if (source == null)
1326                                 throw new ArgumentNullException ("source");
1327                         if (selector == null)
1328                                 throw new ArgumentNullException ("selector");
1329
1330                         return source.Select (selector).Average ();
1331                 }
1332
1333                 public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
1334                 {
1335                         if (source == null)
1336                                 throw new ArgumentNullException ("source");
1337                         if (selector == null)
1338                                 throw new ArgumentNullException ("selector");
1339
1340                         return source.Select (selector).Average ();
1341                 }
1342
1343                 public static float? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
1344                 {
1345                         if (source == null)
1346                                 throw new ArgumentNullException ("source");
1347                         if (selector == null)
1348                                 throw new ArgumentNullException ("selector");
1349
1350                         return source.Select (selector).Average ();
1351                 }
1352
1353                 public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
1354                 {
1355                         if (source == null)
1356                                 throw new ArgumentNullException ("source");
1357                         if (selector == null)
1358                                 throw new ArgumentNullException ("selector");
1359
1360                         return source.Select (selector).Average ();
1361                 }
1362
1363                 public static decimal? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
1364                 {
1365                         if (source == null)
1366                                 throw new ArgumentNullException ("source");
1367                         if (selector == null)
1368                                 throw new ArgumentNullException ("selector");
1369
1370                         return source.Select (selector).Average ();
1371                 }
1372                 #endregion
1373
1374                 #region Sum
1375                 public static int Sum (this ParallelQuery<int> source)
1376                 {
1377                         if (source == null)
1378                                 throw new ArgumentNullException ("source");
1379
1380                         return source.Aggregate (0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
1381                 }
1382
1383                 public static long Sum (this ParallelQuery<long> source)
1384                 {
1385                         if (source == null)
1386                                 throw new ArgumentNullException ("source");
1387
1388                         return source.Aggregate ((long)0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
1389                 }
1390
1391                 public static float Sum (this ParallelQuery<float> source)
1392                 {
1393                         if (source == null)
1394                                 throw new ArgumentNullException ("source");
1395
1396                         return source.Aggregate (0.0f, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
1397                 }
1398
1399                 public static double Sum (this ParallelQuery<double> source)
1400                 {
1401                         if (source == null)
1402                                 throw new ArgumentNullException ("source");
1403
1404                         return source.Aggregate (0.0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
1405                 }
1406
1407                 public static decimal Sum (this ParallelQuery<decimal> source)
1408                 {
1409                         if (source == null)
1410                                 throw new ArgumentNullException ("source");
1411
1412                         return source.Aggregate ((decimal)0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
1413                 }
1414
1415                 public static int? Sum (this ParallelQuery<int?> source)
1416                 {
1417                         return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
1418                 }
1419
1420                 public static long? Sum (this ParallelQuery<long?> source)
1421                 {
1422                         if (source == null)
1423                                 throw new ArgumentNullException ("source");
1424
1425                         return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
1426                 }
1427
1428                 public static float? Sum (this ParallelQuery<float?> source)
1429                 {
1430                         if (source == null)
1431                                 throw new ArgumentNullException ("source");
1432
1433                         return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
1434                 }
1435
1436                 public static double? Sum (this ParallelQuery<double?> source)
1437                 {
1438                         if (source == null)
1439                                 throw new ArgumentNullException ("source");
1440
1441                         return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
1442                 }
1443
1444                 public static decimal? Sum (this ParallelQuery<decimal?> source)
1445                 {
1446                         if (source == null)
1447                                 throw new ArgumentNullException ("source");
1448
1449                         return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
1450                 }
1451
1452                 public static int Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
1453                 {
1454                         if (source == null)
1455                                 throw new ArgumentNullException ("source");
1456                         if (selector == null)
1457                                 throw new ArgumentNullException ("selector");
1458
1459                         return source.Select (selector).Sum ();
1460                 }
1461
1462                 public static long Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
1463                 {
1464                         if (source == null)
1465                                 throw new ArgumentNullException ("source");
1466                         if (selector == null)
1467                                 throw new ArgumentNullException ("selector");
1468
1469                         return source.Select (selector).Sum ();
1470                 }
1471
1472                 public static decimal Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
1473                 {
1474                         if (source == null)
1475                                 throw new ArgumentNullException ("source");
1476                         if (selector == null)
1477                                 throw new ArgumentNullException ("selector");
1478
1479                         return source.Select (selector).Sum ();
1480                 }
1481
1482                 public static float Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
1483                 {
1484                         if (source == null)
1485                                 throw new ArgumentNullException ("source");
1486                         if (selector == null)
1487                                 throw new ArgumentNullException ("selector");
1488
1489                         return source.Select (selector).Sum ();
1490                 }
1491
1492                 public static double Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
1493                 {
1494                         if (source == null)
1495                                 throw new ArgumentNullException ("source");
1496                         if (selector == null)
1497                                 throw new ArgumentNullException ("selector");
1498
1499                         return source.Select (selector).Sum ();
1500                 }
1501
1502                 public static int? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
1503                 {
1504                         if (source == null)
1505                                 throw new ArgumentNullException ("source");
1506                         if (selector == null)
1507                                 throw new ArgumentNullException ("selector");
1508
1509                         return source.Select (selector).Sum ();
1510                 }
1511
1512                 public static long? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
1513                 {
1514                         if (source == null)
1515                                 throw new ArgumentNullException ("source");
1516                         if (selector == null)
1517                                 throw new ArgumentNullException ("selector");
1518
1519                         return source.Select (selector).Sum ();
1520                 }
1521
1522                 public static decimal? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
1523                 {
1524                         if (source == null)
1525                                 throw new ArgumentNullException ("source");
1526                         if (selector == null)
1527                                 throw new ArgumentNullException ("selector");
1528
1529                         return source.Select (selector).Sum ();
1530                 }
1531
1532                 public static float? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
1533                 {
1534                         if (source == null)
1535                                 throw new ArgumentNullException ("source");
1536                         if (selector == null)
1537                                 throw new ArgumentNullException ("selector");
1538
1539                         return source.Select (selector).Sum ();
1540                 }
1541
1542                 public static double? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
1543                 {
1544                         if (source == null)
1545                                 throw new ArgumentNullException ("source");
1546                         if (selector == null)
1547                                 throw new ArgumentNullException ("selector");
1548
1549                         return source.Select (selector).Sum ();
1550                 }
1551                 #endregion
1552
1553                 #region Min-Max
1554                 static T BestOrder<T> (ParallelQuery<T> source, BestOrderComparer<T> bestOrderComparer)
1555                 {
1556                         if (source == null)
1557                                 throw new ArgumentNullException ("source");
1558
1559                         T best = source.Aggregate (bestOrderComparer.Seed,
1560                                                    bestOrderComparer.Intermediate,
1561                                                    bestOrderComparer.Intermediate,
1562                                                    new Identity<T> ().Apply);
1563                         return best;
1564                 }
1565
1566                 class BestOrderComparer<T>
1567                 {
1568                         IComparer<T> comparer;
1569                         int inverter;
1570                         T seed;
1571
1572                         public BestOrderComparer (IComparer<T> comparer, int inverter, T seed)
1573                         {
1574                                 this.comparer = comparer;
1575                                 this.inverter = inverter;
1576                                 this.seed = seed;
1577                         }
1578
1579                         public T Seed ()
1580                         {
1581                                 return seed;
1582                         }
1583
1584                         public T Intermediate (T first, T second)
1585                         {
1586                                 return Better (first, second) ? first : second;
1587                         }
1588
1589                         bool Better (T first, T second)
1590                         {
1591                                 return (inverter * comparer.Compare (first, second)) > 0;
1592                         }
1593                 }
1594
1595                 public static int Min (this ParallelQuery<int> source)
1596                 {
1597                         return BestOrder (source, new BestOrderComparer<int> (Comparer<int>.Default, -1, int.MaxValue));
1598                 }
1599
1600                 public static long Min (this ParallelQuery<long> source)
1601                 {
1602                         return BestOrder (source, new BestOrderComparer<long> (Comparer<long>.Default, -1, long.MaxValue));
1603                 }
1604
1605                 public static float Min (this ParallelQuery<float> source)
1606                 {
1607                         return BestOrder (source, new BestOrderComparer<float> (Comparer<float>.Default, -1, float.MaxValue));
1608                 }
1609
1610                 public static double Min (this ParallelQuery<double> source)
1611                 {
1612                         return BestOrder (source, new BestOrderComparer<double> (Comparer<double>.Default, -1, double.MaxValue));
1613                 }
1614
1615                 public static decimal Min (this ParallelQuery<decimal> source)
1616                 {
1617                         return BestOrder (source, new BestOrderComparer<decimal> (Comparer<decimal>.Default, -1, decimal.MaxValue));
1618                 }
1619
1620                 public static TSource Min<TSource> (this ParallelQuery<TSource> source)
1621                 {
1622                         IComparer<TSource> comparer = Comparer<TSource>.Default;
1623                         return BestOrder (source, new BestOrderComparer<TSource> (comparer, -1, default (TSource)));
1624                 }
1625
1626                 public static TResult Min<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> selector)
1627                 {
1628                         if (source == null)
1629                                 throw new ArgumentNullException ("source");
1630                         if (selector == null)
1631                                 throw new ArgumentNullException ("selector");
1632
1633                         return source.Select (selector).Min ();
1634                 }
1635
1636                 public static int? Min (this ParallelQuery<int?> source)
1637                 {
1638                         if (source == null)
1639                                 throw new ArgumentNullException ("source");
1640
1641                         return source.Select ((e) => e.HasValue ? e.Value : int.MaxValue).Min ();
1642                 }
1643
1644                 public static long? Min (this ParallelQuery<long?> source)
1645                 {
1646                         if (source == null)
1647                                 throw new ArgumentNullException ("source");
1648
1649                         return source.Select ((e) => e.HasValue ? e.Value : long.MaxValue).Min ();
1650                 }
1651
1652                 public static float? Min (this ParallelQuery<float?> source)
1653                 {
1654                         if (source == null)
1655                                 throw new ArgumentNullException ("source");
1656
1657                         return source.Select ((e) => e.HasValue ? e.Value : float.MaxValue).Min ();
1658                 }
1659
1660                 public static double? Min (this ParallelQuery<double?> source)
1661                 {
1662                         if (source == null)
1663                                 throw new ArgumentNullException ("source");
1664
1665                         return source.Select ((e) => e.HasValue ? e.Value : double.MaxValue).Min ();
1666                 }
1667
1668                 public static decimal? Min (this ParallelQuery<decimal?> source)
1669                 {
1670                         if (source == null)
1671                                 throw new ArgumentNullException ("source");
1672
1673                         return source.Select ((e) => e.HasValue ? e.Value : decimal.MaxValue).Min ();
1674                 }
1675
1676                 public static int Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
1677                 {
1678                         if (source == null)
1679                                 throw new ArgumentNullException ("source");
1680                         if (selector == null)
1681                                 throw new ArgumentNullException ("selector");
1682
1683                         return source.Select (selector).Min ();
1684                 }
1685
1686                 public static long Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
1687                 {
1688                         if (source == null)
1689                                 throw new ArgumentNullException ("source");
1690                         if (selector == null)
1691                                 throw new ArgumentNullException ("selector");
1692
1693                         return source.Select (selector).Min ();
1694                 }
1695
1696                 public static float Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
1697                 {
1698                         if (source == null)
1699                                 throw new ArgumentNullException ("source");
1700                         if (selector == null)
1701                                 throw new ArgumentNullException ("selector");
1702
1703                         return source.Select (selector).Min ();
1704                 }
1705
1706                 public static double Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
1707                 {
1708                         if (source == null)
1709                                 throw new ArgumentNullException ("source");
1710                         if (selector == null)
1711                                 throw new ArgumentNullException ("selector");
1712
1713                         return source.Select (selector).Min ();
1714                 }
1715
1716                 public static decimal Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
1717                 {
1718                         if (source == null)
1719                                 throw new ArgumentNullException ("source");
1720                         if (selector == null)
1721                                 throw new ArgumentNullException ("selector");
1722
1723                         return source.Select (selector).Min ();
1724                 }
1725
1726                 public static int? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
1727                 {
1728                         if (source == null)
1729                                 throw new ArgumentNullException ("source");
1730                         if (selector == null)
1731                                 throw new ArgumentNullException ("selector");
1732
1733                         return source.Select (selector).Min ();
1734                 }
1735
1736                 public static long? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
1737                 {
1738                         if (source == null)
1739                                 throw new ArgumentNullException ("source");
1740                         if (selector == null)
1741                                 throw new ArgumentNullException ("selector");
1742
1743                         return source.Select (selector).Min ();
1744                 }
1745
1746                 public static float? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
1747                 {
1748                         if (source == null)
1749                                 throw new ArgumentNullException ("source");
1750                         if (selector == null)
1751                                 throw new ArgumentNullException ("selector");
1752
1753                         return source.Select (selector).Min ();
1754                 }
1755
1756                 public static double? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
1757                 {
1758                         if (source == null)
1759                                 throw new ArgumentNullException ("source");
1760                         if (selector == null)
1761                                 throw new ArgumentNullException ("selector");
1762
1763                         return source.Select (selector).Min ();
1764                 }
1765
1766                 public static decimal? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
1767                 {
1768                         if (source == null)
1769                                 throw new ArgumentNullException ("source");
1770                         if (selector == null)
1771                                 throw new ArgumentNullException ("selector");
1772
1773                         return source.Select (selector).Min ();
1774                 }
1775
1776                 public static int Max (this ParallelQuery<int> source)
1777                 {
1778                         return BestOrder (source, new BestOrderComparer<int> (Comparer<int>.Default, 1, int.MinValue));
1779                 }
1780
1781                 public static long Max (this ParallelQuery<long> source)
1782                 {
1783                         return BestOrder (source, new BestOrderComparer<long> (Comparer<long>.Default, 1, long.MinValue));
1784                 }
1785
1786                 public static float Max (this ParallelQuery<float> source)
1787                 {
1788                         return BestOrder (source, new BestOrderComparer<float> (Comparer<float>.Default, 1, float.MinValue));
1789                 }
1790
1791                 public static double Max (this ParallelQuery<double> source)
1792                 {
1793                         return BestOrder (source, new BestOrderComparer<double> (Comparer<double>.Default, 1, double.MinValue));
1794                 }
1795
1796                 public static decimal Max (this ParallelQuery<decimal> source)
1797                 {
1798                         return BestOrder (source, new BestOrderComparer<decimal> (Comparer<decimal>.Default, 1, decimal.MinValue));
1799                 }
1800
1801                 public static TSource Max<TSource> (this ParallelQuery<TSource> source)
1802                 {
1803                         IComparer<TSource> comparer = Comparer<TSource>.Default;
1804                         return BestOrder (source, new BestOrderComparer<TSource> (comparer, 1, default (TSource)));
1805                 }
1806
1807                 public static TResult Max<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> selector)
1808                 {
1809                         if (source == null)
1810                                 throw new ArgumentNullException ("source");
1811                         if (selector == null)
1812                                 throw new ArgumentNullException ("selector");
1813
1814                         return source.Select (selector).Max ();
1815                 }
1816
1817                 public static int? Max (this ParallelQuery<int?> source)
1818                 {
1819                         if (source == null)
1820                                 throw new ArgumentNullException ("source");
1821
1822                         return source.Select ((e) => e.HasValue ? e.Value : int.MinValue).Max ();
1823                 }
1824
1825                 public static long? Max (this ParallelQuery<long?> source)
1826                 {
1827                         if (source == null)
1828                                 throw new ArgumentNullException ("source");
1829
1830                         return source.Select ((e) => e.HasValue ? e.Value : long.MinValue).Max ();
1831                 }
1832
1833                 public static float? Max (this ParallelQuery<float?> source)
1834                 {
1835                         if (source == null)
1836                                 throw new ArgumentNullException ("source");
1837
1838                         return source.Select ((e) => e.HasValue ? e.Value : float.MinValue).Max ();
1839                 }
1840
1841                 public static double? Max (this ParallelQuery<double?> source)
1842                 {
1843                         if (source == null)
1844                                 throw new ArgumentNullException ("source");
1845
1846                         return source.Select ((e) => e.HasValue ? e.Value : double.MinValue).Max ();
1847                 }
1848
1849                 public static decimal? Max (this ParallelQuery<decimal?> source)
1850                 {
1851                         if (source == null)
1852                                 throw new ArgumentNullException ("source");
1853
1854                         return source.Select ((e) => e.HasValue ? e.Value : decimal.MinValue).Max ();
1855                 }
1856
1857                 public static int Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
1858                 {
1859                         if (source == null)
1860                                 throw new ArgumentNullException ("source");
1861                         if (selector == null)
1862                                 throw new ArgumentNullException ("selector");
1863
1864                         return source.Select (selector).Max ();
1865                 }
1866
1867                 public static long Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
1868                 {
1869                         if (source == null)
1870                                 throw new ArgumentNullException ("source");
1871                         if (selector == null)
1872                                 throw new ArgumentNullException ("selector");
1873
1874                         return source.Select (selector).Max ();
1875                 }
1876
1877                 public static float Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
1878                 {
1879                         if (source == null)
1880                                 throw new ArgumentNullException ("source");
1881                         if (selector == null)
1882                                 throw new ArgumentNullException ("selector");
1883
1884                         return source.Select (selector).Max ();
1885                 }
1886
1887                 public static double Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
1888                 {
1889                         if (source == null)
1890                                 throw new ArgumentNullException ("source");
1891                         if (selector == null)
1892                                 throw new ArgumentNullException ("selector");
1893
1894                         return source.Select (selector).Max ();
1895                 }
1896
1897                 public static decimal Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
1898                 {
1899                         if (source == null)
1900                                 throw new ArgumentNullException ("source");
1901                         if (selector == null)
1902                                 throw new ArgumentNullException ("selector");
1903
1904                         return source.Select (selector).Max ();
1905                 }
1906
1907                 public static int? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
1908                 {
1909                         if (source == null)
1910                                 throw new ArgumentNullException ("source");
1911                         if (selector == null)
1912                                 throw new ArgumentNullException ("selector");
1913
1914                         return source.Select (selector).Max ();
1915                 }
1916
1917                 public static long? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
1918                 {
1919                         if (source == null)
1920                                 throw new ArgumentNullException ("source");
1921                         if (selector == null)
1922                                 throw new ArgumentNullException ("selector");
1923
1924                         return source.Select (selector).Max ();
1925                 }
1926
1927                 public static float? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
1928                 {
1929                         if (source == null)
1930                                 throw new ArgumentNullException ("source");
1931                         if (selector == null)
1932                                 throw new ArgumentNullException ("selector");
1933
1934                         return source.Select (selector).Max ();
1935                 }
1936
1937                 public static double? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
1938                 {
1939                         if (source == null)
1940                                 throw new ArgumentNullException ("source");
1941                         if (selector == null)
1942                                 throw new ArgumentNullException ("selector");
1943
1944                         return source.Select (selector).Max ();
1945                 }
1946
1947                 public static decimal? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
1948                 {
1949                         if (source == null)
1950                                 throw new ArgumentNullException ("source");
1951                         if (selector == null)
1952                                 throw new ArgumentNullException ("selector");
1953
1954                         return source.Select (selector).Max ();
1955                 }
1956                 #endregion
1957
1958                 #region Cast / OfType
1959                 public static ParallelQuery<TResult> Cast<TResult> (this ParallelQuery source)
1960                 {
1961                         if (source == null)
1962                                 throw new ArgumentNullException ("source");
1963
1964                         return source.TypedQuery.Select ((e) => (TResult)e);
1965                 }
1966
1967                 public static ParallelQuery<TResult> OfType<TResult> (this ParallelQuery source)
1968                 {
1969                         if (source == null)
1970                                 throw new ArgumentNullException ("source");
1971
1972                         return source.TypedQuery.Where ((e) => e is TResult).Cast<TResult> ();
1973                 }
1974                 #endregion
1975
1976                 #region Reverse
1977                 public static ParallelQuery<TSource> Reverse<TSource> (this ParallelQuery<TSource> source)
1978                 {
1979                         if (source == null)
1980                                 throw new ArgumentNullException ("source");
1981
1982                         return new ParallelQuery<TSource> (new QueryReverseNode<TSource> (source));
1983                 }
1984                 #endregion
1985
1986                 #region ToArray - ToList - ToDictionary - ToLookup
1987                 public static List<TSource> ToList<TSource> (this ParallelQuery<TSource> source)
1988                 {
1989                         if (source == null)
1990                                 throw new ArgumentNullException ("source");
1991
1992                         if (source.Node.IsOrdered ())
1993                                 return ToListOrdered (source);
1994
1995                         var helper = new ListAggregateHelper<TSource> ();
1996                         List<TSource> temp = source.Aggregate (helper.Seed,
1997                                                                helper.Intermediate,
1998                                                                helper.Reducer,
1999                                                                helper.Final);
2000                         return temp;
2001                 }
2002
2003                 class ListAggregateHelper<TSource>
2004                 {
2005                         public List<TSource> Seed ()
2006                         {
2007                                 return new List<TSource> (50);
2008                         }
2009
2010                         public List<TSource> Intermediate (List<TSource> list, TSource e)
2011                         {
2012                                 list.Add (e);
2013                                 return list;
2014                         }
2015
2016                         public List<TSource> Reducer (List<TSource> list, List<TSource> list2)
2017                         {
2018                                 list.AddRange (list2);
2019                                 return list;
2020                         }
2021
2022                         public List<TSource> Final (List<TSource> list)
2023                         {
2024                                 return list;
2025                         }
2026                 }
2027
2028                 internal static List<TSource> ToListOrdered<TSource> (this ParallelQuery<TSource> source)
2029                 {
2030                         List<TSource> result = new List<TSource> ();
2031
2032                         foreach (TSource element in source)
2033                                 result.Add (element);
2034
2035                         return result;
2036                 }
2037
2038                 public static TSource[] ToArray<TSource> (this ParallelQuery<TSource> source)
2039                 {
2040                         if (source == null)
2041                                 throw new ArgumentNullException ("source");
2042
2043                         if (source.Node.IsOrdered ())
2044                                 return ToListOrdered (source).ToArray ();
2045
2046                         var helper = new ArrayAggregateHelper<TSource> ();
2047                         ParallelExecuter.ProcessAndAggregate<TSource, List<TSource>> (source.Node,
2048                                                                                       helper.Seed,
2049                                                                                       helper.Intermediate,
2050                                                                                       helper.Final);
2051
2052                         return helper.Result;
2053                 }
2054
2055                 class ArrayAggregateHelper<TSource>
2056                 {
2057                         TSource[] result;
2058
2059                         public TSource[] Result {
2060                                 get {
2061                                         return result;
2062                                 }
2063                         }
2064
2065                         internal List<TSource> Seed ()
2066                         {
2067                                 return new List<TSource> ();
2068                         }
2069
2070                         internal List<TSource> Intermediate (List<TSource> list, TSource e)
2071                         {
2072                                 list.Add (e);
2073                                 return list;
2074                         }
2075
2076                         internal void Final (IList<List<TSource>> list)
2077                         {
2078                                 int count = 0;
2079
2080                                 for (int i = 0; i < list.Count; i++)
2081                                         count += list[i].Count;
2082
2083                                 result = new TSource[count];
2084                                 int insertIndex = -1;
2085
2086                                 for (int i = 0; i < list.Count; i++)
2087                                         for (int j = 0; j < list[i].Count; j++)
2088                                                 result [++insertIndex] = list[i][j];
2089                         }
2090                 }
2091
2092                 public static Dictionary<TKey, TSource> ToDictionary<TSource, TKey> (this ParallelQuery<TSource> source,
2093                                                                                      Func<TSource, TKey> keySelector,
2094                                                                                      IEqualityComparer<TKey> comparer)
2095                 {
2096                         return ToDictionary<TSource, TKey, TSource> (source, keySelector, (e) => e, comparer);
2097                 }
2098
2099                 public static Dictionary<TKey, TSource> ToDictionary<TSource, TKey> (this ParallelQuery<TSource> source,
2100                                                                                      Func<TSource, TKey> keySelector)
2101                 {
2102                         return ToDictionary<TSource, TKey, TSource> (source, keySelector, (e) => e, EqualityComparer<TKey>.Default);
2103                 }
2104
2105                 public static Dictionary<TKey, TElement> ToDictionary<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
2106                                                                                                   Func<TSource, TKey> keySelector,
2107                                                                                                   Func<TSource, TElement> elementSelector)
2108                 {
2109                         return ToDictionary<TSource, TKey, TElement> (source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
2110                 }
2111
2112                 public static Dictionary<TKey, TElement> ToDictionary<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
2113                                                                                                   Func<TSource, TKey> keySelector,
2114                                                                                                   Func<TSource, TElement> elementSelector,
2115                                                                                                   IEqualityComparer<TKey> comparer)
2116                 {
2117                         if (source == null)
2118                                 throw new ArgumentNullException ("source");
2119                         if (keySelector == null)
2120                                 throw new ArgumentNullException ("keySelector");
2121                         if (comparer == null)
2122                                 comparer = EqualityComparer<TKey>.Default;
2123                         if (elementSelector == null)
2124                                 throw new ArgumentNullException ("elementSelector");
2125
2126                         var helper = new DictionaryAggregateHelper<TSource, TKey, TElement> (comparer, keySelector, elementSelector);
2127                         return source.Aggregate (helper.Seed,
2128                                                  helper.Intermediate,
2129                                                  helper.Reducer,
2130                                                  helper.Final);
2131                 }
2132
2133                 class DictionaryAggregateHelper<TSource, TKey, TElement>
2134                 {
2135                         IEqualityComparer<TKey> comparer;
2136                         Func<TSource, TKey> keySelector;
2137                         Func<TSource, TElement> elementSelector;
2138
2139                         public DictionaryAggregateHelper (IEqualityComparer<TKey> comparer,
2140                                                           Func<TSource, TKey> keySelector,
2141                                                           Func<TSource, TElement> elementSelector)
2142                         {
2143                                 this.comparer = comparer;
2144                                 this.keySelector = keySelector;
2145                                 this.elementSelector = elementSelector;
2146                         }
2147
2148                         public Dictionary<TKey, TElement> Seed ()
2149                         {
2150                                 return new Dictionary<TKey, TElement> (comparer);
2151                         }
2152
2153                         public Dictionary<TKey, TElement> Intermediate (Dictionary<TKey, TElement> d, TSource e)
2154                         {
2155                                 d.Add (keySelector (e), elementSelector (e));
2156                                 return d;
2157                         }
2158
2159                         public Dictionary<TKey, TElement> Reducer (Dictionary<TKey, TElement> d1, Dictionary<TKey, TElement> d2)
2160                         {
2161                                 foreach (var couple in d2) d1.Add (couple.Key, couple.Value);
2162                                 return d1;
2163                         }
2164
2165                         public Dictionary<TKey, TElement> Final (Dictionary<TKey, TElement> d)
2166                         {
2167                                 return d;
2168                         }
2169                 }
2170
2171                 public static ILookup<TKey, TSource> ToLookup<TSource, TKey> (this ParallelQuery<TSource> source,
2172                                                                               Func<TSource, TKey> keySelector)
2173                 {
2174                         return ToLookup<TSource, TKey, TSource> (source, keySelector, (e) => e, EqualityComparer<TKey>.Default);
2175                 }
2176
2177                 public static ILookup<TKey, TSource> ToLookup<TSource, TKey> (this ParallelQuery<TSource> source,
2178                                                                               Func<TSource, TKey> keySelector,
2179                                                                               IEqualityComparer<TKey> comparer)
2180                 {
2181                         return ToLookup<TSource, TKey, TSource> (source, keySelector, (e) => e, comparer);
2182                 }
2183
2184                 public static ILookup<TKey, TElement> ToLookup<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
2185                                                                                          Func<TSource, TKey> keySelector,
2186                                                                                          Func<TSource, TElement> elementSelector)
2187                 {
2188                         return ToLookup<TSource, TKey, TElement> (source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
2189                 }
2190
2191                 public static ILookup<TKey, TElement> ToLookup<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
2192                                                                                          Func<TSource, TKey> keySelector,
2193                                                                                          Func<TSource, TElement> elementSelector,
2194                                                                                          IEqualityComparer<TKey> comparer)
2195                 {
2196                         if (source == null)
2197                                 throw new ArgumentNullException ("source");
2198                         if (keySelector == null)
2199                                 throw new ArgumentNullException ("keySelector");
2200                         if (comparer == null)
2201                                 comparer = EqualityComparer<TKey>.Default;
2202                         if (elementSelector == null)
2203                                 throw new ArgumentNullException ("elementSelector");
2204
2205                         ConcurrentLookup<TKey, TElement> lookup = new ConcurrentLookup<TKey, TElement> (comparer);
2206                         source.ForAll ((e) => lookup.Add (keySelector (e), elementSelector (e)));
2207
2208                         return lookup;
2209                 }
2210                 #endregion
2211
2212                 #region Concat
2213                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather than "
2214                                    + "System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() extension method "
2215                                    + "to convert the right data source to System.Linq.ParallelQuery<T>.")]
2216                 public static ParallelQuery<TSource> Concat<TSource>(this ParallelQuery<TSource> first,
2217                                                                      IEnumerable<TSource> second)
2218                 {
2219                         throw new NotSupportedException ();
2220                 }
2221
2222                 public static ParallelQuery<TSource> Concat<TSource> (this ParallelQuery<TSource> first, ParallelQuery<TSource> second)
2223                 {
2224                         return new ParallelQuery<TSource> (new QueryConcatNode<TSource> (first.Node, second.Node));
2225                 }
2226                 #endregion
2227
2228                 #region DefaultIfEmpty
2229                 public static ParallelQuery<TSource> DefaultIfEmpty<TSource> (this ParallelQuery<TSource> source)
2230                 {
2231                         return source.DefaultIfEmpty (default (TSource));
2232                 }
2233
2234                 public static ParallelQuery<TSource> DefaultIfEmpty<TSource> (this ParallelQuery<TSource> source, TSource defaultValue)
2235                 {
2236                         return new ParallelQuery<TSource> (new QueryDefaultEmptyNode<TSource> (source.Node, defaultValue));
2237                 }
2238                 #endregion
2239
2240                 #region First
2241                 public static TSource First<TSource> (this ParallelQuery<TSource> source)
2242                 {
2243                         CancellationTokenSource src = new CancellationTokenSource ();
2244                         IEnumerator<TSource> enumerator = source.WithImplementerToken (src).GetEnumerator ();
2245
2246                         if (enumerator == null || !enumerator.MoveNext ())
2247                                 throw new InvalidOperationException ("source contains no element");
2248
2249                         TSource result = enumerator.Current;
2250                         src.Cancel ();
2251                         enumerator.Dispose ();
2252
2253                         return result;
2254                 }
2255
2256                 public static TSource First<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
2257                 {
2258                         return source.Where (predicate).First ();
2259                 }
2260
2261                 public static TSource FirstOrDefault<TSource> (this ParallelQuery<TSource> source)
2262                 {
2263                         return source.DefaultIfEmpty ().First ();
2264                 }
2265
2266                 public static TSource FirstOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
2267                 {
2268                         return source.Where (predicate).FirstOrDefault ();
2269                 }
2270                 #endregion
2271
2272                 #region Last
2273                 public static TSource Last<TSource> (this ParallelQuery<TSource> source)
2274                 {
2275                         return source.Reverse ().First ();
2276                 }
2277
2278                 public static TSource Last<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
2279                 {
2280                         return source.Reverse ().First (predicate);
2281                 }
2282
2283                 public static TSource LastOrDefault<TSource> (this ParallelQuery<TSource> source)
2284                 {
2285                         return source.Reverse ().FirstOrDefault ();
2286                 }
2287
2288                 public static TSource LastOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
2289                 {
2290                         return source.Reverse ().FirstOrDefault (predicate);
2291                 }
2292                 #endregion
2293
2294                 #region Zip
2295                 public static ParallelQuery<TResult> Zip<TFirst, TSecond, TResult> (this ParallelQuery<TFirst> first,
2296                                                                                       ParallelQuery<TSecond> second,
2297                                                                                       Func<TFirst, TSecond, TResult> resultSelector)
2298                 {
2299                         if (first == null)
2300                                 throw new ArgumentNullException ("first");
2301                         if (second == null)
2302                                 throw new ArgumentNullException ("second");
2303                         if (resultSelector == null)
2304                                 throw new ArgumentNullException ("resultSelector");
2305
2306                         return new ParallelQuery<TResult> (new QueryZipNode<TFirst, TSecond, TResult> (resultSelector, first.Node, second.Node));
2307                 }
2308
2309                 [ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
2310                                    + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
2311                                    + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
2312                 public static ParallelQuery<TResult> Zip<TFirst, TSecond, TResult> (this ParallelQuery<TFirst> first,
2313                                                                                       IEnumerable<TSecond> second,
2314                                                                                       Func<TFirst, TSecond, TResult> resultSelector)
2315                 {
2316                         throw new NotSupportedException ();
2317                 }
2318                 #endregion
2319
2320                 #region Helpers
2321
2322                 class Identity<T>
2323                 {
2324                         public T Apply (T input)
2325                         {
2326                                 return input;
2327                         }
2328                 }
2329                 #endregion
2330         }
2331 }
2332 #endif