线程池ThreadPool及Task调度机制分析

/// <summary>
/// Queues the specified work to run on the ThreadPool and returns a Task handle for that work.
/// </summary>
/// <param name="action">The work to execute asynchronously</param>
/// <returns>A Task that represents the work queued to execute in the ThreadPool.</returns>
/// <exception cref="T:System.ArgumentNullException">
/// The <paramref name="action"/> parameter was null.
/// </exception>
public static Task Run(Action action)
{
    return Task.InternalStartNew(null, action, null, default(CancellationToken), TaskScheduler.Default,
        TaskCreationOptions.DenyChildAttach, InternalTaskOptions.None);
}

// Implicitly converts action to object and handles the meat of the StartNew() logic.
internal static Task InternalStartNew(
    Task creatingTask, Delegate action, object state, CancellationToken cancellationToken, TaskScheduler scheduler,
    TaskCreationOptions options, InternalTaskOptions internalOptions)
{
    // Validate arguments.
    if (scheduler == null)
    {
        ThrowHelper.ThrowArgumentNullException(ExceptionArgument.scheduler);
    }
    Contract.EndContractBlock();

    // Create and schedule the task. This throws an InvalidOperationException if already shut down.
    // Here we add the InternalTaskOptions.QueuedByRuntime to the internalOptions, so that TaskConstructorCore can skip the cancellation token registration
    Task t = new Task(action, state, creatingTask, cancellationToken, options, internalOptions | InternalTaskOptions.QueuedByRuntime, scheduler);

    t.ScheduleAndStart(false);
    return t;
}

/// <summary>
/// Schedules the task for execution.
/// </summary>
/// <param name="needsProtection">If true, TASK_STATE_STARTED bit is turned on in
/// an atomic fashion, making sure that TASK_STATE_CANCELED does not get set
/// underneath us.  If false, TASK_STATE_STARTED bit is OR-ed right in.  This
/// allows us to streamline things a bit for StartNew(), where competing cancellations
/// are not a problem.</param>
internal void ScheduleAndStart(bool needsProtection)
{
    Debug.Assert(m_taskScheduler != null, "expected a task scheduler to have been selected");
    Debug.Assert((m_stateFlags & TASK_STATE_STARTED) == 0, "task has already started");

    // Set the TASK_STATE_STARTED bit
    if (needsProtection)
    {
        if (!MarkStarted())
        {
            // A cancel has snuck in before we could get started.  Quietly exit.
            return;
        }
    }
    else
    {
        m_stateFlags |= TASK_STATE_STARTED;
    }

    if (s_asyncDebuggingEnabled)
    {
        AddToActiveTasks(this);
    }

    if (AsyncCausalityTracer.LoggingOn && (Options & (TaskCreationOptions)InternalTaskOptions.ContinuationTask) == 0)
    {
        //For all other task than TaskContinuations we want to log. TaskContinuations log in their constructor
        AsyncCausalityTracer.TraceOperationCreation(CausalityTraceLevel.Required, this.Id, "Task: " + m_action.Method.Name, 0);
    }


    try
    {
        // Queue to the indicated scheduler.
        m_taskScheduler.InternalQueueTask(this);
    }
    catch (ThreadAbortException tae)
    {
        AddException(tae);
        FinishThreadAbortedTask(delegateRan: false);
    }
    catch (Exception e)
    {
        // The scheduler had a problem queueing this task.  Record the exception, leaving this task in
        // a Faulted state.
        TaskSchedulerException tse = new TaskSchedulerException(e);
        AddException(tse);
        Finish(false);

        // Now we need to mark ourselves as "handled" to avoid crashing the finalizer thread if we are called from StartNew(),
        // because the exception is either propagated outside directly, or added to an enclosing parent. However we won't do this for
        // continuation tasks, because in that case we internally eat the exception and therefore we need to make sure the user does
        // later observe it explicitly or see it on the finalizer.

        if ((Options & (TaskCreationOptions)InternalTaskOptions.ContinuationTask) == 0)
        {
            // m_contingentProperties.m_exceptionsHolder *should* already exist after AddException()
            Debug.Assert(
                (m_contingentProperties != null) &&
                (m_contingentProperties.m_exceptionsHolder != null) &&
                (m_contingentProperties.m_exceptionsHolder.ContainsFaultList),
                    "Task.ScheduleAndStart(): Expected m_contingentProperties.m_exceptionsHolder to exist " +
                    "and to have faults recorded.");

            m_contingentProperties.m_exceptionsHolder.MarkAsHandled(false);
        }
        // re-throw the exception wrapped as a TaskSchedulerException.
        throw tse;
    }
}

protected internal override void QueueTask(Task task)
{
    if ((task.Options & TaskCreationOptions.LongRunning) != 0)
    {
        Thread thread = new Thread(s_longRunningThreadWork);
        thread.IsBackground = true;
        thread.Start(task);
    }
    else
    {
        bool forceGlobal = (task.Options & TaskCreationOptions.PreferFairness) != TaskCreationOptions.None;
        ThreadPool.UnsafeQueueCustomWorkItem(task, forceGlobal);
    }
}

public void Enqueue(IThreadPoolWorkItem callback, bool forceGlobal)
{
    if (loggingEnabled)
        System.Diagnostics.Tracing.FrameworkEventSource.Log.ThreadPoolEnqueueWorkObject(callback);

    ThreadPoolWorkQueueThreadLocals tl = null;
    if (!forceGlobal)
        tl = ThreadPoolWorkQueueThreadLocals.threadLocals;

    if (null != tl)
    {
        tl.workStealingQueue.LocalPush(callback);
    }
    else
    {
        workItems.Enqueue(callback);
    }

    EnsureThreadRequested();
}

internal void EnsureThreadRequested()
{
    //
    // If we have not yet requested #procs threads from the VM, then request a new thread.
    // Note that there is a separate count in the VM which will also be incremented in this case, 
    // which is handled by RequestWorkerThread.
    //
    int count = numOutstandingThreadRequests;
    while (count < ThreadPoolGlobals.processorCount)
    {
        int prev = Interlocked.CompareExchange(ref numOutstandingThreadRequests, count + 1, count);
        if (prev == count)
        {
            ThreadPool.RequestWorkerThread();
            break;
        }
        count = prev;
    }
}

Console.WriteLine("CPU={0}", Environment.ProcessorCount);
for (var i = 0; i < 10; i++)
{
    ThreadPool.QueueUserWorkItem(s =>
    {
        var n = (Int32)s;
        Console.WriteLine("{0:HH:mm:ss.fff} th {1} start", DateTime.Now, n);
        Thread.Sleep(2000);
        Console.WriteLine("{0:HH:mm:ss.fff} th {1} end", DateTime.Now, n);
    }, i);
}

internal static bool Dispatch()
{
    var workQueue = ThreadPoolGlobals.workQueue;
    //
    // The clock is ticking!  We have ThreadPoolGlobals.TP_QUANTUM milliseconds to get some work done, and then
    // we need to return to the VM.
    //
    int quantumStartTime = Environment.TickCount;

    //
    // Update our records to indicate that an outstanding request for a thread has now been fulfilled.
    // From this point on, we are responsible for requesting another thread if we stop working for any
    // reason, and we believe there might still be work in the queue.
    //
    // Note that if this thread is aborted before we get a chance to request another one, the VM will
    // record a thread request on our behalf.  So we don't need to worry about getting aborted right here.
    //
    workQueue.MarkThreadRequestSatisfied();

    // Has the desire for logging changed since the last time we entered?
    workQueue.loggingEnabled = FrameworkEventSource.Log.IsEnabled(EventLevel.Verbose, FrameworkEventSource.Keywords.ThreadPool | FrameworkEventSource.Keywords.ThreadTransfer);

    //
    // Assume that we're going to need another thread if this one returns to the VM.  We'll set this to 
    // false later, but only if we're absolutely certain that the queue is empty.
    //
    bool needAnotherThread = true;
    IThreadPoolWorkItem workItem = null;
    try
    {
        //
        // Set up our thread-local data
        //
        ThreadPoolWorkQueueThreadLocals tl = workQueue.EnsureCurrentThreadHasQueue();

        //
        // Loop until our quantum expires.
        //
        while ((Environment.TickCount - quantumStartTime) < ThreadPoolGlobals.TP_QUANTUM)
        {
            bool missedSteal = false;
            workItem = workQueue.Dequeue(tl, ref missedSteal);

            if (workItem == null)
            {
                //
                // No work.  We're going to return to the VM once we leave this protected region.
                // If we missed a steal, though, there may be more work in the queue.
                // Instead of looping around and trying again, we'll just request another thread.  This way
                // we won't starve other AppDomains while we spin trying to get locks, and hopefully the thread
                // that owns the contended work-stealing queue will pick up its own workitems in the meantime, 
                // which will be more efficient than this thread doing it anyway.
                //
                needAnotherThread = missedSteal;

                // Tell the VM we're returning normally, not because Hill Climbing asked us to return.
                return true;
            }

            if (workQueue.loggingEnabled)
                System.Diagnostics.Tracing.FrameworkEventSource.Log.ThreadPoolDequeueWorkObject(workItem);

            //
            // If we found work, there may be more work.  Ask for another thread so that the other work can be processed
            // in parallel.  Note that this will only ask for a max of #procs threads, so it's safe to call it for every dequeue.
            //
            workQueue.EnsureThreadRequested();

            //
            // Execute the workitem outside of any finally blocks, so that it can be aborted if needed.
            //
            if (ThreadPoolGlobals.enableWorkerTracking)
            {
                bool reportedStatus = false;
                try
                {
                    ThreadPool.ReportThreadStatus(isWorking: true);
                    reportedStatus = true;
                    workItem.ExecuteWorkItem();
                }
                finally
                {
                    if (reportedStatus)
                        ThreadPool.ReportThreadStatus(isWorking: false);
                }
            }
            else
            {
                workItem.ExecuteWorkItem();
            }
            workItem = null;

            // 
            // Notify the VM that we executed this workitem.  This is also our opportunity to ask whether Hill Climbing wants
            // us to return the thread to the pool or not.
            //
            if (!ThreadPool.NotifyWorkItemComplete())
                return false;
        }

        // If we get here, it's because our quantum expired.  Tell the VM we're returning normally.
        return true;
    }
    catch (ThreadAbortException tae)
    {
        //
        // This is here to catch the case where this thread is aborted between the time we exit the finally block in the dispatch
        // loop, and the time we execute the work item.  QueueUserWorkItemCallback uses this to update its accounting of whether
        // it was executed or not (in debug builds only).  Task uses this to communicate the ThreadAbortException to anyone
        // who waits for the task to complete.
        //
        workItem?.MarkAborted(tae);

        //
        // In this case, the VM is going to request another thread on our behalf.  No need to do it twice.
        //
        needAnotherThread = false;
        // throw;  //no need to explicitly rethrow a ThreadAbortException, and doing so causes allocations on amd64.
    }
    finally
    {
        //
        // If we are exiting for any reason other than that the queue is definitely empty, ask for another
        // thread to pick up where we left off.
        //
        if (needAnotherThread)
            workQueue.EnsureThreadRequested();
    }

    // we can never reach this point, but the C# compiler doesn't know that, because it doesn't know the ThreadAbortException will be reraised above.
    Debug.Fail("Should never reach this point");
    return true;
}

public IThreadPoolWorkItem Dequeue(ThreadPoolWorkQueueThreadLocals tl, ref bool missedSteal)
{
    WorkStealingQueue localWsq = tl.workStealingQueue;
    IThreadPoolWorkItem callback;

    if ((callback = localWsq.LocalPop()) == null && // first try the local queue
        !workItems.TryDequeue(out callback)) // then try the global queue
    {
        // finally try to steal from another thread's local queue
        WorkStealingQueue[] queues = WorkStealingQueueList.Queues;
        int c = queues.Length;
        Debug.Assert(c > 0, "There must at least be a queue for this thread.");
        int maxIndex = c - 1;
        int i = tl.random.Next(c);
        while (c > 0)
        {
            i = (i < maxIndex) ? i + 1 : 0;
            WorkStealingQueue otherQueue = queues[i];
            if (otherQueue != localWsq && otherQueue.CanSteal)
            {
                callback = otherQueue.TrySteal(ref missedSteal);
                if (callback != null)
                {
                    break;
                }
            }
            c--;
        }
    }

    return callback;
}