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