RequestQueue.cs source code in C# .NET

Source code for the .NET framework in C#

                        
Code:

                         / FXUpdate3074 / FXUpdate3074 / 1.1 / DEVDIV / depot / DevDiv / releases / whidbey / QFE / ndp / fx / src / xsp / System / Web / RequestQueue.cs / 2 / 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 
//

namespace System.Web {
    using System.Threading; 
    using System.Collections;
    using System.Web.Util; 
    using System.Web.Hosting; 
    using System.Web.Configuration;
 
    internal class RequestQueue {
        // configuration params
        private int _minExternFreeThreads;
        private int _minLocalFreeThreads; 
        private int _queueLimit;
        private TimeSpan _clientConnectedTime; 
        private bool _iis6; 

        // 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;
        private bool _draining;

        // timer to drain the queue 
        private readonly TimeSpan   _timerPeriod = new TimeSpan(0, 0, 10); // 10 seconds
        private Timer               _timer; 
 

        // helpers 
        private static bool IsLocal(HttpWorkerRequest wr) {
            String remoteAddress = wr.GetRemoteAddress();

            // 
            if (remoteAddress == "127.0.0.1" || remoteAddress == "::1")
                return true; 
 
            // if unknown, assume not local
            if (String.IsNullOrEmpty(remoteAddress)) 
                return false;

            // compare with local address
            if (remoteAddress == wr.GetLocalAddress()) 
                return true;
 
            return false; 
        }
 
        private void QueueRequest(HttpWorkerRequest wr, bool isLocal) {
            lock (this) {
                if (isLocal) {
                    _localQueue.Enqueue(wr); 
                }
                else  { 
                    _externQueue.Enqueue(wr); 
                }
 
                _count++;
            }

            PerfCounters.IncrementGlobalCounter(GlobalPerfCounter.REQUESTS_QUEUED); 
            PerfCounters.IncrementCounter(AppPerfCounter.REQUESTS_IN_APPLICATION_QUEUE);
            if (EtwTrace.IsTraceEnabled(EtwTraceLevel.Information, EtwTraceFlags.Infrastructure)) EtwTrace.Trace(EtwTraceType.ETW_TYPE_REQ_QUEUED, wr); 
        } 

        private HttpWorkerRequest DequeueRequest(bool localOnly) { 
            HttpWorkerRequest wr = null;

            while (_count > 0) {
                lock (this) { 
                    if (_localQueue.Count > 0) {
                        wr = (HttpWorkerRequest)_localQueue.Dequeue(); 
                        _count--; 
                    }
                    else if (!localOnly && _externQueue.Count > 0) { 
                        wr = (HttpWorkerRequest)_externQueue.Dequeue();
                        _count--;
                    }
                } 

                if (wr == null) { 
                    break; 
                }
                else { 
                    PerfCounters.DecrementGlobalCounter(GlobalPerfCounter.REQUESTS_QUEUED);
                    PerfCounters.DecrementCounter(AppPerfCounter.REQUESTS_IN_APPLICATION_QUEUE);
                    if (EtwTrace.IsTraceEnabled(EtwTraceLevel.Information, EtwTraceFlags.Infrastructure)) EtwTrace.Trace(EtwTraceType.ETW_TYPE_REQ_DEQUEUED, wr);
 
                    if (!CheckClientConnected(wr)) {
                        HttpRuntime.RejectRequestNow(wr, true); 
                        wr = null; 

                        PerfCounters.IncrementGlobalCounter(GlobalPerfCounter.REQUESTS_DISCONNECTED); 
                        PerfCounters.IncrementCounter(AppPerfCounter.APP_REQUEST_DISCONNECTED);
                    }
                    else {
                        break; 
                    }
                } 
            } 

            return wr; 
        }

        // This method will
 

        private bool CheckClientConnected(HttpWorkerRequest wr) { 
            if (DateTime.UtcNow - wr.GetStartTime() > _clientConnectedTime) 
                return wr.IsClientConnected();
            else 
                return true;
        }

        // ctor 
        internal RequestQueue(int minExternFreeThreads, int minLocalFreeThreads, int queueLimit, TimeSpan clientConnectedTime) {
            _minExternFreeThreads = minExternFreeThreads; 
            _minLocalFreeThreads = minLocalFreeThreads; 
            _queueLimit = queueLimit;
            _clientConnectedTime = clientConnectedTime; 

            _workItemCallback = new WaitCallback(this.WorkItemCallback);

            _timer = new Timer(new TimerCallback(this.TimerCompletionCallback), null, _timerPeriod, _timerPeriod); 
            _iis6 = HostingEnvironment.IsUnderIIS6Process;
 
            // set the minimum number of requests that must be executing in order to detect a deadlock 
            int maxWorkerThreads, maxIoThreads;
            ThreadPool.GetMaxThreads(out maxWorkerThreads, out maxIoThreads); 
            UnsafeNativeMethods.SetMinRequestsExecutingToDetectDeadlock(maxWorkerThreads - minExternFreeThreads);
        }

        // method called from HttpRuntime for incoming requests 
        internal HttpWorkerRequest GetRequestToExecute(HttpWorkerRequest wr) {
            int workerThreads, ioThreads; 
            ThreadPool.GetAvailableThreads(out workerThreads, out ioThreads); 

            int freeThreads; 
            if (_iis6)
                freeThreads = workerThreads; // ignore IO threads to avoid starvation from Indigo TCP requests
            else
                freeThreads = (ioThreads > workerThreads) ? workerThreads : ioThreads; 

            // fast path when there are threads available and nothing queued 
            if (freeThreads >= _minExternFreeThreads && _count == 0) 
                return wr;
 
            bool isLocal = IsLocal(wr);

            // fast path when there are threads for local requests available and nothing queued
            if (isLocal && freeThreads >= _minLocalFreeThreads && _count == 0) 
                return wr;
 
            // reject if queue limit exceeded 
            if (_count >= _queueLimit) {
                HttpRuntime.RejectRequestNow(wr, false); 
                return null;
            }

            // can't execute the current request on the current thread -- need to queue 
            QueueRequest(wr, isLocal);
 
            // maybe can execute a request previously queued 
            if (freeThreads >= _minExternFreeThreads) {
                wr = DequeueRequest(false); // enough threads to process even external requests 
            }
            else if (freeThreads >= _minLocalFreeThreads) {
                wr = DequeueRequest(true);  // enough threads to process only local requests
            } 
            else {
                wr = null;                  // not enough threads -> do nothing on this thread 
                ScheduleMoreWorkIfNeeded(); // try to schedule to worker thread 
            }
 
            return wr;
        }

        // method called from HttpRuntime at the end of request 
        internal void ScheduleMoreWorkIfNeeded() {
            // too late for more work if draining 
            if (_draining) 
                return;
 
            // is queue empty?
            if (_count == 0)
                return;
 
            // already scheduled enough work items
            if (_workItemCount >= _workItemLimit) 
                return; 

            // enough worker threads? 
            int workerThreads, ioThreads;
            ThreadPool.GetAvailableThreads(out workerThreads, out ioThreads);
            if (workerThreads < _minLocalFreeThreads)
                return; 

            // queue the work item 
            Interlocked.Increment(ref _workItemCount); 
            ThreadPool.QueueUserWorkItem(_workItemCallback);
        } 

        // is empty property
        internal bool IsEmpty {
            get { return (_count == 0); } 
        }
 
        // method called to pick up more work 
        private void WorkItemCallback(Object state) {
            Interlocked.Decrement(ref _workItemCount); 

            // too late for more work if draining
            if (_draining)
                return; 

            // is queue empty? 
            if (_count == 0) 
                return;
 
            int workerThreads, ioThreads;
            ThreadPool.GetAvailableThreads(out workerThreads, out ioThreads);

            // not enough worker threads to do anything 
            if (workerThreads < _minLocalFreeThreads)
                return; 
 
            // pick up request from the queue
            HttpWorkerRequest wr = DequeueRequest(workerThreads < _minExternFreeThreads); 
            if (wr == null)
                return;

            // let another work item through before processing the request 
            ScheduleMoreWorkIfNeeded();
 
            // call the runtime to process request 
            HttpRuntime.ProcessRequestNow(wr);
        } 

        // periodic timer to pick up more work
        private void TimerCompletionCallback(Object state) {
            ScheduleMoreWorkIfNeeded(); 
        }
 
        // reject all requests 
        internal void Drain() {
            // set flag before killing timer to shorten the code path 
            // in the callback after the timer is disposed
            _draining = true;

            // stop the timer 
            if (_timer != null) {
                ((IDisposable)_timer).Dispose(); 
                _timer = null; 
            }
 
            // wait for all work items to finish
            while (_workItemCount > 0)
                Thread.Sleep(100);
 
            // is queue empty?
            if (_count == 0) 
                return; 

            // reject the remaining requests 
            for (;;) {
                HttpWorkerRequest wr = DequeueRequest(false);
                if (wr == null)
                    break; 
                HttpRuntime.RejectRequestNow(wr, false);
            } 
        } 
    }
} 

// File provided for Reference Use Only by Microsoft Corporation (c) 2007.
// Copyright (c) Microsoft Corporation. All rights reserved.
//------------------------------------------------------------------------------ 
// 
//     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 
//

namespace System.Web {
    using System.Threading; 
    using System.Collections;
    using System.Web.Util; 
    using System.Web.Hosting; 
    using System.Web.Configuration;
 
    internal class RequestQueue {
        // configuration params
        private int _minExternFreeThreads;
        private int _minLocalFreeThreads; 
        private int _queueLimit;
        private TimeSpan _clientConnectedTime; 
        private bool _iis6; 

        // 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;
        private bool _draining;

        // timer to drain the queue 
        private readonly TimeSpan   _timerPeriod = new TimeSpan(0, 0, 10); // 10 seconds
        private Timer               _timer; 
 

        // helpers 
        private static bool IsLocal(HttpWorkerRequest wr) {
            String remoteAddress = wr.GetRemoteAddress();

            // 
            if (remoteAddress == "127.0.0.1" || remoteAddress == "::1")
                return true; 
 
            // if unknown, assume not local
            if (String.IsNullOrEmpty(remoteAddress)) 
                return false;

            // compare with local address
            if (remoteAddress == wr.GetLocalAddress()) 
                return true;
 
            return false; 
        }
 
        private void QueueRequest(HttpWorkerRequest wr, bool isLocal) {
            lock (this) {
                if (isLocal) {
                    _localQueue.Enqueue(wr); 
                }
                else  { 
                    _externQueue.Enqueue(wr); 
                }
 
                _count++;
            }

            PerfCounters.IncrementGlobalCounter(GlobalPerfCounter.REQUESTS_QUEUED); 
            PerfCounters.IncrementCounter(AppPerfCounter.REQUESTS_IN_APPLICATION_QUEUE);
            if (EtwTrace.IsTraceEnabled(EtwTraceLevel.Information, EtwTraceFlags.Infrastructure)) EtwTrace.Trace(EtwTraceType.ETW_TYPE_REQ_QUEUED, wr); 
        } 

        private HttpWorkerRequest DequeueRequest(bool localOnly) { 
            HttpWorkerRequest wr = null;

            while (_count > 0) {
                lock (this) { 
                    if (_localQueue.Count > 0) {
                        wr = (HttpWorkerRequest)_localQueue.Dequeue(); 
                        _count--; 
                    }
                    else if (!localOnly && _externQueue.Count > 0) { 
                        wr = (HttpWorkerRequest)_externQueue.Dequeue();
                        _count--;
                    }
                } 

                if (wr == null) { 
                    break; 
                }
                else { 
                    PerfCounters.DecrementGlobalCounter(GlobalPerfCounter.REQUESTS_QUEUED);
                    PerfCounters.DecrementCounter(AppPerfCounter.REQUESTS_IN_APPLICATION_QUEUE);
                    if (EtwTrace.IsTraceEnabled(EtwTraceLevel.Information, EtwTraceFlags.Infrastructure)) EtwTrace.Trace(EtwTraceType.ETW_TYPE_REQ_DEQUEUED, wr);
 
                    if (!CheckClientConnected(wr)) {
                        HttpRuntime.RejectRequestNow(wr, true); 
                        wr = null; 

                        PerfCounters.IncrementGlobalCounter(GlobalPerfCounter.REQUESTS_DISCONNECTED); 
                        PerfCounters.IncrementCounter(AppPerfCounter.APP_REQUEST_DISCONNECTED);
                    }
                    else {
                        break; 
                    }
                } 
            } 

            return wr; 
        }

        // This method will
 

        private bool CheckClientConnected(HttpWorkerRequest wr) { 
            if (DateTime.UtcNow - wr.GetStartTime() > _clientConnectedTime) 
                return wr.IsClientConnected();
            else 
                return true;
        }

        // ctor 
        internal RequestQueue(int minExternFreeThreads, int minLocalFreeThreads, int queueLimit, TimeSpan clientConnectedTime) {
            _minExternFreeThreads = minExternFreeThreads; 
            _minLocalFreeThreads = minLocalFreeThreads; 
            _queueLimit = queueLimit;
            _clientConnectedTime = clientConnectedTime; 

            _workItemCallback = new WaitCallback(this.WorkItemCallback);

            _timer = new Timer(new TimerCallback(this.TimerCompletionCallback), null, _timerPeriod, _timerPeriod); 
            _iis6 = HostingEnvironment.IsUnderIIS6Process;
 
            // set the minimum number of requests that must be executing in order to detect a deadlock 
            int maxWorkerThreads, maxIoThreads;
            ThreadPool.GetMaxThreads(out maxWorkerThreads, out maxIoThreads); 
            UnsafeNativeMethods.SetMinRequestsExecutingToDetectDeadlock(maxWorkerThreads - minExternFreeThreads);
        }

        // method called from HttpRuntime for incoming requests 
        internal HttpWorkerRequest GetRequestToExecute(HttpWorkerRequest wr) {
            int workerThreads, ioThreads; 
            ThreadPool.GetAvailableThreads(out workerThreads, out ioThreads); 

            int freeThreads; 
            if (_iis6)
                freeThreads = workerThreads; // ignore IO threads to avoid starvation from Indigo TCP requests
            else
                freeThreads = (ioThreads > workerThreads) ? workerThreads : ioThreads; 

            // fast path when there are threads available and nothing queued 
            if (freeThreads >= _minExternFreeThreads && _count == 0) 
                return wr;
 
            bool isLocal = IsLocal(wr);

            // fast path when there are threads for local requests available and nothing queued
            if (isLocal && freeThreads >= _minLocalFreeThreads && _count == 0) 
                return wr;
 
            // reject if queue limit exceeded 
            if (_count >= _queueLimit) {
                HttpRuntime.RejectRequestNow(wr, false); 
                return null;
            }

            // can't execute the current request on the current thread -- need to queue 
            QueueRequest(wr, isLocal);
 
            // maybe can execute a request previously queued 
            if (freeThreads >= _minExternFreeThreads) {
                wr = DequeueRequest(false); // enough threads to process even external requests 
            }
            else if (freeThreads >= _minLocalFreeThreads) {
                wr = DequeueRequest(true);  // enough threads to process only local requests
            } 
            else {
                wr = null;                  // not enough threads -> do nothing on this thread 
                ScheduleMoreWorkIfNeeded(); // try to schedule to worker thread 
            }
 
            return wr;
        }

        // method called from HttpRuntime at the end of request 
        internal void ScheduleMoreWorkIfNeeded() {
            // too late for more work if draining 
            if (_draining) 
                return;
 
            // is queue empty?
            if (_count == 0)
                return;
 
            // already scheduled enough work items
            if (_workItemCount >= _workItemLimit) 
                return; 

            // enough worker threads? 
            int workerThreads, ioThreads;
            ThreadPool.GetAvailableThreads(out workerThreads, out ioThreads);
            if (workerThreads < _minLocalFreeThreads)
                return; 

            // queue the work item 
            Interlocked.Increment(ref _workItemCount); 
            ThreadPool.QueueUserWorkItem(_workItemCallback);
        } 

        // is empty property
        internal bool IsEmpty {
            get { return (_count == 0); } 
        }
 
        // method called to pick up more work 
        private void WorkItemCallback(Object state) {
            Interlocked.Decrement(ref _workItemCount); 

            // too late for more work if draining
            if (_draining)
                return; 

            // is queue empty? 
            if (_count == 0) 
                return;
 
            int workerThreads, ioThreads;
            ThreadPool.GetAvailableThreads(out workerThreads, out ioThreads);

            // not enough worker threads to do anything 
            if (workerThreads < _minLocalFreeThreads)
                return; 
 
            // pick up request from the queue
            HttpWorkerRequest wr = DequeueRequest(workerThreads < _minExternFreeThreads); 
            if (wr == null)
                return;

            // let another work item through before processing the request 
            ScheduleMoreWorkIfNeeded();
 
            // call the runtime to process request 
            HttpRuntime.ProcessRequestNow(wr);
        } 

        // periodic timer to pick up more work
        private void TimerCompletionCallback(Object state) {
            ScheduleMoreWorkIfNeeded(); 
        }
 
        // reject all requests 
        internal void Drain() {
            // set flag before killing timer to shorten the code path 
            // in the callback after the timer is disposed
            _draining = true;

            // stop the timer 
            if (_timer != null) {
                ((IDisposable)_timer).Dispose(); 
                _timer = null; 
            }
 
            // wait for all work items to finish
            while (_workItemCount > 0)
                Thread.Sleep(100);
 
            // is queue empty?
            if (_count == 0) 
                return; 

            // reject the remaining requests 
            for (;;) {
                HttpWorkerRequest wr = DequeueRequest(false);
                if (wr == null)
                    break; 
                HttpRuntime.RejectRequestNow(wr, false);
            } 
        } 
    }
} 

// File provided for Reference Use Only by Microsoft Corporation (c) 2007.
// Copyright (c) Microsoft Corporation. All rights reserved.

Link Menu

Network programming in C#, Network Programming in VB.NET, Network Programming in .NET
This book is available now!
Buy at Amazon US or
Buy at Amazon UK