QueryStartNode.cs 4.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125
  1. //
  2. // QueryStartNode.cs
  3. //
  4. // Author:
  5. // Jérémie "Garuma" Laval <[email protected]>
  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. using System;
  27. using System.Collections;
  28. using System.Collections.Generic;
  29. using System.Collections.Concurrent;
  30. using System.Threading;
  31. using System.Threading.Tasks;
  32. namespace System.Linq.Parallel.QueryNodes
  33. {
  34. internal interface QueryStartNode : IVisitableNode {
  35. int Count { get; }
  36. }
  37. internal class QueryStartNode<T> : QueryBaseNode<T>, QueryStartNode
  38. {
  39. readonly IEnumerable<T> source;
  40. readonly Partitioner<T> customPartitioner;
  41. internal QueryStartNode (IEnumerable<T> source)
  42. {
  43. if (source == null)
  44. throw new ArgumentNullException ("source");
  45. this.source = source;
  46. }
  47. internal QueryStartNode (Partitioner<T> custom)
  48. {
  49. if (custom == null)
  50. throw new ArgumentNullException ("custom");
  51. this.customPartitioner = custom;
  52. }
  53. // If possible, this property will return the number of element the query
  54. // is going to process. If that number if pretty low, executing the query
  55. // sequentially is better
  56. public int Count {
  57. get {
  58. if (source == null)
  59. return -1;
  60. ICollection coll = source as ICollection;
  61. return coll == null ? -1 : coll.Count;
  62. }
  63. }
  64. public override void Visit (INodeVisitor visitor)
  65. {
  66. visitor.Visit ((QueryStartNode)this);
  67. }
  68. internal override IEnumerable<T> GetSequential ()
  69. {
  70. if (source != null)
  71. return source;
  72. return WrapHelper.Wrap (customPartitioner.GetPartitions (1))[0];
  73. }
  74. internal override IList<IEnumerable<T>> GetEnumerables (QueryOptions options)
  75. {
  76. if (customPartitioner != null) {
  77. return WrapHelper.Wrap (customPartitioner.GetPartitions (options.PartitionCount));
  78. }
  79. Partitioner<T> partitioner
  80. = (options.UseStrip) ? ParallelPartitioner.CreateForStrips (source, 1) : ParallelPartitioner.CreateBest (source);
  81. return WrapHelper.Wrap (partitioner.GetPartitions (options.PartitionCount));
  82. }
  83. internal override IList<IEnumerable<KeyValuePair<long, T>>> GetOrderedEnumerables (QueryOptions options)
  84. {
  85. OrderablePartitioner<T> partitioner = null;
  86. if (customPartitioner != null) {
  87. partitioner = customPartitioner as OrderablePartitioner<T>;
  88. if (partitioner == null)
  89. throw new InvalidOperationException ("The partitionner you are using doesn't support ordered partitionning");
  90. } else {
  91. partitioner =
  92. (options.UseStrip) ? ParallelPartitioner.CreateForStrips (source, 1) : ParallelPartitioner.CreateBest (source);
  93. }
  94. options.PartitionerSettings = Tuple.Create (partitioner.KeysOrderedAcrossPartitions,
  95. partitioner.KeysOrderedInEachPartition,
  96. partitioner.KeysNormalized);
  97. // We only support one style of partitioning at the moment.
  98. // Standard partitioners follow this style.
  99. if (options.UseStrip && (!partitioner.KeysOrderedInEachPartition || partitioner.KeysOrderedAcrossPartitions))
  100. throw new NotImplementedException ("Partitioner must have KeysOrderedInEachPartition "
  101. + "and !KeysOrderedAcrossPartitions"
  102. + "to be used with indexed operators");
  103. return WrapHelper.Wrap (partitioner.GetOrderablePartitions (options.PartitionCount));
  104. }
  105. }
  106. }