mcs/class/System.Core/System.Linq.Parallel.QueryNodes/QueryWhereNode.cs

   1 //
   2 // QueryWhereNode.cs
   3 //
   4 // Author:
   5 //       Jérémie "Garuma" Laval <jeremie.laval@gmail.com>
   6 //
   7 // Copyright (c) 2010 Jérémie "Garuma" Laval
   8 //
   9 // Permission is hereby granted, free of charge, to any person obtaining a copy
  10 // of this software and associated documentation files (the "Software"), to deal
  11 // in the Software without restriction, including without limitation the rights
  12 // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  13 // copies of the Software, and to permit persons to whom the Software is
  14 // furnished to do so, subject to the following conditions:
  15 //
  16 // The above copyright notice and this permission notice shall be included in
  17 // all copies or substantial portions of the Software.
  18 //
  19 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  20 // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  21 // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  22 // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  23 // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  24 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  25 // THE SOFTWARE.
  26
  27 #if NET_4_0
  28 using System;
  29 using System.Linq;
  30 using System.Threading;
  31 using System.Collections.Generic;
  32
  33 namespace System.Linq.Parallel.QueryNodes
  34 {
  35
  36         internal class QueryWhereNode<TSource> : QueryStreamNode<TSource, TSource>
  37         {
  38                 Func<TSource, int, bool> indexedPredicate;
  39                 Func<TSource, bool> predicate;
  40
  41                 internal QueryWhereNode (QueryBaseNode<TSource> parent, Func<TSource, bool> predicate)
  42                         : base (parent, false)
  43                 {
  44                         this.predicate = predicate;
  45                 }
  46
  47                 internal QueryWhereNode (QueryBaseNode<TSource> parent, Func<TSource, int, bool> predicate)
  48                         : base (parent, true)
  49                 {
  50                         this.indexedPredicate = predicate;
  51                 }
  52
  53                 internal override IEnumerable<TSource> GetSequential ()
  54                 {
  55                         IEnumerable<TSource> parent = Parent.GetSequential ();
  56
  57                         if (indexedPredicate != null)
  58                                 return parent.Where (indexedPredicate);
  59                         else
  60                                 return parent.Where (predicate);
  61                 }
  62
  63                 internal override IList<IEnumerable<TSource>> GetEnumerablesIndexed (QueryOptions options)
  64                 {
  65                         return Parent.GetOrderedEnumerables (options)
  66                                 .Select ((i) => i.Where ((e) => indexedPredicate (e.Value, (int)e.Key)).Select ((e) => e.Value))
  67                                 .ToList ();
  68                 }
  69
  70                 internal override IList<IEnumerable<TSource>> GetEnumerablesNonIndexed (QueryOptions options)
  71                 {
  72                         return Parent.GetEnumerables (options)
  73                                 .Select ((i) => i.Where (predicate))
  74                                 .ToList ();
  75                 }
  76
  77                 internal override IList<IEnumerable<KeyValuePair<long, TSource>>> GetOrderedEnumerables (QueryOptions options)
  78                 {
  79                         IList<IEnumerable<KeyValuePair<long, TSource>>> sources = Parent.GetOrderedEnumerables (options);
  80
  81                         Tuple<TSource, long, bool>[] store = new Tuple<TSource, long, bool>[sources.Count];
  82                         long lastIndex = 0;
  83
  84                         Barrier barrier = new Barrier (sources.Count, delegate (Barrier b) {
  85                                 // Sort the store
  86                                 Array.Sort (store, ArraySortMethod);
  87
  88                                 // Reassign a good index
  89                                 int i = 0;
  90
  91                                 for (i = 0; i < store.Length && store[i].Item3; i++) {
  92                                         Tuple<TSource, long, bool> old = store[i];
  93                                         store[i] = Tuple.Create (old.Item1, lastIndex + i, old.Item3);
  94                                 }
  95
  96                                 // Update lastIndex for next round
  97                                 lastIndex += i;
  98                         });
  99
 100                         return sources
 101                                 .Select ((s, i) => GetEnumerator (s, barrier, options.MergedToken, store, i))
 102                                 .ToList ();
 103                 }
 104
 105                 static int ArraySortMethod (Tuple<TSource, long, bool> lhs, Tuple<TSource, long, bool> rhs)
 106                 {
 107                         if (lhs.Item3 && !rhs.Item3)
 108                                 return -1;
 109                         if (!lhs.Item3 && rhs.Item3)
 110                                 return 1;
 111                         if (!lhs.Item3 && !rhs.Item3)
 112                                 return 0;
 113
 114                         return (lhs.Item2 < rhs.Item2) ? -1 : 1;
 115                 }
 116
 117                 IEnumerable<KeyValuePair<long, TSource>> GetEnumerator (IEnumerable<KeyValuePair<long, TSource>> source,
 118                                                                         Barrier barrier,
 119                                                                         CancellationToken token,
 120                                                                         Tuple<TSource, long, bool>[] store, int index)
 121                 {
 122                         IEnumerator<KeyValuePair<long, TSource>> current = source.GetEnumerator ();
 123                         bool result;
 124                         Tuple<TSource, long, bool> reset = Tuple.Create (default (TSource), long.MaxValue, false);
 125
 126                         try {
 127                                 while (current.MoveNext ()) {
 128                                         KeyValuePair<long, TSource> curr = current.Current;
 129
 130                                         result = IsIndexed ? indexedPredicate (curr.Value, (int)curr.Key) : predicate (curr.Value);
 131                                         store[index] = Tuple.Create (curr.Value, curr.Key, result);
 132
 133                                         barrier.SignalAndWait (token);
 134
 135                                         Tuple<TSource, long, bool> value = store [index];
 136
 137                                         if (value.Item3)
 138                                                 yield return new KeyValuePair<long, TSource> (value.Item2, value.Item1);
 139
 140                                         // Reset
 141                                         store[index] = reset;
 142                                 }
 143                         } finally {
 144                                 // Remove our participation
 145                                 barrier.RemoveParticipant ();
 146                                 current.Dispose ();
 147                         }
 148                 }
 149         }
 150 }
 151 #endif