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. /// ///internal sealed class ElementAtQueryOperator : 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) { IEnumerablechildAsSequential = 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. /// ///internal sealed class ElementAtQueryOperator : 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) { IEnumerablechildAsSequential = 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
- StaticResourceExtension.cs
- DataTableNameHandler.cs
- validationstate.cs
- TypeSystem.cs
- Attributes.cs
- ListViewGroupItemCollection.cs
- DtrList.cs
- AppDomainProtocolHandler.cs
- ConfigXmlWhitespace.cs
- HttpRequestCacheValidator.cs
- MetabaseSettings.cs
- PeerInputChannel.cs
- PrimitiveType.cs
- AnimationStorage.cs
- MsmqIntegrationMessagePool.cs
- WebPartConnectionsCancelEventArgs.cs
- AddInEnvironment.cs
- Utils.cs
- DecoratedNameAttribute.cs
- UIElementIsland.cs
- NativeMethods.cs
- DataBindingCollection.cs
- DocumentApplication.cs
- ManagementInstaller.cs
- TextProviderWrapper.cs
- DiscoveryClientProtocol.cs
- DataGridViewRowPostPaintEventArgs.cs
- MsmqInputChannel.cs
- WizardForm.cs
- GridViewCellAutomationPeer.cs
- ContextActivityUtils.cs
- DataGridViewAutoSizeColumnModeEventArgs.cs
- ConcurrentBag.cs
- ShaderRenderModeValidation.cs
- SafeLocalMemHandle.cs
- PageParser.cs
- MobileFormsAuthentication.cs
- TextLine.cs
- ValidatingPropertiesEventArgs.cs
- ProfileBuildProvider.cs
- EdmToObjectNamespaceMap.cs
- OleDragDropHandler.cs
- RelatedView.cs
- ContextMenuStrip.cs
- JapaneseLunisolarCalendar.cs
- SqlResolver.cs
- EasingKeyFrames.cs
- JournalEntryStack.cs
- IdlingCommunicationPool.cs
- ResXResourceSet.cs
- RSAProtectedConfigurationProvider.cs
- ElapsedEventArgs.cs
- TraceEventCache.cs
- UrlMappingsModule.cs
- ResourceDefaultValueAttribute.cs
- DebuggerAttributes.cs
- StyleCollection.cs
- MailHeaderInfo.cs
- PersonalizableTypeEntry.cs
- UpdatePanel.cs
- UriTemplateLiteralPathSegment.cs
- TextComposition.cs
- DataServiceSaveChangesEventArgs.cs
- MsmqProcessProtocolHandler.cs
- MatrixStack.cs
- EntityDataSourceQueryBuilder.cs
- RequestCacheManager.cs
- ComboBoxRenderer.cs
- Int32Converter.cs
- Identity.cs
- WorkflowStateRollbackService.cs
- HeaderCollection.cs
- DocumentOrderQuery.cs
- TcpConnectionPoolSettingsElement.cs
- CngUIPolicy.cs
- DataBindingExpressionBuilder.cs
- SizeAnimationUsingKeyFrames.cs
- MulticastIPAddressInformationCollection.cs
- RecommendedAsConfigurableAttribute.cs
- HttpCachePolicy.cs
- PaperSource.cs
- MyContact.cs
- WebHttpBindingCollectionElement.cs
- ConditionalDesigner.cs
- _SafeNetHandles.cs
- EmulateRecognizeCompletedEventArgs.cs
- Int32AnimationBase.cs
- TextPatternIdentifiers.cs
- JpegBitmapEncoder.cs
- HashHelper.cs
- PackageRelationshipCollection.cs
- NamespaceCollection.cs
- VScrollBar.cs
- TextEffectResolver.cs
- WebPartTransformerAttribute.cs
- EncryptedKey.cs
- StreamWriter.cs
- TrackingLocation.cs
- PriorityItem.cs
- XmlArrayAttribute.cs