1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
|
// ==++==
//
// Copyright (c) Microsoft Corporation. All rights reserved.
//
// ==--==
// =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
//
// NullableIntAverageAggregationOperator.cs
//
// <OWNER>[....]</OWNER>
//
// =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Threading;
namespace System.Linq.Parallel
{
/// <summary>
/// An inlined average aggregation operator and its enumerator, for Nullable ints.
/// </summary>
internal sealed class NullableIntAverageAggregationOperator : InlinedAggregationOperator<int?, Pair<long, long>, double?>
{
//---------------------------------------------------------------------------------------
// Constructs a new instance of an average associative operator.
//
internal NullableIntAverageAggregationOperator(IEnumerable<int?> child) : base(child)
{
}
//---------------------------------------------------------------------------------------
// Executes the entire query tree, and aggregates the intermediate results into the
// final result based on the binary operators and final reduction.
//
// Return Value:
// The single result of aggregation.
//
protected override double? InternalAggregate(ref Exception singularExceptionToThrow)
{
// Because the final reduction is typically much cheaper than the intermediate
// reductions over the individual partitions, and because each parallel partition
// will do a lot of work to produce a single output element, we prefer to turn off
// pipelining, and process the final reductions serially.
using (IEnumerator<Pair<long, long>> enumerator = GetEnumerator(ParallelMergeOptions.FullyBuffered, true))
{
// If the sequence was empty, return null right away.
if (!enumerator.MoveNext())
{
return null;
}
Pair<long, long> result = enumerator.Current;
// Simply add together the sums and totals.
while (enumerator.MoveNext())
{
checked
{
result.First += enumerator.Current.First;
result.Second += enumerator.Current.Second;
}
}
// And divide the sum by the total to obtain the final result.
return (double)result.First / result.Second;
}
}
//---------------------------------------------------------------------------------------
// Creates an enumerator that is used internally for the final aggregation step.
//
protected override QueryOperatorEnumerator<Pair<long, long>,int> CreateEnumerator<TKey>(
int index, int count, QueryOperatorEnumerator<int?, TKey> source, object sharedData, CancellationToken cancellationToken)
{
return new NullableIntAverageAggregationOperatorEnumerator<TKey>(source, index, cancellationToken);
}
//---------------------------------------------------------------------------------------
// This enumerator type encapsulates the intermediary aggregation over the underlying
// (possibly partitioned) data source.
//
private class NullableIntAverageAggregationOperatorEnumerator<TKey> : InlinedAggregationOperatorEnumerator<Pair<long, long>>
{
private QueryOperatorEnumerator<int?, TKey> m_source; // The source data.
//---------------------------------------------------------------------------------------
// Instantiates a new aggregation operator.
//
internal NullableIntAverageAggregationOperatorEnumerator(QueryOperatorEnumerator<int?, TKey> source, int partitionIndex,
CancellationToken cancellationToken) :
base(partitionIndex, cancellationToken)
{
Contract.Assert(source != null);
m_source = source;
}
//---------------------------------------------------------------------------------------
// Tallies up the average of the underlying data source, walking the entire thing the first
// time MoveNext is called on this object.
//
protected override bool MoveNextCore(ref Pair<long, long> currentElement)
{
// The temporary result contains the running sum and count, respectively.
long sum = 0;
long count = 0;
QueryOperatorEnumerator<int?, TKey> source = m_source;
int? current = default(int?);
TKey keyUnused = default(TKey);
int i = 0;
while (source.MoveNext(ref current, ref keyUnused))
{
if ((i++ & CancellationState.POLL_INTERVAL) == 0)
CancellationState.ThrowIfCanceled(m_cancellationToken);
if (current.HasValue)
{
sum += current.GetValueOrDefault();
count++;
}
}
currentElement = new Pair<long, long>(sum, count);
return count > 0;
}
//---------------------------------------------------------------------------------------
// Dispose of resources associated with the underlying enumerator.
//
protected override void Dispose(bool disposing)
{
Contract.Assert(m_source != null);
m_source.Dispose();
}
}
}
}
|
