Code:
/ 4.0 / 4.0 / DEVDIV_TFS / Dev10 / Releases / RTMRel / ndp / clr / src / ManagedLibraries / Remoting / Channels / CORE / RequestQueue.cs / 1305376 / RequestQueue.cs
// ==++== // // Copyright (c) Microsoft Corporation. All rights reserved. // // ==--== // // Request Queue // queues up the requests to avoid thread pool starvation, // making sure that there are always available threads to process requests //// This code has been taken from the XSP code base and modified for Remoting. // namespace System.Runtime.Remoting.Channels { using System.Threading; using System.Collections; internal class RequestQueue { // configuration params private int _minExternFreeThreads; private int _minLocalFreeThreads; private int _queueLimit; // two queues -- one for local requests, one for external private Queue _localQueue = new Queue(); private Queue _externQueue = new Queue(); // total count private int _count; // work items queued to pick up new work private WaitCallback _workItemCallback; private int _workItemCount; private const int _workItemLimit = 2; // helpers private static bool IsLocal(SocketHandler sh) { return sh.IsLocal(); } private void QueueRequest(SocketHandler sh, bool isLocal) { lock (this) { if (isLocal) _localQueue.Enqueue(sh); else _externQueue.Enqueue(sh); _count++; } } private SocketHandler DequeueRequest(bool localOnly) { Object sh = null; if (_count > 0) { lock (this) { if (_localQueue.Count > 0) { sh = _localQueue.Dequeue(); _count--; } else if (!localOnly && _externQueue.Count > 0) { sh = _externQueue.Dequeue(); _count--; } } } return (SocketHandler)sh; } // ctor internal RequestQueue(int minExternFreeThreads, int minLocalFreeThreads, int queueLimit) { _minExternFreeThreads = minExternFreeThreads; _minLocalFreeThreads = minLocalFreeThreads; _queueLimit = queueLimit; _workItemCallback = new WaitCallback(this.WorkItemCallback); } // method called to process the next request internal void ProcessNextRequest(SocketHandler sh) { sh = GetRequestToExecute(sh); if (sh != null) sh.ProcessRequestNow(); } // ProcessNextRequest // method called when data arrives for incoming requests internal SocketHandler GetRequestToExecute(SocketHandler sh) { int workerThreads, ioThreads; ThreadPool.GetAvailableThreads(out workerThreads, out ioThreads); int freeThreads = (ioThreads > workerThreads) ? workerThreads : ioThreads; // fast path when there are threads available and nothing queued if (freeThreads >= _minExternFreeThreads && _count == 0) return sh; bool isLocal = IsLocal(sh); // fast path when there are threads for local requests available and nothing queued if (isLocal && freeThreads >= _minLocalFreeThreads && _count == 0) return sh; // reject if queue limit exceeded if (_count >= _queueLimit) { sh.RejectRequestNowSinceServerIsBusy(); return null; } // can't execute the current request on the current thread -- need to queue QueueRequest(sh, isLocal); // maybe can execute a request previously queued if (freeThreads >= _minExternFreeThreads) { sh = DequeueRequest(false); // enough threads to process even external requests } else if (freeThreads >= _minLocalFreeThreads) { sh = DequeueRequest(true); // enough threads to process only local requests } else{ sh = null; } if (sh == null){ // not enough threads -> do nothing on this thread ScheduleMoreWorkIfNeeded(); // try to schedule to worker thread } return sh; } // method called from SocketHandler at the end of request internal void ScheduleMoreWorkIfNeeded() { // is queue empty? if (_count == 0) return; // already scheduled enough work items if (_workItemCount >= _workItemLimit) return; // queue the work item Interlocked.Increment(ref _workItemCount); ThreadPool.UnsafeQueueUserWorkItem(_workItemCallback, null); } // method called to pick up more work private void WorkItemCallback(Object state) { Interlocked.Decrement(ref _workItemCount); // is queue empty? if (_count == 0) return; int workerThreads, ioThreads; ThreadPool.GetAvailableThreads(out workerThreads, out ioThreads); bool bHandledRequest = false; // service another request if enough worker threads are available if (workerThreads >= _minLocalFreeThreads) { // pick up request from the queue SocketHandler sh = DequeueRequest(workerThreads < _minExternFreeThreads); if (sh != null) { sh.ProcessRequestNow(); bHandledRequest = true; } } if (!bHandledRequest){ Thread.Sleep(250); ScheduleMoreWorkIfNeeded(); } } } } // File provided for Reference Use Only by Microsoft Corporation (c) 2007. // ==++== // // Copyright (c) Microsoft Corporation. All rights reserved. // // ==--== // // Request Queue // queues up the requests to avoid thread pool starvation, // making sure that there are always available threads to process requests //// This code has been taken from the XSP code base and modified for Remoting. // namespace System.Runtime.Remoting.Channels { using System.Threading; using System.Collections; internal class RequestQueue { // configuration params private int _minExternFreeThreads; private int _minLocalFreeThreads; private int _queueLimit; // two queues -- one for local requests, one for external private Queue _localQueue = new Queue(); private Queue _externQueue = new Queue(); // total count private int _count; // work items queued to pick up new work private WaitCallback _workItemCallback; private int _workItemCount; private const int _workItemLimit = 2; // helpers private static bool IsLocal(SocketHandler sh) { return sh.IsLocal(); } private void QueueRequest(SocketHandler sh, bool isLocal) { lock (this) { if (isLocal) _localQueue.Enqueue(sh); else _externQueue.Enqueue(sh); _count++; } } private SocketHandler DequeueRequest(bool localOnly) { Object sh = null; if (_count > 0) { lock (this) { if (_localQueue.Count > 0) { sh = _localQueue.Dequeue(); _count--; } else if (!localOnly && _externQueue.Count > 0) { sh = _externQueue.Dequeue(); _count--; } } } return (SocketHandler)sh; } // ctor internal RequestQueue(int minExternFreeThreads, int minLocalFreeThreads, int queueLimit) { _minExternFreeThreads = minExternFreeThreads; _minLocalFreeThreads = minLocalFreeThreads; _queueLimit = queueLimit; _workItemCallback = new WaitCallback(this.WorkItemCallback); } // method called to process the next request internal void ProcessNextRequest(SocketHandler sh) { sh = GetRequestToExecute(sh); if (sh != null) sh.ProcessRequestNow(); } // ProcessNextRequest // method called when data arrives for incoming requests internal SocketHandler GetRequestToExecute(SocketHandler sh) { int workerThreads, ioThreads; ThreadPool.GetAvailableThreads(out workerThreads, out ioThreads); int freeThreads = (ioThreads > workerThreads) ? workerThreads : ioThreads; // fast path when there are threads available and nothing queued if (freeThreads >= _minExternFreeThreads && _count == 0) return sh; bool isLocal = IsLocal(sh); // fast path when there are threads for local requests available and nothing queued if (isLocal && freeThreads >= _minLocalFreeThreads && _count == 0) return sh; // reject if queue limit exceeded if (_count >= _queueLimit) { sh.RejectRequestNowSinceServerIsBusy(); return null; } // can't execute the current request on the current thread -- need to queue QueueRequest(sh, isLocal); // maybe can execute a request previously queued if (freeThreads >= _minExternFreeThreads) { sh = DequeueRequest(false); // enough threads to process even external requests } else if (freeThreads >= _minLocalFreeThreads) { sh = DequeueRequest(true); // enough threads to process only local requests } else{ sh = null; } if (sh == null){ // not enough threads -> do nothing on this thread ScheduleMoreWorkIfNeeded(); // try to schedule to worker thread } return sh; } // method called from SocketHandler at the end of request internal void ScheduleMoreWorkIfNeeded() { // is queue empty? if (_count == 0) return; // already scheduled enough work items if (_workItemCount >= _workItemLimit) return; // queue the work item Interlocked.Increment(ref _workItemCount); ThreadPool.UnsafeQueueUserWorkItem(_workItemCallback, null); } // method called to pick up more work private void WorkItemCallback(Object state) { Interlocked.Decrement(ref _workItemCount); // is queue empty? if (_count == 0) return; int workerThreads, ioThreads; ThreadPool.GetAvailableThreads(out workerThreads, out ioThreads); bool bHandledRequest = false; // service another request if enough worker threads are available if (workerThreads >= _minLocalFreeThreads) { // pick up request from the queue SocketHandler sh = DequeueRequest(workerThreads < _minExternFreeThreads); if (sh != null) { sh.ProcessRequestNow(); bHandledRequest = true; } } if (!bHandledRequest){ Thread.Sleep(250); ScheduleMoreWorkIfNeeded(); } } } } // 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
- TextEndOfParagraph.cs
- PrincipalPermission.cs
- WsdlImporterElement.cs
- Point4DValueSerializer.cs
- FolderLevelBuildProviderAppliesToAttribute.cs
- ExtractorMetadata.cs
- WindowsEditBox.cs
- SqlDataSourceEnumerator.cs
- EventProvider.cs
- ACE.cs
- SystemIPv4InterfaceProperties.cs
- WindowsListViewItemStartMenu.cs
- securestring.cs
- CodeArrayIndexerExpression.cs
- IncrementalCompileAnalyzer.cs
- X509Certificate2Collection.cs
- NativeMethods.cs
- BufferedGraphics.cs
- FixedSOMContainer.cs
- TextTreeDeleteContentUndoUnit.cs
- CollectionDataContract.cs
- CombinedTcpChannel.cs
- NameValueCollection.cs
- CompressionTransform.cs
- prefixendpointaddressmessagefiltertable.cs
- ApplicationFileCodeDomTreeGenerator.cs
- ObjectIDGenerator.cs
- TextEditorDragDrop.cs
- SymmetricKeyWrap.cs
- SchemaNotation.cs
- SchemaHelper.cs
- FlowSwitchLink.cs
- ChainOfDependencies.cs
- XmlSchemaSimpleContent.cs
- SliderAutomationPeer.cs
- ExclusiveTcpListener.cs
- ListBase.cs
- FindSimilarActivitiesVerb.cs
- XmlStreamStore.cs
- SubstitutionResponseElement.cs
- WebConfigurationHost.cs
- UnionExpr.cs
- ByteAnimationBase.cs
- TdsParserSessionPool.cs
- ToolStripItemRenderEventArgs.cs
- Baml2006Reader.cs
- FontWeight.cs
- ShaperBuffers.cs
- LambdaCompiler.Binary.cs
- ExtentKey.cs
- ToolStripContentPanel.cs
- LocalizationCodeDomSerializer.cs
- HostedHttpTransportManager.cs
- SizeConverter.cs
- DelegateHelpers.cs
- SmtpReplyReader.cs
- HostSecurityManager.cs
- NativeMethods.cs
- WebUtil.cs
- IProvider.cs
- PointAnimationUsingKeyFrames.cs
- ScrollItemPatternIdentifiers.cs
- EventEntry.cs
- CodeConditionStatement.cs
- PackWebRequestFactory.cs
- TextEditorDragDrop.cs
- assemblycache.cs
- HttpEncoder.cs
- FormsAuthenticationEventArgs.cs
- httpapplicationstate.cs
- DateTimeSerializationSection.cs
- RbTree.cs
- RouteParser.cs
- DropTarget.cs
- Vertex.cs
- ConnectionStringsExpressionEditor.cs
- InternalConfigEventArgs.cs
- nulltextnavigator.cs
- NetMsmqSecurity.cs
- HttpCapabilitiesEvaluator.cs
- Vector3DValueSerializer.cs
- TTSVoice.cs
- Matrix3DValueSerializer.cs
- XmlRawWriter.cs
- StringArrayConverter.cs
- HandlerWithFactory.cs
- OperationInfoBase.cs
- InstanceNormalEvent.cs
- ResourceManagerWrapper.cs
- StandardOleMarshalObject.cs
- VirtualizingStackPanel.cs
- NullReferenceException.cs
- DispatchWrapper.cs
- DecoderReplacementFallback.cs
- InkCollectionBehavior.cs
- EditorBrowsableAttribute.cs
- XmlQueryTypeFactory.cs
- JsonClassDataContract.cs
- UICuesEvent.cs
- SerializationBinder.cs