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
![Network programming in C#, Network Programming in VB.NET, Network Programming in .NET](/images/book.jpg)
This book is available now!
Buy at Amazon US or
Buy at Amazon UK
- DataGridViewSelectedRowCollection.cs
- QueueException.cs
- ItemList.cs
- NavigationWindow.cs
- KeyValuePairs.cs
- StackBuilderSink.cs
- Freezable.cs
- FileCodeGroup.cs
- ImageSource.cs
- Point.cs
- DataGridViewRowPostPaintEventArgs.cs
- CompilerErrorCollection.cs
- Constraint.cs
- WebPartEditorOkVerb.cs
- BitmapMetadataEnumerator.cs
- EntityConnection.cs
- BrushMappingModeValidation.cs
- PropertyGridEditorPart.cs
- ModuleElement.cs
- BrowserCapabilitiesFactory.cs
- UnauthorizedAccessException.cs
- HierarchicalDataBoundControlAdapter.cs
- HTMLTagNameToTypeMapper.cs
- StringOutput.cs
- MouseGesture.cs
- TextBoxBase.cs
- Debug.cs
- ScaleTransform.cs
- Substitution.cs
- CodeGeneratorAttribute.cs
- UnsafeMethods.cs
- CryptoApi.cs
- SimpleWorkerRequest.cs
- ExpressionBuilder.cs
- ApplicationManager.cs
- InProcStateClientManager.cs
- WinFormsSpinner.cs
- Currency.cs
- ToolZoneDesigner.cs
- NaturalLanguageHyphenator.cs
- Util.cs
- DesignerTransaction.cs
- Lock.cs
- SqlCommandBuilder.cs
- EdmSchemaError.cs
- HttpSessionStateWrapper.cs
- SimpleBitVector32.cs
- PropertyDescriptorCollection.cs
- WeakEventManager.cs
- ParallelTimeline.cs
- CodeCommentStatement.cs
- TextFragmentEngine.cs
- TaskScheduler.cs
- EntityCodeGenerator.cs
- MultiBindingExpression.cs
- DataBindingCollection.cs
- EventLogEntryCollection.cs
- CommandField.cs
- InitiatorServiceModelSecurityTokenRequirement.cs
- ListViewDataItem.cs
- Selector.cs
- WebControlsSection.cs
- X509Chain.cs
- DataGridCell.cs
- Span.cs
- ConfigurationProperty.cs
- PkcsUtils.cs
- ViewStateChangedEventArgs.cs
- XPathAxisIterator.cs
- PropertyDescriptor.cs
- WebPartConnectVerb.cs
- RepeatInfo.cs
- RequestCacheEntry.cs
- DispatchWrapper.cs
- PolyBezierSegment.cs
- CodePageEncoding.cs
- DocumentPageView.cs
- ThreadAttributes.cs
- XmlSchemaSimpleContent.cs
- PersistenceParticipant.cs
- PrintDialog.cs
- EmptyStringExpandableObjectConverter.cs
- InkCanvas.cs
- MeshGeometry3D.cs
- EpmTargetTree.cs
- DataGridItem.cs
- StringSorter.cs
- Section.cs
- AuthStoreRoleProvider.cs
- CatalogZone.cs
- webproxy.cs
- SystemIPInterfaceStatistics.cs
- SimpleApplicationHost.cs
- ClusterRegistryConfigurationProvider.cs
- InternalBase.cs
- Expressions.cs
- RequestCacheValidator.cs
- ToolStripContainer.cs
- WindowsAuthenticationModule.cs
- PeerNameRecordCollection.cs