Code:
/ 4.0 / 4.0 / DEVDIV_TFS / Dev10 / Releases / RTMRel / ndp / fx / src / Core / System / Linq / Parallel / QueryOperators / Inlined / DoubleAverageAggregationOperator.cs / 1305376 / DoubleAverageAggregationOperator.cs
// ==++==
//
// Copyright (c) Microsoft Corporation. All rights reserved.
//
// ==--==
// =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
//
// DoubleAverageAggregationOperator.cs
//
// [....]
//
// =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Threading;
namespace System.Linq.Parallel
{
///
/// An inlined average aggregation operator and its enumerator, for doubles.
///
internal sealed class DoubleAverageAggregationOperator : InlinedAggregationOperator, double>
{
//----------------------------------------------------------------------------------------
// Constructs a new instance of an average associative operator.
//
internal DoubleAverageAggregationOperator(IEnumerable 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> enumerator = GetEnumerator(ParallelMergeOptions.FullyBuffered, true))
{
// Throw an error for empty results.
if (!enumerator.MoveNext())
{
singularExceptionToThrow = new InvalidOperationException(SR.GetString(SR.NoElements));
return default(double);
}
Pair 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 result.First / result.Second;
}
}
//---------------------------------------------------------------------------------------
// Creates an enumerator that is used internally for the final aggregation step.
//
protected override QueryOperatorEnumerator, int> CreateEnumerator(
int index, int count, QueryOperatorEnumerator source, object sharedData,
CancellationToken cancellationToken)
{
return new DoubleAverageAggregationOperatorEnumerator(source, index, cancellationToken);
}
//---------------------------------------------------------------------------------------
// This enumerator type encapsulates the intermediary aggregation over the underlying
// (possibly partitioned) data source.
//
private class DoubleAverageAggregationOperatorEnumerator : InlinedAggregationOperatorEnumerator>
{
private QueryOperatorEnumerator m_source; // The source data.
//----------------------------------------------------------------------------------------
// Instantiates a new aggregation operator.
//
internal DoubleAverageAggregationOperatorEnumerator(QueryOperatorEnumerator 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 currentElement)
{
// The temporary result contains the running sum and count, respectively.
double sum = 0.0;
long count = 0;
QueryOperatorEnumerator source = m_source;
double current = default(double);
TKey keyUnused = default(TKey);
if (source.MoveNext(ref current, ref keyUnused))
{
int i = 0;
do
{
if ((i++ & CancellationState.POLL_INTERVAL) == 0)
CancellationState.ThrowIfCanceled(m_cancellationToken);
checked
{
sum += current;
count++;
}
}
while (source.MoveNext(ref current, ref keyUnused));
currentElement = new Pair(sum, count);
return true;
}
return false;
}
//----------------------------------------------------------------------------------------
// Dispose of resources associated with the underlying enumerator.
//
protected override void Dispose(bool disposing)
{
Contract.Assert(m_source != null);
m_source.Dispose();
}
}
}
}
// File provided for Reference Use Only by Microsoft Corporation (c) 2007.
// ==++==
//
// Copyright (c) Microsoft Corporation. All rights reserved.
//
// ==--==
// =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
//
// DoubleAverageAggregationOperator.cs
//
// [....]
//
// =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Threading;
namespace System.Linq.Parallel
{
///
/// An inlined average aggregation operator and its enumerator, for doubles.
///
internal sealed class DoubleAverageAggregationOperator : InlinedAggregationOperator, double>
{
//----------------------------------------------------------------------------------------
// Constructs a new instance of an average associative operator.
//
internal DoubleAverageAggregationOperator(IEnumerable 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> enumerator = GetEnumerator(ParallelMergeOptions.FullyBuffered, true))
{
// Throw an error for empty results.
if (!enumerator.MoveNext())
{
singularExceptionToThrow = new InvalidOperationException(SR.GetString(SR.NoElements));
return default(double);
}
Pair 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 result.First / result.Second;
}
}
//---------------------------------------------------------------------------------------
// Creates an enumerator that is used internally for the final aggregation step.
//
protected override QueryOperatorEnumerator, int> CreateEnumerator(
int index, int count, QueryOperatorEnumerator source, object sharedData,
CancellationToken cancellationToken)
{
return new DoubleAverageAggregationOperatorEnumerator(source, index, cancellationToken);
}
//---------------------------------------------------------------------------------------
// This enumerator type encapsulates the intermediary aggregation over the underlying
// (possibly partitioned) data source.
//
private class DoubleAverageAggregationOperatorEnumerator : InlinedAggregationOperatorEnumerator>
{
private QueryOperatorEnumerator m_source; // The source data.
//----------------------------------------------------------------------------------------
// Instantiates a new aggregation operator.
//
internal DoubleAverageAggregationOperatorEnumerator(QueryOperatorEnumerator 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 currentElement)
{
// The temporary result contains the running sum and count, respectively.
double sum = 0.0;
long count = 0;
QueryOperatorEnumerator source = m_source;
double current = default(double);
TKey keyUnused = default(TKey);
if (source.MoveNext(ref current, ref keyUnused))
{
int i = 0;
do
{
if ((i++ & CancellationState.POLL_INTERVAL) == 0)
CancellationState.ThrowIfCanceled(m_cancellationToken);
checked
{
sum += current;
count++;
}
}
while (source.MoveNext(ref current, ref keyUnused));
currentElement = new Pair(sum, count);
return true;
}
return false;
}
//----------------------------------------------------------------------------------------
// Dispose of resources associated with the underlying enumerator.
//
protected override void Dispose(bool disposing)
{
Contract.Assert(m_source != null);
m_source.Dispose();
}
}
}
}
// File provided for Reference Use Only by Microsoft Corporation (c) 2007.
Link Menu

This book is available now!
Buy at Amazon US or
Buy at Amazon UK
- HeaderUtility.cs
- Decoder.cs
- HttpApplicationFactory.cs
- NamedPipeTransportManager.cs
- LinqDataSource.cs
- OleServicesContext.cs
- ListViewTableCell.cs
- TrustSection.cs
- NullReferenceException.cs
- IProvider.cs
- Identity.cs
- RedistVersionInfo.cs
- QueryCacheKey.cs
- FilterableAttribute.cs
- MatrixTransform3D.cs
- HebrewNumber.cs
- StoryFragments.cs
- Input.cs
- ValuePatternIdentifiers.cs
- StylusOverProperty.cs
- ConnectionsZone.cs
- Control.cs
- BinaryMethodMessage.cs
- StateElement.cs
- BamlTreeNode.cs
- OleDbFactory.cs
- XhtmlConformanceSection.cs
- AssemblyHash.cs
- SiteMapProvider.cs
- Pen.cs
- DescendentsWalker.cs
- AuthenticationConfig.cs
- DetailsViewCommandEventArgs.cs
- RewritingProcessor.cs
- SystemIPInterfaceProperties.cs
- Utility.cs
- XmlCDATASection.cs
- WebPartActionVerb.cs
- OSFeature.cs
- RepeaterItemCollection.cs
- CfgSemanticTag.cs
- JsonGlobals.cs
- RequestCachePolicy.cs
- AddInPipelineAttributes.cs
- ControlBindingsCollection.cs
- InputEventArgs.cs
- OracleSqlParser.cs
- TextCollapsingProperties.cs
- CompoundFileReference.cs
- ContentOperations.cs
- ToolStripMenuItem.cs
- BaseParser.cs
- PersonalizationEntry.cs
- SymbolMethod.cs
- OperatingSystem.cs
- MouseDevice.cs
- GiveFeedbackEvent.cs
- TraceContextEventArgs.cs
- MatrixConverter.cs
- ServerType.cs
- FunctionImportMapping.ReturnTypeRenameMapping.cs
- SqlBulkCopyColumnMapping.cs
- Rotation3DAnimation.cs
- TextParagraph.cs
- DbParameterCollectionHelper.cs
- SelectedPathEditor.cs
- HostingEnvironmentSection.cs
- MetadataSerializer.cs
- MessageSecurityOverHttp.cs
- EmptyControlCollection.cs
- PolyQuadraticBezierSegment.cs
- TextEditorCopyPaste.cs
- DragEvent.cs
- StringInfo.cs
- SettingsProperty.cs
- EqualityComparer.cs
- StoreItemCollection.cs
- TreeNodeStyleCollection.cs
- BatchParser.cs
- DetailsViewPagerRow.cs
- BlockUIContainer.cs
- ButtonStandardAdapter.cs
- _NtlmClient.cs
- SystemMulticastIPAddressInformation.cs
- InvokeBase.cs
- UserNameSecurityToken.cs
- KnownTypesProvider.cs
- EditorAttribute.cs
- CodeTypeMemberCollection.cs
- SequenceQuery.cs
- ListBindingConverter.cs
- Hashtable.cs
- SafeArchiveContext.cs
- PreservationFileWriter.cs
- HwndSource.cs
- FileSystemWatcher.cs
- HMAC.cs
- Types.cs
- ComponentEditorPage.cs
- ArcSegment.cs