Code:
/ 4.0 / 4.0 / DEVDIV_TFS / Dev10 / Releases / RTMRel / ndp / fx / src / Core / System / Linq / Parallel / QueryOperators / Unary / LastQueryOperator.cs / 1305376 / LastQueryOperator.cs
// ==++==
//
// Copyright (c) Microsoft Corporation. All rights reserved.
//
// ==--==
// =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
//
// LastQueryOperator.cs
//
// [....]
//
// =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
using System.Collections.Generic;
using System.Threading;
using System.Diagnostics.Contracts;
namespace System.Linq.Parallel
{
///
/// Last tries to discover the last element in the source, optionally matching a
/// predicate. All partitions search in parallel, publish the greatest index for a
/// candidate match, and reach a barrier. Only the partition that "wins" the ----,
/// i.e. who found the candidate with the largest index, will yield an element.
///
/// @
internal sealed class LastQueryOperator : 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 last operator.
//
// Arguments:
// child - the child whose data we will reverse
//
internal LastQueryOperator(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)
{
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)
{
PartitionedStream intKeyStream;
int partitionCount = inputStream.PartitionCount;
// If the index is not at least increasing, we need to reindex.
if (m_prematureMergeNeeded)
{
intKeyStream = ExecuteAndCollectResults(inputStream, partitionCount, Child.OutputOrdered, preferStriping, settings).GetPartitionedStream();
}
else
{
Contract.Assert(inputStream is PartitionedStream);
intKeyStream = (PartitionedStream)(object)inputStream;
}
// Generate the shared data.
Shared sharedLastCandidate = new Shared(-1);
CountdownEvent sharedBarrier = new CountdownEvent(partitionCount);
PartitionedStream outputStream =
new PartitionedStream(partitionCount, intKeyStream.KeyComparer, OrdinalIndexState.Shuffled);
for (int i = 0; i < partitionCount; i++)
{
outputStream[i] = new LastQueryOperatorEnumerator(
intKeyStream[i], m_predicate, sharedLastCandidate, 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 last operation.
//
class LastQueryOperatorEnumerator : 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_sharedLastCandidate; // The current last 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 LastQueryOperatorEnumerator(
QueryOperatorEnumerator source, Func predicate,
Shared sharedLastCandidate, CountdownEvent sharedBarrier, CancellationToken cancelToken)
{
Contract.Assert(source != null);
Contract.Assert(sharedLastCandidate != null);
Contract.Assert(sharedBarrier != null);
m_source = source;
m_predicate = predicate;
m_sharedLastCandidate = sharedLastCandidate;
m_sharedBarrier = sharedBarrier;
m_cancellationToken = cancelToken;
}
//----------------------------------------------------------------------------------------
// 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 greatest element.
TSource candidate = default(TSource);
int candidateIndex = -1;
try
{
TSource current = default(TSource);
int key = default(int);
int loopCount = 0; //counter to help with cancellation
while (m_source.MoveNext(ref current, ref key))
{
if ((loopCount & CancellationState.POLL_INTERVAL) == 0)
CancellationState.ThrowIfCanceled(m_cancellationToken);
// If the predicate is null or the current element satisfies it, we will remember
// it as the current partition's candidate for the last element, and move on.
if (m_predicate == null || m_predicate(current))
{
candidate = current;
candidateIndex = key;
}
loopCount++;
}
// If we found a candidate element, try to publish it, so long as it's greater.
if (candidateIndex != -1)
{
int observedSharedIndex;
do
{
observedSharedIndex = m_sharedLastCandidate.Value;
}
while ((observedSharedIndex == -1 || candidateIndex > observedSharedIndex) &&
Interlocked.CompareExchange(ref m_sharedLastCandidate.Value, candidateIndex, observedSharedIndex) != observedSharedIndex);
}
}
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 have 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_sharedLastCandidate.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.
//
// ==--==
// =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
//
// LastQueryOperator.cs
//
// [....]
//
// =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-
using System.Collections.Generic;
using System.Threading;
using System.Diagnostics.Contracts;
namespace System.Linq.Parallel
{
///
/// Last tries to discover the last element in the source, optionally matching a
/// predicate. All partitions search in parallel, publish the greatest index for a
/// candidate match, and reach a barrier. Only the partition that "wins" the ----,
/// i.e. who found the candidate with the largest index, will yield an element.
///
/// @
internal sealed class LastQueryOperator : 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 last operator.
//
// Arguments:
// child - the child whose data we will reverse
//
internal LastQueryOperator(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)
{
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)
{
PartitionedStream intKeyStream;
int partitionCount = inputStream.PartitionCount;
// If the index is not at least increasing, we need to reindex.
if (m_prematureMergeNeeded)
{
intKeyStream = ExecuteAndCollectResults(inputStream, partitionCount, Child.OutputOrdered, preferStriping, settings).GetPartitionedStream();
}
else
{
Contract.Assert(inputStream is PartitionedStream);
intKeyStream = (PartitionedStream)(object)inputStream;
}
// Generate the shared data.
Shared sharedLastCandidate = new Shared(-1);
CountdownEvent sharedBarrier = new CountdownEvent(partitionCount);
PartitionedStream outputStream =
new PartitionedStream(partitionCount, intKeyStream.KeyComparer, OrdinalIndexState.Shuffled);
for (int i = 0; i < partitionCount; i++)
{
outputStream[i] = new LastQueryOperatorEnumerator(
intKeyStream[i], m_predicate, sharedLastCandidate, 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 last operation.
//
class LastQueryOperatorEnumerator : 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_sharedLastCandidate; // The current last 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 LastQueryOperatorEnumerator(
QueryOperatorEnumerator source, Func predicate,
Shared sharedLastCandidate, CountdownEvent sharedBarrier, CancellationToken cancelToken)
{
Contract.Assert(source != null);
Contract.Assert(sharedLastCandidate != null);
Contract.Assert(sharedBarrier != null);
m_source = source;
m_predicate = predicate;
m_sharedLastCandidate = sharedLastCandidate;
m_sharedBarrier = sharedBarrier;
m_cancellationToken = cancelToken;
}
//----------------------------------------------------------------------------------------
// 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 greatest element.
TSource candidate = default(TSource);
int candidateIndex = -1;
try
{
TSource current = default(TSource);
int key = default(int);
int loopCount = 0; //counter to help with cancellation
while (m_source.MoveNext(ref current, ref key))
{
if ((loopCount & CancellationState.POLL_INTERVAL) == 0)
CancellationState.ThrowIfCanceled(m_cancellationToken);
// If the predicate is null or the current element satisfies it, we will remember
// it as the current partition's candidate for the last element, and move on.
if (m_predicate == null || m_predicate(current))
{
candidate = current;
candidateIndex = key;
}
loopCount++;
}
// If we found a candidate element, try to publish it, so long as it's greater.
if (candidateIndex != -1)
{
int observedSharedIndex;
do
{
observedSharedIndex = m_sharedLastCandidate.Value;
}
while ((observedSharedIndex == -1 || candidateIndex > observedSharedIndex) &&
Interlocked.CompareExchange(ref m_sharedLastCandidate.Value, candidateIndex, observedSharedIndex) != observedSharedIndex);
}
}
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 have 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_sharedLastCandidate.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
- ProtocolsSection.cs
- QueryStringHandler.cs
- NetworkInformationPermission.cs
- Random.cs
- BuildManagerHost.cs
- Italic.cs
- WebPartAddingEventArgs.cs
- MULTI_QI.cs
- IItemContainerGenerator.cs
- InvalidFilterCriteriaException.cs
- EmissiveMaterial.cs
- EnumBuilder.cs
- UInt16Storage.cs
- BinaryNegotiation.cs
- XmlAnyElementAttributes.cs
- XPathCompileException.cs
- ApplicationSecurityManager.cs
- ComplexTypeEmitter.cs
- NetworkInformationException.cs
- TextServicesPropertyRanges.cs
- CodeComment.cs
- MinimizableAttributeTypeConverter.cs
- FormsAuthenticationUser.cs
- ReadOnlyCollectionBase.cs
- DebuggerAttributes.cs
- RequiredAttributeAttribute.cs
- MouseGesture.cs
- Activator.cs
- LoginView.cs
- XsdValidatingReader.cs
- ControlIdConverter.cs
- CodeIdentifiers.cs
- WorkflowView.cs
- PixelFormat.cs
- CodeSnippetStatement.cs
- DataControlFieldCollection.cs
- FieldToken.cs
- XamlStyleSerializer.cs
- ViewStateException.cs
- PageTheme.cs
- BatchServiceHost.cs
- StorageScalarPropertyMapping.cs
- QueueTransferProtocol.cs
- TemplatedMailWebEventProvider.cs
- SRef.cs
- dataprotectionpermission.cs
- SqlGenerator.cs
- WindowsToolbarItemAsMenuItem.cs
- TypeConverterValueSerializer.cs
- ImageList.cs
- ImmComposition.cs
- MappableObjectManager.cs
- GetIsBrowserClientRequest.cs
- NumberSubstitution.cs
- EmptyEnumerable.cs
- DesignerToolboxInfo.cs
- PagesChangedEventArgs.cs
- AuthorizationSection.cs
- SettingsPropertyCollection.cs
- WsdlWriter.cs
- ProjectionCamera.cs
- WhitespaceRuleLookup.cs
- OdbcStatementHandle.cs
- VirtualPathUtility.cs
- XmlSchemaSimpleTypeList.cs
- X509CertificateClaimSet.cs
- ComplexTypeEmitter.cs
- DecimalAnimation.cs
- LambdaCompiler.Statements.cs
- HttpValueCollection.cs
- ArgumentNullException.cs
- EventSetter.cs
- Pair.cs
- MethodCallTranslator.cs
- ReflectionHelper.cs
- DetailsViewCommandEventArgs.cs
- SmtpLoginAuthenticationModule.cs
- User.cs
- VisualTreeHelper.cs
- ColumnHeader.cs
- SpotLight.cs
- DataGridViewLinkCell.cs
- Invariant.cs
- XmlDeclaration.cs
- TypeSystemProvider.cs
- DataGridViewSelectedColumnCollection.cs
- GridErrorDlg.cs
- HandleRef.cs
- SHA512Cng.cs
- JpegBitmapDecoder.cs
- ArglessEventHandlerProxy.cs
- DataContractSerializerMessageContractImporter.cs
- ArgumentReference.cs
- XmlAttributeProperties.cs
- JavascriptCallbackResponseProperty.cs
- Int64KeyFrameCollection.cs
- Int32AnimationUsingKeyFrames.cs
- CqlIdentifiers.cs
- AnnotationStore.cs
- SqlLiftWhereClauses.cs