Code:
/ 4.0 / 4.0 / untmp / DEVDIV_TFS / Dev10 / Releases / RTMRel / ndp / fx / src / Core / System / Linq / Parallel / Merging / MergeExecutor.cs / 1305376 / MergeExecutor.cs
// ==++== // // Copyright (c) Microsoft Corporation. All rights reserved. // // ==--== // =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+ // // MergeExecutor.cs // //[....] // // =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=- using System.Collections; using System.Collections.Generic; using System.Threading; using System.Threading.Tasks; using System.Diagnostics.Contracts; namespace System.Linq.Parallel { ////// Drives execution of an actual merge operation, including creating channel data /// structures and scheduling parallel work as appropriate. The algorithms used /// internally are parameterized based on the type of data in the partitions; e.g. /// if an order preserved stream is found, the merge will automatically use an /// order preserving merge, and so forth. /// ///internal class MergeExecutor : IEnumerable { // Many internal algorithms are parameterized based on the data. The IMergeHelper // is the pluggable interface whose implementations perform those algorithms. private IMergeHelper m_mergeHelper; // Private constructor. MergeExecutor should only be constructed via the // MergeExecutor.Execute static method. private MergeExecutor() { } //------------------------------------------------------------------------------------ // Creates and executes a new merge executor object. // // Arguments: // partitions - the partitions whose data will be merged into one stream // ignoreOutput - if true, we are enumerating "for effect", and we won't actually // generate data in the output stream // pipeline - whether to use a pipelined merge or not. // isOrdered - whether to perform an ordering merge. // internal static MergeExecutor Execute ( PartitionedStream partitions, bool ignoreOutput, ParallelMergeOptions options, TaskScheduler taskScheduler, bool isOrdered, CancellationState cancellationState, int queryId) { Contract.Assert(partitions != null); Contract.Assert(partitions.PartitionCount > 0); Contract.Assert(!ignoreOutput || options == ParallelMergeOptions.FullyBuffered, "@BUGBUG: pipelining w/ no output not supported -- need it?"); MergeExecutor mergeExecutor = new MergeExecutor (); if (isOrdered && !ignoreOutput) { if (options != ParallelMergeOptions.FullyBuffered && !partitions.OrdinalIndexState.IsWorseThan(OrdinalIndexState.Increasing)) { Contract.Assert(options == ParallelMergeOptions.NotBuffered || options == ParallelMergeOptions.AutoBuffered); bool autoBuffered = (options == ParallelMergeOptions.AutoBuffered); if (partitions.PartitionCount > 1) { // We use a pipelining ordered merge mergeExecutor.m_mergeHelper = new OrderPreservingPipeliningMergeHelper ( (PartitionedStream )(object)partitions, taskScheduler, cancellationState, autoBuffered, queryId); } else { // When DOP=1, the default merge simply returns the single producer enumerator to the consumer. This way, ordering // does not add any extra overhead, and no producer task needs to be scheduled. mergeExecutor.m_mergeHelper = new DefaultMergeHelper ( partitions, false, options, taskScheduler, cancellationState, queryId); } } else { // We use a stop-and-go ordered merge helper mergeExecutor.m_mergeHelper = new OrderPreservingMergeHelper (partitions, taskScheduler, cancellationState, queryId); } } else { // We use a default - unordered - merge helper. mergeExecutor.m_mergeHelper = new DefaultMergeHelper (partitions, ignoreOutput, options, taskScheduler, cancellationState, queryId); } mergeExecutor.Execute(); return mergeExecutor; } //----------------------------------------------------------------------------------- // Initiates execution of the merge. // private void Execute() { Contract.Assert(m_mergeHelper != null); m_mergeHelper.Execute(); } //----------------------------------------------------------------------------------- // Returns an enumerator that will yield elements from the resulting merged data // stream. // IEnumerator IEnumerable.GetEnumerator() { return ((IEnumerable )this).GetEnumerator(); } public IEnumerator GetEnumerator() { Contract.Assert(m_mergeHelper != null); return m_mergeHelper.GetEnumerator(); } //----------------------------------------------------------------------------------- // Returns the merged results as an array. // internal TInputOutput[] GetResultsAsArray() { return m_mergeHelper.GetResultsAsArray(); } //------------------------------------------------------------------------------------ // This internal helper method is used to generate a set of asynchronous channels. // The algorithm used by each channel contains the necessary synchronizationis to // ensure it is suitable for pipelined consumption. // // Arguments: // partitionsCount - the number of partitions for which to create new channels. // // Return Value: // An array of asynchronous channels, one for each partition. // internal static AsynchronousChannel [] MakeAsynchronousChannels(int partitionCount, ParallelMergeOptions options, CancellationToken cancellationToken) { AsynchronousChannel [] channels = new AsynchronousChannel [partitionCount]; Contract.Assert(options == ParallelMergeOptions.NotBuffered || options == ParallelMergeOptions.AutoBuffered); TraceHelpers.TraceInfo("MergeExecutor::MakeChannels: setting up {0} async channels in prep for pipeline", partitionCount); // If we are pipelining, we need a channel that contains the necessary synchronization // in it. We choose a bounded/blocking channel data structure: bounded so that we can // limit the amount of memory overhead used by the query by putting a cap on the // buffer size into which producers place data, and blocking so that the consumer can // wait for additional data to arrive in the case that it's found to be empty. int chunkSize = 0; // 0 means automatic chunk size if (options == ParallelMergeOptions.NotBuffered) { chunkSize = 1; } for (int i = 0; i < channels.Length; i++) { channels[i] = new AsynchronousChannel (chunkSize, cancellationToken); } return channels; } //----------------------------------------------------------------------------------- // This internal helper method is used to generate a set of synchronous channels. // The channel data structure used has been optimized for sequential execution and // does not support pipelining. // // Arguments: // partitionsCount - the number of partitions for which to create new channels. // // Return Value: // An array of synchronous channels, one for each partition. // internal static SynchronousChannel [] MakeSynchronousChannels(int partitionCount) { SynchronousChannel [] channels = new SynchronousChannel [partitionCount]; TraceHelpers.TraceInfo("MergeExecutor::MakeChannels: setting up {0} channels in prep for stop-and-go", partitionCount); // We just build up the results in memory using simple, dynamically growable FIFO queues. for (int i = 0; i < channels.Length; i++) { channels[i] = new SynchronousChannel (); } return channels; } } } // 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
- FlowDocumentPaginator.cs
- ApplyHostConfigurationBehavior.cs
- Enum.cs
- SchemaManager.cs
- BCryptHashAlgorithm.cs
- RawStylusInputCustomDataList.cs
- ComplexType.cs
- ResourceAttributes.cs
- SHA512Cng.cs
- OpCellTreeNode.cs
- _RequestLifetimeSetter.cs
- ReadOnlyObservableCollection.cs
- ModelPropertyDescriptor.cs
- RepeaterItemEventArgs.cs
- WebColorConverter.cs
- XmlBoundElement.cs
- TextEmbeddedObject.cs
- CallContext.cs
- RedistVersionInfo.cs
- ForeignConstraint.cs
- GreaterThan.cs
- StackOverflowException.cs
- SaveFileDialog.cs
- webbrowsersite.cs
- DBParameter.cs
- OneWayElement.cs
- HttpCookieCollection.cs
- TabItemWrapperAutomationPeer.cs
- TablePattern.cs
- TriggerAction.cs
- FocusTracker.cs
- XmlCharType.cs
- WebDescriptionAttribute.cs
- MultiPageTextView.cs
- XmlTypeMapping.cs
- ListViewInsertEventArgs.cs
- BufferModeSettings.cs
- Margins.cs
- KeyProperty.cs
- DateTimeConverter.cs
- PageAsyncTask.cs
- XsltContext.cs
- ExpressionNode.cs
- DataRecord.cs
- PrintDocument.cs
- PerformanceCounterPermissionEntry.cs
- MethodBody.cs
- altserialization.cs
- StateChangeEvent.cs
- UnsafeNativeMethods.cs
- CmsInterop.cs
- CustomAttribute.cs
- ResolveNameEventArgs.cs
- VerticalAlignConverter.cs
- X509Utils.cs
- MatrixValueSerializer.cs
- TypographyProperties.cs
- MeshGeometry3D.cs
- DecimalAnimation.cs
- Site.cs
- DragStartedEventArgs.cs
- TemplateBuilder.cs
- TreeViewItemAutomationPeer.cs
- OdbcCommandBuilder.cs
- InstallerTypeAttribute.cs
- ScrollContentPresenter.cs
- SiteMapNodeItem.cs
- QilReference.cs
- RegexBoyerMoore.cs
- GridViewRowCollection.cs
- AutoResetEvent.cs
- PlanCompiler.cs
- BaseCodeDomTreeGenerator.cs
- RuleRefElement.cs
- MD5CryptoServiceProvider.cs
- PropertiesTab.cs
- HostingEnvironmentSection.cs
- InstanceKey.cs
- MSHTMLHost.cs
- WithParamAction.cs
- WebPartsPersonalization.cs
- ProfileInfo.cs
- DateTimeUtil.cs
- SortFieldComparer.cs
- OutputCacheSection.cs
- Context.cs
- Section.cs
- ExpressionCopier.cs
- DirectoryInfo.cs
- SortExpressionBuilder.cs
- MouseGesture.cs
- BitStream.cs
- TriggerBase.cs
- MD5CryptoServiceProvider.cs
- Certificate.cs
- SQLGuid.cs
- NonSerializedAttribute.cs
- versioninfo.cs
- MetaTable.cs
- PeerNode.cs