smartthreadpool.cs

线程池实例,1.1版本,用于代替.net自带线程池

		/// <summary>
		/// Starts new threads
		/// </summary>
		/// <param name="threadsCount">The number of threads to start</param>
		private void StartThreads(int threadsCount)
		{
			if (_stpStartInfo.StartSuspended)
			{
				return;
			}

			lock(_workerThreads.SyncRoot)
			{
				// Don't start threads on shut down
				if (_shutdown)
				{
					return;
				}

				for(int i = 0; i < threadsCount; ++i)
				{
					// Don't create more threads then the upper limit
					if (_workerThreads.Count >= _stpStartInfo.MaxWorkerThreads)
					{
						return;
					}

					// Create a new thread
					Thread workerThread = new Thread(new ThreadStart(ProcessQueuedItems));

					// Configure the new thread and start it
					workerThread.Name = "STP " + Name + " Thread #" + _threadCounter;
					workerThread.IsBackground = true;
					workerThread.Priority = _stpStartInfo.ThreadPriority;
					workerThread.Start();
					++_threadCounter;

					// Add the new thread to the hashtable and update its creation
					// time.
					_workerThreads[workerThread] = DateTime.Now;
					_pcs.SampleThreads(_workerThreads.Count, _inUseWorkerThreads);
				}
			}
		}

		/// <summary>
		/// A worker thread method that processes work items from the work items queue.
		/// </summary>
		private void ProcessQueuedItems()
		{
			// Initialize the _smartThreadPool variable
			_smartThreadPool = this;

			try
			{
				bool bInUseWorkerThreadsWasIncremented = false;

				// Process until shutdown.
				while(!_shutdown)
				{
					// Update the last time this thread was seen alive.
					// It's good for debugging.
					_workerThreads[Thread.CurrentThread] = DateTime.Now;

					// Wait for a work item, shutdown, or timeout
					WorkItem workItem = Dequeue();

					// Update the last time this thread was seen alive.
					// It's good for debugging.
					_workerThreads[Thread.CurrentThread] = DateTime.Now;

					// On timeout or shut down.
					if (null == workItem)
					{
						// Double lock for quit.
						if (_workerThreads.Count > _stpStartInfo.MinWorkerThreads)
						{
							lock(_workerThreads.SyncRoot)
							{
								if (_workerThreads.Count > _stpStartInfo.MinWorkerThreads)
								{
									// Inform that the thread is quiting and then quit.
									// This method must be called within this lock or else
									// more threads will quit and the thread pool will go
									// below the lower limit.
									InformCompleted();
									break;
								}
							}
						}
					}

					// If we didn't quit then skip to the next iteration.
					if (null == workItem)
					{
						continue;
					}

					try 
					{
						// Initialize the value to false
						bInUseWorkerThreadsWasIncremented = false;

						// Change the state of the work item to 'in progress' if possible.
						// We do it here so if the work item has been canceled we won't 
						// increment the _inUseWorkerThreads.
						// The cancel mechanism doesn't delete items from the queue,  
						// it marks the work item as canceled, and when the work item
						// is dequeued, we just skip it.
						// If the post execute of work item is set to always or to
						// call when the work item is canceled then the StartingWorkItem()
						// will return true, so the post execute can run.
						if (!workItem.StartingWorkItem())
						{
							continue;
						}

						// Execute the callback.  Make sure to accurately
						// record how many callbacks are currently executing.
						int inUseWorkerThreads = Interlocked.Increment(ref _inUseWorkerThreads);
						_pcs.SampleThreads(_workerThreads.Count, inUseWorkerThreads);

						// Mark that the _inUseWorkerThreads incremented, so in the finally{}
						// statement we will decrement it correctly.
						bInUseWorkerThreadsWasIncremented = true;

						// Set the _currentWorkItem to the current work item
						_currentWorkItem = workItem;

						ExecuteWorkItem(workItem);
					}
					catch(Exception ex)
					{
                        ex.GetHashCode();
						// Do nothing
					}
					finally
					{
						if (null != workItem)
						{
							workItem.DisposeOfState();
						}

						// Set the _currentWorkItem to null, since we 
						// no longer run user's code.
						_currentWorkItem = null;

						// Decrement the _inUseWorkerThreads only if we had 
						// incremented it. Note the cancelled work items don't
						// increment _inUseWorkerThreads.
						if (bInUseWorkerThreadsWasIncremented)
						{
							int inUseWorkerThreads = Interlocked.Decrement(ref _inUseWorkerThreads);
							_pcs.SampleThreads(_workerThreads.Count, inUseWorkerThreads);
						}

						// Notify that the work item has been completed.
						// WorkItemsGroup may enqueue their next work item.
						workItem.FireWorkItemCompleted();

						// Decrement the number of work items here so the idle 
						// ManualResetEvent won't fluctuate.
						DecrementWorkItemsCount();
					}
				}
			} 
			catch(ThreadAbortException tae)
			{
                tae.GetHashCode();
				// Handle the abort exception gracfully.
				Thread.ResetAbort();
			}
			catch(Exception e)
			{
				Debug.Assert(null != e);
			}
			finally
			{
				InformCompleted();
			}
		}

		private void ExecuteWorkItem(WorkItem workItem)
		{
			_pcs.SampleWorkItemsWaitTime(workItem.WaitingTime);
			try
			{
				workItem.Execute();
			}
			catch
			{
				throw;
			}
			finally
			{
				_pcs.SampleWorkItemsProcessTime(workItem.ProcessTime);
			}
		}


		#endregion

		#region Public Methods

		/// <summary>
		/// Queue a work item
		/// </summary>
		/// <param name="callback">A callback to execute</param>
		/// <returns>Returns a work item result</returns>
		public IWorkItemResult QueueWorkItem(WorkItemCallback callback)
		{
			ValidateNotDisposed();
			ValidateCallback(callback);
			WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _stpStartInfo, callback);
			Enqueue(workItem);
			return workItem.GetWorkItemResult();
		}

		/// <summary>
		/// Queue a work item
		/// </summary>
		/// <param name="callback">A callback to execute</param>
		/// <param name="workItemPriority">The priority of the work item</param>
		/// <returns>Returns a work item result</returns>
		public IWorkItemResult QueueWorkItem(WorkItemCallback callback, WorkItemPriority workItemPriority)
		{
			ValidateNotDisposed();
			ValidateCallback(callback);
			WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _stpStartInfo, callback, workItemPriority);
			Enqueue(workItem);
			return workItem.GetWorkItemResult();
		}

		/// <summary>
		/// Queue a work item
		/// </summary>
		/// <param name="workItemInfo">Work item info</param>
		/// <param name="callback">A callback to execute</param>
		/// <returns>Returns a work item result</returns>
		public IWorkItemResult QueueWorkItem(WorkItemInfo workItemInfo, WorkItemCallback callback)
		{
			ValidateNotDisposed();
			ValidateCallback(callback);
			WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _stpStartInfo, workItemInfo, callback);
			Enqueue(workItem);
			return workItem.GetWorkItemResult();
		}

		/// <summary>
		/// Queue a work item
		/// </summary>
		/// <param name="callback">A callback to execute</param>
		/// <param name="state">
		/// The context object of the work item. Used for passing arguments to the work item. 
		/// </param>
		/// <returns>Returns a work item result</returns>
		public IWorkItemResult QueueWorkItem(WorkItemCallback callback, object state)
		{
			ValidateNotDisposed();
			ValidateCallback(callback);
			WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _stpStartInfo, callback, state);
			Enqueue(workItem);
			return workItem.GetWorkItemResult();
		}

		/// <summary>
		/// Queue a work item
		/// </summary>
		/// <param name="callback">A callback to execute</param>
		/// <param name="state">
		/// The context object of the work item. Used for passing arguments to the work item. 
		/// </param>
		/// <param name="workItemPriority">The work item priority</param>
		/// <returns>Returns a work item result</returns>
		public IWorkItemResult QueueWorkItem(WorkItemCallback callback, object state, WorkItemPriority workItemPriority)
		{
			ValidateNotDisposed();
			ValidateCallback(callback);
			WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _stpStartInfo, callback, state, workItemPriority);
			Enqueue(workItem);
			return workItem.GetWorkItemResult();
		}

		/// <summary>
		/// Queue a work item
		/// </summary>
		/// <param name="workItemInfo">Work item information</param>
		/// <param name="callback">A callback to execute</param>
		/// <param name="state">
		/// The context object of the work item. Used for passing arguments to the work item. 
		/// </param>
		/// <returns>Returns a work item result</returns>
		public IWorkItemResult QueueWorkItem(WorkItemInfo workItemInfo, WorkItemCallback callback, object state)
		{
			ValidateNotDisposed();
			ValidateCallback(callback);
			WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _stpStartInfo, workItemInfo, callback, state);
			Enqueue(workItem);
			return workItem.GetWorkItemResult();
		}

		/// <summary>
		/// Queue a work item
		/// </summary>
		/// <param name="callback">A callback to execute</param>
		/// <param name="state">
		/// The context object of the work item. Used for passing arguments to the work item. 
		/// </param>
		/// <param name="postExecuteWorkItemCallback">
		/// A delegate to call after the callback completion
		/// </param>
		/// <returns>Returns a work item result</returns>
		public IWorkItemResult QueueWorkItem(
			WorkItemCallback callback, 
			object state,
			PostExecuteWorkItemCallback postExecuteWorkItemCallback)
		{
			ValidateNotDisposed();
			ValidateCallback(callback);
			WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _stpStartInfo, callback, state, postExecuteWorkItemCallback);
			Enqueue(workItem);
			return workItem.GetWorkItemResult();
		}

		/// <summary>
		/// Queue a work item
		/// </summary>
		/// <param name="callback">A callback to execute</param>
		/// <param name="state">
		/// The context object of the work item. Used for passing arguments to the work item. 
		/// </param>
		/// <param name="postExecuteWorkItemCallback">
		/// A delegate to call after the callback completion
		/// </param>
		/// <param name="workItemPriority">The work item priority</param>
		/// <returns>Returns a work item result</returns>
		public IWorkItemResult QueueWorkItem(
			WorkItemCallback callback, 
			object state,
			PostExecuteWorkItemCallback postExecuteWorkItemCallback,
			WorkItemPriority workItemPriority)
		{
			ValidateNotDisposed();
			ValidateCallback(callback);
			WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _stpStartInfo, callback, state, postExecuteWorkItemCallback, workItemPriority);
			Enqueue(workItem);
			return workItem.GetWorkItemResult();
		}

		/// <summary>
		/// Queue a work item
		/// </summary>
		/// <param name="callback">A callback to execute</param>
		/// <param name="state">
		/// The context object of the work item. Used for passing arguments to the work item. 
		/// </param>
		/// <param name="postExecuteWorkItemCallback">
		/// A delegate to call after the callback completion
		/// </param>
		/// <param name="callToPostExecute">Indicates on which cases to call to the post execute callback</param>
		/// <returns>Returns a work item result</returns>
		public IWorkItemResult QueueWorkItem(
			WorkItemCallback callback, 
			object state,
			PostExecuteWorkItemCallback postExecuteWorkItemCallback,
			CallToPostExecute callToPostExecute)
		{
			ValidateNotDisposed();
			ValidateCallback(callback);
			WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _stpStartInfo, callback, state, postExecuteWorkItemCallback, callToPostExecute);
			Enqueue(workItem);
			return workItem.GetWorkItemResult();
		}

		/// <summary>
		/// Queue a work item
		/// </summary>
		/// <param name="callback">A callback to execute</param>
		/// <param name="state">
		/// The context object of the work item. Used for passing arguments to the work item. 
		/// </param>
		/// <param name="postExecuteWorkItemCallback">
		/// A delegate to call after the callback completion
		/// </param>
		/// <param name="callToPostExecute">Indicates on which cases to call to the post execute callback</param>
		/// <param name="workItemPriority">The work item priority</param>
		/// <returns>Returns a work item result</returns>
		public IWorkItemResult QueueWorkItem(
			WorkItemCallback callback, 
			object state,
			PostExecuteWorkItemCallback postExecuteWorkItemCallback,
			CallToPostExecute callToPostExecute,
			WorkItemPriority workItemPriority)
		{
			ValidateNotDisposed();
			ValidateCallback(callback);
			WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _stpStartInfo, callback, state, postExecuteWorkItemCallback, callToPostExecute, workItemPriority);
			Enqueue(workItem);
			return workItem.GetWorkItemResult();
		}

		/// <summary>
		/// Wait for the thread pool to be idle
		/// </summary>
		public void WaitForIdle()
		{
			WaitForIdle(Timeout.Infinite);
		}

		/// <summary>
		/// Wait for the thread pool to be idle
		/// </summary>
		public bool WaitForIdle(TimeSpan timeout)
		{
			return WaitForIdle((int)timeout.TotalMilliseconds);
		}

		/// <summary>
		/// Wait for the thread pool to be idle
		/// </summary>
		public bool WaitForIdle(int millisecondsTimeout)
		{
			ValidateWaitForIdle();
			return _isIdleWaitHandle.WaitOne(millisecondsTimeout, false);
		}

		private void ValidateWaitForIdle()
		{
			if(_smartThreadPool == this)
			{
				throw new NotSupportedException(
					"WaitForIdle cannot be called from a thread on its SmartThreadPool, it will cause may cause a deadlock");
			}
		}

		internal void ValidateWorkItemsGroupWaitForIdle(IWorkItemsGroup workItemsGroup)
		{
			ValidateWorkItemsGroupWaitForIdleImpl(workItemsGroup, SmartThreadPool._currentWorkItem);
			if ((null != workItemsGroup) && 
				(null != SmartThreadPool._currentWorkItem) &&
				SmartThreadPool._currentWorkItem.WasQueuedBy(workItemsGroup))
			{
				throw new NotSupportedException("WaitForIdle cannot be called from a thread on its SmartThreadPool, it will cause may cause a deadlock");
			}
		}

		[MethodImpl(MethodImplOptions.NoInlining)]
		private void ValidateWorkItemsGroupWaitForIdleImpl(IWorkItemsGroup workItemsGroup, WorkItem workItem)
		{
			if ((null != workItemsGroup) && 
				(null != workItem) &&
				workItem.WasQueuedBy(workItemsGroup))
			{
				throw new NotSupportedException("WaitForIdle cannot be called from a thread on its SmartThreadPool, it will cause may cause a deadlock");
			}
		}



		/// <summary>
		/// Force the SmartThreadPool to shutdown
		/// </summary>
		public void Shutdown()
		{
			Shutdown(true, 0);
		}

		public void Shutdown(bool forceAbort, TimeSpan timeout)
		{
			Shutdown(forceAbort, (int)timeout.TotalMilliseconds);
		}