Code:
/ 4.0 / 4.0 / DEVDIV_TFS / Dev10 / Releases / RTMRel / ndp / fx / src / Core / System / Linq / Parallel / QueryOperators / Unary / ElementAtQueryOperator.cs / 1305376 / ElementAtQueryOperator.cs
// ==++==
//
// Copyright (c) Microsoft Corporation. All rights reserved.
//
// ==--==
// =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
//
// ElementAtQueryOperator.cs
//
// [....]
//
// =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Threading;
namespace System.Linq.Parallel
{
///
/// ElementAt just retrieves an element at a specific index. There is some cross-partition
/// coordination to force partitions to stop looking once a partition has found the
/// sought-after element.
///
/// : UnaryQueryOperator
{
private readonly int m_index; // The index that we're looking for.
private bool m_prematureMerge = false; // Whether to prematurely merge the input of this operator.
//----------------------------------------------------------------------------------------
// Constructs a new instance of the contains search operator.
//
// Arguments:
// child - the child tree to enumerate.
// index - index we are searching for.
//
internal ElementAtQueryOperator(IEnumerable child, int index)
:base(child)
{
Contract.Assert(child != null, "child data source cannot be null");
Contract.Assert(index >= 0, "index can't be less than 0");
m_index = index;
if (ExchangeUtilities.IsWorseThan(Child.OrdinalIndexState, OrdinalIndexState.Correct))
{
m_prematureMerge = true;
}
}
//---------------------------------------------------------------------------------------
// Just opens the current operator, including opening the child and wrapping it with
// partitions as needed.
//
internal override QueryResults Open(
QuerySettings settings, bool preferStriping)
{
// We just open the child operator.
QueryResults childQueryResults = Child.Open(settings, false);
return new UnaryQueryOperatorResults(childQueryResults, this, settings, preferStriping);
}
internal override void WrapPartitionedStream(
PartitionedStream inputStream, IPartitionedStreamRecipient recipient, bool preferStriping, QuerySettings settings)
{
// If the child OOP index is not correct, reindex.
int partitionCount = inputStream.PartitionCount;
PartitionedStream intKeyStream;
if (m_prematureMerge)
{
intKeyStream = ExecuteAndCollectResults(inputStream, partitionCount, Child.OutputOrdered, preferStriping, settings).GetPartitionedStream();
Contract.Assert(intKeyStream.OrdinalIndexState == OrdinalIndexState.Indexible);
}
else
{
intKeyStream = (PartitionedStream)(object)inputStream;
}
// Create a shared cancelation variable and then return a possibly wrapped new enumerator.
Shared resultFoundFlag = new Shared(false);
PartitionedStream outputStream = new PartitionedStream(
partitionCount, Util.GetDefaultComparer(), OrdinalIndexState.Correct);
for (int i = 0; i < partitionCount; i++)
{
outputStream[i] = new ElementAtQueryOperatorEnumerator(intKeyStream[i], m_index, resultFoundFlag, settings.CancellationState.MergedCancellationToken);
}
recipient.Receive(outputStream);
}
//---------------------------------------------------------------------------------------
// Returns an enumerable that represents the query executing sequentially.
//
internal override IEnumerable AsSequentialQuery(CancellationToken token)
{
Contract.Assert(false, "This method should never be called as fallback to sequential is handled in Aggregate().");
throw new NotSupportedException();
}
//---------------------------------------------------------------------------------------
// Whether this operator performs a premature merge.
//
internal override bool LimitsParallelism
{
get { return m_prematureMerge; }
}
///
/// Executes the query, either sequentially or in parallel, depending on the query execution mode and
/// whether a premature merge was inserted by this ElementAt operator.
///
/// result
/// withDefaultValue
/// whether an element with this index exists
internal bool Aggregate(out TSource result, bool withDefaultValue)
{
// If we were to insert a premature merge before this ElementAt, and we are executing in conservative mode, run the whole query
// sequentially.
if (LimitsParallelism && SpecifiedQuerySettings.WithDefaults().ExecutionMode.Value != ParallelExecutionMode.ForceParallelism)
{
CancellationState cancelState = SpecifiedQuerySettings.CancellationState;
if (withDefaultValue)
{
IEnumerable childAsSequential = Child.AsSequentialQuery(cancelState.ExternalCancellationToken);
IEnumerable childWithCancelChecks = CancellableEnumerable.Wrap(childAsSequential, cancelState.ExternalCancellationToken);
result = ExceptionAggregator.WrapEnumerable(childWithCancelChecks, cancelState).ElementAtOrDefault(m_index);
}
else
{
IEnumerable childAsSequential = Child.AsSequentialQuery(cancelState.ExternalCancellationToken);
IEnumerable childWithCancelChecks = CancellableEnumerable.Wrap(childAsSequential, cancelState.ExternalCancellationToken);
result = ExceptionAggregator.WrapEnumerable(childWithCancelChecks, cancelState).ElementAt(m_index);
}
return true;
}
using (IEnumerator e = GetEnumerator(ParallelMergeOptions.FullyBuffered))
{
if (e.MoveNext())
{
TSource current = e.Current;
Contract.Assert(!e.MoveNext(), "expected enumerator to be empty");
result = current;
return true;
}
}
result = default(TSource);
return false;
}
//----------------------------------------------------------------------------------------
// This enumerator performs the search for the element at the specified index.
//
class ElementAtQueryOperatorEnumerator : QueryOperatorEnumerator
{
private QueryOperatorEnumerator m_source; // The source data.
private int m_index; // The index of the element to seek.
private Shared m_resultFoundFlag; // Whether to cancel the operation.
private CancellationToken m_cancellationToken;
//---------------------------------------------------------------------------------------
// Instantiates a new any/all search operator.
//
internal ElementAtQueryOperatorEnumerator(QueryOperatorEnumerator source,
int index, Shared resultFoundFlag,
CancellationToken cancellationToken)
{
Contract.Assert(source != null);
Contract.Assert(index >= 0);
Contract.Assert(resultFoundFlag != null);
m_source = source;
m_index = index;
m_resultFoundFlag = resultFoundFlag;
m_cancellationToken = cancellationToken;
}
//----------------------------------------------------------------------------------------
// Enumerates the entire input until the element with the specified is found or another
// partition has signaled that it found the element.
//
internal override bool MoveNext(ref TSource currentElement, ref int currentKey)
{
// Just walk the enumerator until we've found the element.
int i = 0;
while (m_source.MoveNext(ref currentElement, ref currentKey))
{
if ((i++ & CancellationState.POLL_INTERVAL) == 0)
CancellationState.ThrowIfCanceled(m_cancellationToken);
if (m_resultFoundFlag.Value)
{
// Another partition found the element.
break;
}
if (currentKey == m_index)
{
// We have found the element. Cancel other searches and return true.
m_resultFoundFlag.Value = true;
return true;
}
}
return false;
}
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.
//
// ==--==
// =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
//
// ElementAtQueryOperator.cs
//
// [....]
//
// =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Threading;
namespace System.Linq.Parallel
{
///
/// ElementAt just retrieves an element at a specific index. There is some cross-partition
/// coordination to force partitions to stop looking once a partition has found the
/// sought-after element.
///
/// : UnaryQueryOperator
{
private readonly int m_index; // The index that we're looking for.
private bool m_prematureMerge = false; // Whether to prematurely merge the input of this operator.
//----------------------------------------------------------------------------------------
// Constructs a new instance of the contains search operator.
//
// Arguments:
// child - the child tree to enumerate.
// index - index we are searching for.
//
internal ElementAtQueryOperator(IEnumerable child, int index)
:base(child)
{
Contract.Assert(child != null, "child data source cannot be null");
Contract.Assert(index >= 0, "index can't be less than 0");
m_index = index;
if (ExchangeUtilities.IsWorseThan(Child.OrdinalIndexState, OrdinalIndexState.Correct))
{
m_prematureMerge = true;
}
}
//---------------------------------------------------------------------------------------
// Just opens the current operator, including opening the child and wrapping it with
// partitions as needed.
//
internal override QueryResults Open(
QuerySettings settings, bool preferStriping)
{
// We just open the child operator.
QueryResults childQueryResults = Child.Open(settings, false);
return new UnaryQueryOperatorResults(childQueryResults, this, settings, preferStriping);
}
internal override void WrapPartitionedStream(
PartitionedStream inputStream, IPartitionedStreamRecipient recipient, bool preferStriping, QuerySettings settings)
{
// If the child OOP index is not correct, reindex.
int partitionCount = inputStream.PartitionCount;
PartitionedStream intKeyStream;
if (m_prematureMerge)
{
intKeyStream = ExecuteAndCollectResults(inputStream, partitionCount, Child.OutputOrdered, preferStriping, settings).GetPartitionedStream();
Contract.Assert(intKeyStream.OrdinalIndexState == OrdinalIndexState.Indexible);
}
else
{
intKeyStream = (PartitionedStream)(object)inputStream;
}
// Create a shared cancelation variable and then return a possibly wrapped new enumerator.
Shared resultFoundFlag = new Shared(false);
PartitionedStream outputStream = new PartitionedStream(
partitionCount, Util.GetDefaultComparer(), OrdinalIndexState.Correct);
for (int i = 0; i < partitionCount; i++)
{
outputStream[i] = new ElementAtQueryOperatorEnumerator(intKeyStream[i], m_index, resultFoundFlag, settings.CancellationState.MergedCancellationToken);
}
recipient.Receive(outputStream);
}
//---------------------------------------------------------------------------------------
// Returns an enumerable that represents the query executing sequentially.
//
internal override IEnumerable AsSequentialQuery(CancellationToken token)
{
Contract.Assert(false, "This method should never be called as fallback to sequential is handled in Aggregate().");
throw new NotSupportedException();
}
//---------------------------------------------------------------------------------------
// Whether this operator performs a premature merge.
//
internal override bool LimitsParallelism
{
get { return m_prematureMerge; }
}
///
/// Executes the query, either sequentially or in parallel, depending on the query execution mode and
/// whether a premature merge was inserted by this ElementAt operator.
///
/// result
/// withDefaultValue
/// whether an element with this index exists
internal bool Aggregate(out TSource result, bool withDefaultValue)
{
// If we were to insert a premature merge before this ElementAt, and we are executing in conservative mode, run the whole query
// sequentially.
if (LimitsParallelism && SpecifiedQuerySettings.WithDefaults().ExecutionMode.Value != ParallelExecutionMode.ForceParallelism)
{
CancellationState cancelState = SpecifiedQuerySettings.CancellationState;
if (withDefaultValue)
{
IEnumerable childAsSequential = Child.AsSequentialQuery(cancelState.ExternalCancellationToken);
IEnumerable childWithCancelChecks = CancellableEnumerable.Wrap(childAsSequential, cancelState.ExternalCancellationToken);
result = ExceptionAggregator.WrapEnumerable(childWithCancelChecks, cancelState).ElementAtOrDefault(m_index);
}
else
{
IEnumerable childAsSequential = Child.AsSequentialQuery(cancelState.ExternalCancellationToken);
IEnumerable childWithCancelChecks = CancellableEnumerable.Wrap(childAsSequential, cancelState.ExternalCancellationToken);
result = ExceptionAggregator.WrapEnumerable(childWithCancelChecks, cancelState).ElementAt(m_index);
}
return true;
}
using (IEnumerator e = GetEnumerator(ParallelMergeOptions.FullyBuffered))
{
if (e.MoveNext())
{
TSource current = e.Current;
Contract.Assert(!e.MoveNext(), "expected enumerator to be empty");
result = current;
return true;
}
}
result = default(TSource);
return false;
}
//----------------------------------------------------------------------------------------
// This enumerator performs the search for the element at the specified index.
//
class ElementAtQueryOperatorEnumerator : QueryOperatorEnumerator
{
private QueryOperatorEnumerator m_source; // The source data.
private int m_index; // The index of the element to seek.
private Shared m_resultFoundFlag; // Whether to cancel the operation.
private CancellationToken m_cancellationToken;
//---------------------------------------------------------------------------------------
// Instantiates a new any/all search operator.
//
internal ElementAtQueryOperatorEnumerator(QueryOperatorEnumerator source,
int index, Shared resultFoundFlag,
CancellationToken cancellationToken)
{
Contract.Assert(source != null);
Contract.Assert(index >= 0);
Contract.Assert(resultFoundFlag != null);
m_source = source;
m_index = index;
m_resultFoundFlag = resultFoundFlag;
m_cancellationToken = cancellationToken;
}
//----------------------------------------------------------------------------------------
// Enumerates the entire input until the element with the specified is found or another
// partition has signaled that it found the element.
//
internal override bool MoveNext(ref TSource currentElement, ref int currentKey)
{
// Just walk the enumerator until we've found the element.
int i = 0;
while (m_source.MoveNext(ref currentElement, ref currentKey))
{
if ((i++ & CancellationState.POLL_INTERVAL) == 0)
CancellationState.ThrowIfCanceled(m_cancellationToken);
if (m_resultFoundFlag.Value)
{
// Another partition found the element.
break;
}
if (currentKey == m_index)
{
// We have found the element. Cancel other searches and return true.
m_resultFoundFlag.Value = true;
return true;
}
}
return false;
}
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
- XmlSchemaSimpleType.cs
- UIPropertyMetadata.cs
- XhtmlBasicPageAdapter.cs
- CryptoApi.cs
- AsmxEndpointPickerExtension.cs
- StringReader.cs
- RtfNavigator.cs
- LinkArea.cs
- DynamicObject.cs
- WebMessageEncoderFactory.cs
- FormViewDeletedEventArgs.cs
- AssemblyCache.cs
- MailAddressParser.cs
- InkPresenterAutomationPeer.cs
- XmlSchemaSimpleTypeRestriction.cs
- AppAction.cs
- EntityDataSourceWrapperPropertyDescriptor.cs
- XPathException.cs
- EdmFunction.cs
- SessionPageStateSection.cs
- ErrorLog.cs
- DictionarySectionHandler.cs
- SecureStringHasher.cs
- JsonDeserializer.cs
- InputLanguageEventArgs.cs
- QueryCursorEventArgs.cs
- ScriptReferenceBase.cs
- AnnotationResourceChangedEventArgs.cs
- FindSimilarActivitiesVerb.cs
- DefinitionBase.cs
- LoadRetryConstantStrategy.cs
- AttributedMetaModel.cs
- SafeNativeMethodsOther.cs
- SqlProcedureAttribute.cs
- ArcSegment.cs
- CounterSet.cs
- UserPreferenceChangedEventArgs.cs
- TryCatch.cs
- JoinTreeNode.cs
- _HelperAsyncResults.cs
- RenderData.cs
- BindingNavigator.cs
- TagMapInfo.cs
- WebPartExportVerb.cs
- Property.cs
- LinkLabel.cs
- PerformanceCounterTraceRecord.cs
- LongValidator.cs
- MaskedTextBox.cs
- Formatter.cs
- OptimizerPatterns.cs
- BaseValidator.cs
- TabPanel.cs
- UrlPath.cs
- FlowLayoutSettings.cs
- VectorAnimationUsingKeyFrames.cs
- WorkflowLayouts.cs
- DbDataSourceEnumerator.cs
- BitmapEffectrendercontext.cs
- KeySpline.cs
- PortCache.cs
- XmlDeclaration.cs
- DataGridViewCellValidatingEventArgs.cs
- SocketAddress.cs
- HttpRequestCacheValidator.cs
- TimelineCollection.cs
- DoubleConverter.cs
- EntityContainerEmitter.cs
- DataServices.cs
- FontFaceLayoutInfo.cs
- DesignerActionUIStateChangeEventArgs.cs
- PackagePartCollection.cs
- SchemaDeclBase.cs
- XmlTextEncoder.cs
- PngBitmapEncoder.cs
- HtmlWindow.cs
- ArraySet.cs
- XmlSchemaCompilationSettings.cs
- WebScriptEndpointElement.cs
- TargetConverter.cs
- ADMembershipProvider.cs
- HtmlInputPassword.cs
- EncodedStreamFactory.cs
- PathFigureCollection.cs
- EdmType.cs
- ObfuscateAssemblyAttribute.cs
- ValidationPropertyAttribute.cs
- IsolatedStoragePermission.cs
- XmlDocumentSerializer.cs
- TextParaLineResult.cs
- X509CertificateRecipientServiceCredential.cs
- RectangleGeometry.cs
- Paragraph.cs
- FileVersion.cs
- GlyphElement.cs
- ComplexTypeEmitter.cs
- TextChange.cs
- CompilerCollection.cs
- BamlStream.cs
- DbBuffer.cs