Code:
/ 4.0 / 4.0 / DEVDIV_TFS / Dev10 / Releases / RTMRel / ndp / fx / src / Core / System / Linq / Parallel / Scheduling / OrderPreservingSpoolingTask.cs / 1305376 / OrderPreservingSpoolingTask.cs
// ==++== // // Copyright (c) Microsoft Corporation. All rights reserved. // // ==--== // =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+ // // OrderPreservingSpoolingTask.cs // //[....] // // =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=- using System.Threading; using System.Threading.Tasks; using System.Diagnostics.Contracts; namespace System.Linq.Parallel { ////// A spooling task handles marshaling data from a producer to a consumer. It's given /// a single enumerator object that contains all of the production algorithms, a single /// destination channel from which consumers draw results, and (optionally) a /// synchronization primitive using which to notify asynchronous consumers. This /// particular task variant preserves sort order in the final data. /// ////// internal class OrderPreservingSpoolingTask : SpoolingTaskBase { private Shared m_results; // The destination array cell into which data is placed. private SortHelper m_sortHelper; // A helper that performs the sorting. //------------------------------------------------------------------------------------ // Creates, but does not execute, a new spooling task. // // Arguments: // taskIndex - the unique index of this task // ordinalIndexState - the state of ordinal indices // source - the producer enumerator // destination - the destination channel into which to spool elements // // Assumptions: // Source cannot be null, although the other arguments may be. // private OrderPreservingSpoolingTask( int taskIndex, QueryTaskGroupState groupState, Shared results, SortHelper sortHelper) : base(taskIndex, groupState) { Contract.Assert(groupState != null); Contract.Assert(results != null); Contract.Assert(sortHelper != null); m_results = results; m_sortHelper = sortHelper; } //----------------------------------------------------------------------------------- // Creates and begins execution of a new spooling task. If pipelineMerges is specified, // we will execute the task asynchronously; otherwise, this is done synchronously, // and by the time this API has returned all of the results have been produced. // // Arguments: // source - the producer enumerator // destination - the destination channel into which to spool elements // ordinalIndexState - state of the index of the input to the merge // // Assumptions: // Source cannot be null, although the other arguments may be. // internal static void Spool( QueryTaskGroupState groupState, PartitionedStream partitions, Shared results, TaskScheduler taskScheduler) { Contract.Assert(groupState != null); Contract.Assert(partitions != null); Contract.Assert(results != null); Contract.Assert(results.Value == null); // Determine how many async tasks to create. int maxToRunInParallel = partitions.PartitionCount - 1; // Generate a set of sort helpers. SortHelper [] sortHelpers = SortHelper .GenerateSortHelpers(partitions, groupState); // Ensure all tasks in this query are parented under a common root. Task rootTask = new Task( () => { // Create tasks that will enumerate the partitions in parallel. We'll use the current // thread for one task and then block before returning to the caller, until all results // have been accumulated. Pipelining is not supported by sort merges. for (int i = 0; i < maxToRunInParallel; i++) { TraceHelpers.TraceInfo("OrderPreservingSpoolingTask::Spool: Running partition[{0}] asynchronously", i); QueryTask asyncTask = new OrderPreservingSpoolingTask ( i, groupState, results, sortHelpers[i]); asyncTask.RunAsynchronously(taskScheduler); } // Run one task synchronously on the current thread. TraceHelpers.TraceInfo("OrderPreservingSpoolingTask::Spool: Running partition[{0}] synchronously", maxToRunInParallel); QueryTask syncTask = new OrderPreservingSpoolingTask ( maxToRunInParallel, groupState, results, sortHelpers[maxToRunInParallel]); syncTask.RunSynchronously(taskScheduler); }); // Begin the query on the calling thread. groupState.QueryBegin(rootTask); // We don't want to return until the task is finished. Run it on the calling thread. rootTask.RunSynchronously(taskScheduler); // Destroy the state associated with our sort helpers. for (int i = 0; i < sortHelpers.Length; i++) { sortHelpers[i].Dispose(); } // End the query, which has the effect of propagating any unhandled exceptions. groupState.QueryEnd(false); } //----------------------------------------------------------------------------------- // This method is responsible for enumerating results and enqueueing them to // the output channel(s) as appropriate. Each base class implements its own. // protected override void SpoolingWork() { Contract.Assert(m_sortHelper != null); // This task must perform a sort just prior to handing data to the merge. // We just defer to a sort helper object for this task. TInputOutput[] sortedOutput = m_sortHelper.Sort(); if (!m_groupState.CancellationState.MergedCancellationToken.IsCancellationRequested) { // The 0th task is responsible for communicating the results to the merging infrastructure. // By this point, the results have been sorted, so we just publish a reference to the array. if (m_taskIndex == 0) { Contract.Assert(sortedOutput != null); m_results.Value = sortedOutput; } } } } } // File provided for Reference Use Only by Microsoft Corporation (c) 2007. // ==++== // // Copyright (c) Microsoft Corporation. All rights reserved. // // ==--== // =+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+ // // OrderPreservingSpoolingTask.cs // // [....] // // =-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=-=- using System.Threading; using System.Threading.Tasks; using System.Diagnostics.Contracts; namespace System.Linq.Parallel { ////// A spooling task handles marshaling data from a producer to a consumer. It's given /// a single enumerator object that contains all of the production algorithms, a single /// destination channel from which consumers draw results, and (optionally) a /// synchronization primitive using which to notify asynchronous consumers. This /// particular task variant preserves sort order in the final data. /// ////// internal class OrderPreservingSpoolingTask : SpoolingTaskBase { private Shared m_results; // The destination array cell into which data is placed. private SortHelper m_sortHelper; // A helper that performs the sorting. //------------------------------------------------------------------------------------ // Creates, but does not execute, a new spooling task. // // Arguments: // taskIndex - the unique index of this task // ordinalIndexState - the state of ordinal indices // source - the producer enumerator // destination - the destination channel into which to spool elements // // Assumptions: // Source cannot be null, although the other arguments may be. // private OrderPreservingSpoolingTask( int taskIndex, QueryTaskGroupState groupState, Shared results, SortHelper sortHelper) : base(taskIndex, groupState) { Contract.Assert(groupState != null); Contract.Assert(results != null); Contract.Assert(sortHelper != null); m_results = results; m_sortHelper = sortHelper; } //----------------------------------------------------------------------------------- // Creates and begins execution of a new spooling task. If pipelineMerges is specified, // we will execute the task asynchronously; otherwise, this is done synchronously, // and by the time this API has returned all of the results have been produced. // // Arguments: // source - the producer enumerator // destination - the destination channel into which to spool elements // ordinalIndexState - state of the index of the input to the merge // // Assumptions: // Source cannot be null, although the other arguments may be. // internal static void Spool( QueryTaskGroupState groupState, PartitionedStream partitions, Shared results, TaskScheduler taskScheduler) { Contract.Assert(groupState != null); Contract.Assert(partitions != null); Contract.Assert(results != null); Contract.Assert(results.Value == null); // Determine how many async tasks to create. int maxToRunInParallel = partitions.PartitionCount - 1; // Generate a set of sort helpers. SortHelper [] sortHelpers = SortHelper .GenerateSortHelpers(partitions, groupState); // Ensure all tasks in this query are parented under a common root. Task rootTask = new Task( () => { // Create tasks that will enumerate the partitions in parallel. We'll use the current // thread for one task and then block before returning to the caller, until all results // have been accumulated. Pipelining is not supported by sort merges. for (int i = 0; i < maxToRunInParallel; i++) { TraceHelpers.TraceInfo("OrderPreservingSpoolingTask::Spool: Running partition[{0}] asynchronously", i); QueryTask asyncTask = new OrderPreservingSpoolingTask ( i, groupState, results, sortHelpers[i]); asyncTask.RunAsynchronously(taskScheduler); } // Run one task synchronously on the current thread. TraceHelpers.TraceInfo("OrderPreservingSpoolingTask::Spool: Running partition[{0}] synchronously", maxToRunInParallel); QueryTask syncTask = new OrderPreservingSpoolingTask ( maxToRunInParallel, groupState, results, sortHelpers[maxToRunInParallel]); syncTask.RunSynchronously(taskScheduler); }); // Begin the query on the calling thread. groupState.QueryBegin(rootTask); // We don't want to return until the task is finished. Run it on the calling thread. rootTask.RunSynchronously(taskScheduler); // Destroy the state associated with our sort helpers. for (int i = 0; i < sortHelpers.Length; i++) { sortHelpers[i].Dispose(); } // End the query, which has the effect of propagating any unhandled exceptions. groupState.QueryEnd(false); } //----------------------------------------------------------------------------------- // This method is responsible for enumerating results and enqueueing them to // the output channel(s) as appropriate. Each base class implements its own. // protected override void SpoolingWork() { Contract.Assert(m_sortHelper != null); // This task must perform a sort just prior to handing data to the merge. // We just defer to a sort helper object for this task. TInputOutput[] sortedOutput = m_sortHelper.Sort(); if (!m_groupState.CancellationState.MergedCancellationToken.IsCancellationRequested) { // The 0th task is responsible for communicating the results to the merging infrastructure. // By this point, the results have been sorted, so we just publish a reference to the array. if (m_taskIndex == 0) { Contract.Assert(sortedOutput != null); m_results.Value = sortedOutput; } } } } } // 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
- ListControlConvertEventArgs.cs
- DataGridRow.cs
- FrameworkContentElementAutomationPeer.cs
- DocumentsTrace.cs
- FixedHighlight.cs
- SqlConnectionStringBuilder.cs
- XmlObjectSerializerReadContextComplexJson.cs
- PathSegment.cs
- EventKeyword.cs
- ModelPerspective.cs
- RadioButtonRenderer.cs
- TreeNodeConverter.cs
- Set.cs
- DiscoveryClientDocuments.cs
- ThreadAttributes.cs
- EmptyQuery.cs
- BatchStream.cs
- GlobalizationSection.cs
- ApplicationTrust.cs
- PropertyFilterAttribute.cs
- MeshGeometry3D.cs
- CultureInfoConverter.cs
- LineInfo.cs
- WebDisplayNameAttribute.cs
- ComponentManagerBroker.cs
- ServiceDescriptionData.cs
- ActivityCodeDomSerializer.cs
- EnvelopedPkcs7.cs
- InternalResources.cs
- SignatureToken.cs
- CqlLexer.cs
- Image.cs
- FormViewUpdatedEventArgs.cs
- DesignerValidationSummaryAdapter.cs
- ConfigXmlText.cs
- ObjectPersistData.cs
- CellParaClient.cs
- FreezableCollection.cs
- WebPartCancelEventArgs.cs
- TimeSpanConverter.cs
- Vector.cs
- Clock.cs
- TextServicesCompartmentContext.cs
- WCFServiceClientProxyGenerator.cs
- ProxyManager.cs
- ImageListDesigner.cs
- DeflateEmulationStream.cs
- SystemWebExtensionsSectionGroup.cs
- PackageRelationship.cs
- DropSourceBehavior.cs
- X509PeerCertificateElement.cs
- PointLight.cs
- EmissiveMaterial.cs
- WebPartConnection.cs
- UnsafeNativeMethods.cs
- AndCondition.cs
- PropertyCondition.cs
- ToolStripSettings.cs
- ResourceAttributes.cs
- NestPullup.cs
- XmlSchemaComplexContent.cs
- SizeAnimationClockResource.cs
- DataRecordObjectView.cs
- WebPartConnectionsCancelEventArgs.cs
- GenerateDerivedKeyRequest.cs
- RewritingSimplifier.cs
- Function.cs
- HtmlShimManager.cs
- CanonicalFontFamilyReference.cs
- ColorAnimationBase.cs
- InternalsVisibleToAttribute.cs
- BuildManager.cs
- RemoteWebConfigurationHost.cs
- StateWorkerRequest.cs
- XmlFormatExtensionAttribute.cs
- TagMapInfo.cs
- SqlProfileProvider.cs
- OleDbFactory.cs
- MiniConstructorInfo.cs
- RootBuilder.cs
- XmlQueryStaticData.cs
- EntityContainerAssociationSetEnd.cs
- QfeChecker.cs
- PointCollectionConverter.cs
- ShimAsPublicXamlType.cs
- _SpnDictionary.cs
- DbConnectionPool.cs
- MultipleCopiesCollection.cs
- GraphicsState.cs
- IsolatedStoragePermission.cs
- OdbcParameterCollection.cs
- ModelVisual3D.cs
- IODescriptionAttribute.cs
- DecimalKeyFrameCollection.cs
- ProxyAttribute.cs
- WebPartPersonalization.cs
- SimpleParser.cs
- RelationshipDetailsRow.cs
- DataServiceQuery.cs
- BitmapFrameDecode.cs