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                         ProcessingSlot[] store = new ProcessingSlot[sources.Count];
  82                         Comparison<ProcessingSlot> arrayComparison = ArraySortMethod;
  83                         long lastIndex = 0;
  84
  85                         Barrier barrier = new Barrier (sources.Count, delegate (Barrier b) {
  86                                 // Sort the store
  87                                 Array.Sort (store, arrayComparison);
  88
  89                                 // Reassign a good index
  90                                 int i = 0;
  91                                 for (i = 0; i < store.Length && store[i].IsValid; ++i)
  92                                         store[i].Index = lastIndex + i;
  93
  94                                 // Update lastIndex for next round
  95                                 lastIndex += i;
  96                         });
  97
  98                         return sources
  99                                 .Select ((s, i) => GetEnumerator (s, barrier, options.MergedToken, store, i))
 100                                 .ToList ();
 101                 }
 102
 103                 static int ArraySortMethod (ProcessingSlot lhs, ProcessingSlot rhs)
 104                 {
 105                         if (lhs.IsValid && !rhs.IsValid)
 106                                 return -1;
 107                         if (!lhs.IsValid && rhs.IsValid)
 108                                 return 1;
 109                         if (!lhs.IsValid && !rhs.IsValid)
 110                                 return 0;
 111
 112                         return lhs.Index < rhs.Index ? -1 : 1;
 113                 }
 114
 115                 IEnumerable<KeyValuePair<long, TSource>> GetEnumerator (IEnumerable<KeyValuePair<long, TSource>> source,
 116                                                                         Barrier barrier,
 117                                                                         CancellationToken token,
 118                                                                         ProcessingSlot[] store, int index)
 119                 {
 120                         IEnumerator<KeyValuePair<long, TSource>> current = source.GetEnumerator ();
 121                         bool isIndexed = IsIndexed;
 122
 123                         try {
 124                                 while (current.MoveNext ()) {
 125                                         KeyValuePair<long, TSource> curr = current.Current;
 126
 127                                         bool result = isIndexed ? indexedPredicate (curr.Value, (int)curr.Key) : predicate (curr.Value);
 128                                         store[index].Update (curr.Value, curr.Key, result);
 129
 130                                         barrier.SignalAndWait (token);
 131
 132                                         var value = store [index];
 133
 134                                         if (value.IsValid)
 135                                                 yield return new KeyValuePair<long, TSource> (value.Index, value.Value);
 136
 137                                         // Reset
 138                                         store[index].Clear ();
 139                                 }
 140                         } finally {
 141                                 // Remove our participation
 142                                 barrier.RemoveParticipant ();
 143                                 current.Dispose ();
 144                         }
 145                 }
 146
 147                 struct ProcessingSlot
 148                 {
 149                         public TSource Value;
 150                         public long Index;
 151                         public bool IsValid;
 152
 153                         public void Update (TSource v, long i, bool t)
 154                         {
 155                                 Value = v;
 156                                 Index = i;
 157                                 IsValid = t;
 158                         }
 159
 160                         public void Clear ()
 161                         {
 162                                 Update (default (TSource), long.MaxValue, false);
 163                         }
 164                 }
 165         }
 166 }
 167 #endif