Code:
/ 4.0 / 4.0 / DEVDIV_TFS / Dev10 / Releases / RTMRel / ndp / fx / src / Core / System / Linq / Parallel / QueryOperators / Unary / FirstQueryOperator.cs / 1305376 / FirstQueryOperator.cs
// ==++==
//
// Copyright (c) Microsoft Corporation. All rights reserved.
//
// ==--==
// =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
//
// FirstQueryOperator.cs
//
// [....]
//
// =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Threading;
namespace System.Linq.Parallel
{
///
/// First tries to discover the first element in the source, optionally matching a
/// predicate. All partitions search in parallel, publish the lowest index for a
/// candidate match, and reach a barrier. Only the partition that "wins" the ----,
/// i.e. who found the candidate with the smallest index, will yield an element.
///
/// : UnaryQueryOperator
{
private readonly Func m_predicate; // The optional predicate used during the search.
private readonly bool m_prematureMergeNeeded; // Whether to prematurely merge the input of this operator.
//----------------------------------------------------------------------------------------
// Initializes a new first operator.
//
// Arguments:
// child - the child whose data we will reverse
//
internal FirstQueryOperator(IEnumerable child, Func predicate)
:base(child)
{
Contract.Assert(child != null, "child data source cannot be null");
m_predicate = predicate;
m_prematureMergeNeeded = Child.OrdinalIndexState.IsWorseThan(OrdinalIndexState.Increasing);
}
//---------------------------------------------------------------------------------------
// 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)
{
OrdinalIndexState inputIndexState = inputStream.OrdinalIndexState;
PartitionedStream intKeyStream;
int partitionCount = inputStream.PartitionCount;
// If the index is not at least increasing, we need to reindex.
if (m_prematureMergeNeeded)
{
ListQueryResults listResults = ExecuteAndCollectResults(inputStream, partitionCount, Child.OutputOrdered, preferStriping, settings);
intKeyStream = listResults.GetPartitionedStream();
}
else
{
Contract.Assert(typeof(TKey) == typeof(int));
intKeyStream = (PartitionedStream)(object)inputStream;
}
// Generate the shared data.
Shared sharedFirstCandidate = new Shared(-1);
CountdownEvent sharedBarrier = new CountdownEvent(partitionCount);
PartitionedStream outputStream = new PartitionedStream(
partitionCount, Util.GetDefaultComparer(), OrdinalIndexState.Shuffled);
for (int i = 0; i < partitionCount; i++)
{
outputStream[i] = new FirstQueryOperatorEnumerator(
intKeyStream[i], m_predicate, sharedFirstCandidate, sharedBarrier, 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 ParallelEnumerable.First().");
throw new NotSupportedException();
}
//---------------------------------------------------------------------------------------
// Whether this operator performs a premature merge.
//
internal override bool LimitsParallelism
{
get { return m_prematureMergeNeeded; }
}
//----------------------------------------------------------------------------------------
// The enumerator type responsible for executing the first operation.
//
class FirstQueryOperatorEnumerator : QueryOperatorEnumerator
{
private QueryOperatorEnumerator m_source; // The data source to enumerate.
private Func m_predicate; // The optional predicate used during the search.
private bool m_alreadySearched; // Set once the enumerator has performed the search.
// Data shared among partitions.
private Shared m_sharedFirstCandidate; // The current first candidate.
private CountdownEvent m_sharedBarrier; // Shared barrier, signaled when partitions find their 1st element.
private CancellationToken m_cancellationToken; // Token used to cancel this operator.
//---------------------------------------------------------------------------------------
// Instantiates a new enumerator.
//
internal FirstQueryOperatorEnumerator(
QueryOperatorEnumerator source, Func predicate,
Shared sharedFirstCandidate, CountdownEvent sharedBarrier, CancellationToken cancellationToken)
{
Contract.Assert(source != null);
Contract.Assert(sharedFirstCandidate != null);
Contract.Assert(sharedBarrier != null);
m_source = source;
m_predicate = predicate;
m_sharedFirstCandidate = sharedFirstCandidate;
m_sharedBarrier = sharedBarrier;
m_cancellationToken = cancellationToken;
}
//----------------------------------------------------------------------------------------
// Straightforward IEnumerator methods.
//
internal override bool MoveNext(ref TSource currentElement, ref int currentKey)
{
Contract.Assert(m_source != null);
if (m_alreadySearched)
{
return false;
}
// Look for the lowest element.
TSource candidate = default(TSource);
int candidateIndex = -1;
try
{
int key = default(int);
int i = 0;
while (m_source.MoveNext(ref candidate, ref key))
{
if ((i++ & CancellationState.POLL_INTERVAL) == 0)
CancellationState.ThrowIfCanceled(m_cancellationToken);
// If the predicate is null or the current element satisfies it, we have found the
// current partition's "candidate" for the first element. Note it.
if (m_predicate == null || m_predicate(candidate))
{
candidateIndex = key;
// Try to swap our index with the shared one, so long as it's smaller.
int observedSharedIndex;
do
{
observedSharedIndex = m_sharedFirstCandidate.Value;
} while ((observedSharedIndex == -1 || candidateIndex < observedSharedIndex) &&
Interlocked.CompareExchange(ref m_sharedFirstCandidate.Value, candidateIndex, observedSharedIndex) != observedSharedIndex);
break;
}
else if (m_sharedFirstCandidate.Value != -1 && key > m_sharedFirstCandidate.Value)
{
// We've scanned past another partition's best element. Bail.
break;
}
}
}
finally
{
// No matter whether we exit due to an exception or normal completion, we must ensure
// that we signal other partitions that we have completed. Otherwise, we can cause deadlocks.
m_sharedBarrier.Signal();
}
m_alreadySearched = true;
// Only if we might be a candidate do we wait.
if (candidateIndex != -1)
{
m_sharedBarrier.Wait(m_cancellationToken);
// Now re-read the shared index. If it's the same as ours, we won and return true.
if (m_sharedFirstCandidate.Value == candidateIndex)
{
currentElement = candidate;
currentKey = 0; // 1st (and only) element, so we hardcode the output index to 0.
return true;
}
}
// If we got here, we didn't win. Return false.
return false;
}
protected override void Dispose(bool disposing)
{
m_source.Dispose();
}
}
}
}
// File provided for Reference Use Only by Microsoft Corporation (c) 2007.
// ==++==
//
// Copyright (c) Microsoft Corporation. All rights reserved.
//
// ==--==
// =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
//
// FirstQueryOperator.cs
//
// [....]
//
// =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Threading;
namespace System.Linq.Parallel
{
///
/// First tries to discover the first element in the source, optionally matching a
/// predicate. All partitions search in parallel, publish the lowest index for a
/// candidate match, and reach a barrier. Only the partition that "wins" the ----,
/// i.e. who found the candidate with the smallest index, will yield an element.
///
/// : UnaryQueryOperator
{
private readonly Func m_predicate; // The optional predicate used during the search.
private readonly bool m_prematureMergeNeeded; // Whether to prematurely merge the input of this operator.
//----------------------------------------------------------------------------------------
// Initializes a new first operator.
//
// Arguments:
// child - the child whose data we will reverse
//
internal FirstQueryOperator(IEnumerable child, Func predicate)
:base(child)
{
Contract.Assert(child != null, "child data source cannot be null");
m_predicate = predicate;
m_prematureMergeNeeded = Child.OrdinalIndexState.IsWorseThan(OrdinalIndexState.Increasing);
}
//---------------------------------------------------------------------------------------
// 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)
{
OrdinalIndexState inputIndexState = inputStream.OrdinalIndexState;
PartitionedStream intKeyStream;
int partitionCount = inputStream.PartitionCount;
// If the index is not at least increasing, we need to reindex.
if (m_prematureMergeNeeded)
{
ListQueryResults listResults = ExecuteAndCollectResults(inputStream, partitionCount, Child.OutputOrdered, preferStriping, settings);
intKeyStream = listResults.GetPartitionedStream();
}
else
{
Contract.Assert(typeof(TKey) == typeof(int));
intKeyStream = (PartitionedStream)(object)inputStream;
}
// Generate the shared data.
Shared sharedFirstCandidate = new Shared(-1);
CountdownEvent sharedBarrier = new CountdownEvent(partitionCount);
PartitionedStream outputStream = new PartitionedStream(
partitionCount, Util.GetDefaultComparer(), OrdinalIndexState.Shuffled);
for (int i = 0; i < partitionCount; i++)
{
outputStream[i] = new FirstQueryOperatorEnumerator(
intKeyStream[i], m_predicate, sharedFirstCandidate, sharedBarrier, 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 ParallelEnumerable.First().");
throw new NotSupportedException();
}
//---------------------------------------------------------------------------------------
// Whether this operator performs a premature merge.
//
internal override bool LimitsParallelism
{
get { return m_prematureMergeNeeded; }
}
//----------------------------------------------------------------------------------------
// The enumerator type responsible for executing the first operation.
//
class FirstQueryOperatorEnumerator : QueryOperatorEnumerator
{
private QueryOperatorEnumerator m_source; // The data source to enumerate.
private Func m_predicate; // The optional predicate used during the search.
private bool m_alreadySearched; // Set once the enumerator has performed the search.
// Data shared among partitions.
private Shared m_sharedFirstCandidate; // The current first candidate.
private CountdownEvent m_sharedBarrier; // Shared barrier, signaled when partitions find their 1st element.
private CancellationToken m_cancellationToken; // Token used to cancel this operator.
//---------------------------------------------------------------------------------------
// Instantiates a new enumerator.
//
internal FirstQueryOperatorEnumerator(
QueryOperatorEnumerator source, Func predicate,
Shared sharedFirstCandidate, CountdownEvent sharedBarrier, CancellationToken cancellationToken)
{
Contract.Assert(source != null);
Contract.Assert(sharedFirstCandidate != null);
Contract.Assert(sharedBarrier != null);
m_source = source;
m_predicate = predicate;
m_sharedFirstCandidate = sharedFirstCandidate;
m_sharedBarrier = sharedBarrier;
m_cancellationToken = cancellationToken;
}
//----------------------------------------------------------------------------------------
// Straightforward IEnumerator methods.
//
internal override bool MoveNext(ref TSource currentElement, ref int currentKey)
{
Contract.Assert(m_source != null);
if (m_alreadySearched)
{
return false;
}
// Look for the lowest element.
TSource candidate = default(TSource);
int candidateIndex = -1;
try
{
int key = default(int);
int i = 0;
while (m_source.MoveNext(ref candidate, ref key))
{
if ((i++ & CancellationState.POLL_INTERVAL) == 0)
CancellationState.ThrowIfCanceled(m_cancellationToken);
// If the predicate is null or the current element satisfies it, we have found the
// current partition's "candidate" for the first element. Note it.
if (m_predicate == null || m_predicate(candidate))
{
candidateIndex = key;
// Try to swap our index with the shared one, so long as it's smaller.
int observedSharedIndex;
do
{
observedSharedIndex = m_sharedFirstCandidate.Value;
} while ((observedSharedIndex == -1 || candidateIndex < observedSharedIndex) &&
Interlocked.CompareExchange(ref m_sharedFirstCandidate.Value, candidateIndex, observedSharedIndex) != observedSharedIndex);
break;
}
else if (m_sharedFirstCandidate.Value != -1 && key > m_sharedFirstCandidate.Value)
{
// We've scanned past another partition's best element. Bail.
break;
}
}
}
finally
{
// No matter whether we exit due to an exception or normal completion, we must ensure
// that we signal other partitions that we have completed. Otherwise, we can cause deadlocks.
m_sharedBarrier.Signal();
}
m_alreadySearched = true;
// Only if we might be a candidate do we wait.
if (candidateIndex != -1)
{
m_sharedBarrier.Wait(m_cancellationToken);
// Now re-read the shared index. If it's the same as ours, we won and return true.
if (m_sharedFirstCandidate.Value == candidateIndex)
{
currentElement = candidate;
currentKey = 0; // 1st (and only) element, so we hardcode the output index to 0.
return true;
}
}
// If we got here, we didn't win. Return false.
return false;
}
protected override void Dispose(bool disposing)
{
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
- WinEventWrap.cs
- FixedDocumentPaginator.cs
- Package.cs
- Oid.cs
- COM2PropertyDescriptor.cs
- PackageRelationshipCollection.cs
- DbProviderManifest.cs
- DocumentEventArgs.cs
- WebResourceUtil.cs
- RectangleHotSpot.cs
- PasswordTextContainer.cs
- WebContext.cs
- IfAction.cs
- EntityDataSourceViewSchema.cs
- XsltSettings.cs
- LocatorBase.cs
- GenerateScriptTypeAttribute.cs
- EntryIndex.cs
- CheckBoxField.cs
- SqlConnectionFactory.cs
- OpacityConverter.cs
- RegexMatchCollection.cs
- keycontainerpermission.cs
- PropertyTabAttribute.cs
- BaseCodePageEncoding.cs
- WindowsListBox.cs
- Pool.cs
- TailCallAnalyzer.cs
- SelectorItemAutomationPeer.cs
- ConsoleCancelEventArgs.cs
- TypeConstant.cs
- BitmapDecoder.cs
- PopupEventArgs.cs
- CodeAttributeArgument.cs
- DataColumnChangeEvent.cs
- Wildcard.cs
- BaseValidator.cs
- RangeValuePatternIdentifiers.cs
- FloatUtil.cs
- ToolStripDropDown.cs
- BCryptSafeHandles.cs
- ReturnValue.cs
- SearchForVirtualItemEventArgs.cs
- CharacterString.cs
- XmlSortKeyAccumulator.cs
- Array.cs
- PropertyOverridesDialog.cs
- SqlExpander.cs
- SqlXml.cs
- FormatterServices.cs
- TransportChannelListener.cs
- WebAdminConfigurationHelper.cs
- ExtendedPropertyDescriptor.cs
- BrowserDefinition.cs
- elementinformation.cs
- BuildManager.cs
- SoapCodeExporter.cs
- RuntimeHelpers.cs
- HiddenField.cs
- VBIdentifierDesigner.xaml.cs
- DBParameter.cs
- FrameworkTemplate.cs
- DoubleCollectionValueSerializer.cs
- SqlUdtInfo.cs
- ContextInformation.cs
- DataServiceConfiguration.cs
- SecurityTokenAttachmentMode.cs
- GroupBox.cs
- MatrixCamera.cs
- SessionStateContainer.cs
- FrameworkReadOnlyPropertyMetadata.cs
- SecurityManager.cs
- RtfNavigator.cs
- CounterCreationDataCollection.cs
- GridViewRowPresenterBase.cs
- CheckBox.cs
- Bits.cs
- DataServiceContext.cs
- MetadataItem.cs
- EventHandlerList.cs
- PermissionListSet.cs
- Rules.cs
- BitmapInitialize.cs
- Command.cs
- FrameworkElementAutomationPeer.cs
- XmlWhitespace.cs
- EntityParameter.cs
- Knowncolors.cs
- LinqDataSourceValidationException.cs
- _BufferOffsetSize.cs
- XmlElementElement.cs
- StringAnimationUsingKeyFrames.cs
- StringWriter.cs
- ExtenderControl.cs
- StringConverter.cs
- StateChangeEvent.cs
- PropertyGridCommands.cs
- BooleanFacetDescriptionElement.cs
- Timeline.cs
- XpsSerializerWriter.cs