[Day 13] .Net Task 底层(6)

前言

今天我们来看看, 当创建一个 task 物件, 设定完要做的行为已及排程方式後, 会调用两个方法, 他们分别代表甚麽意思。

前情提要

以下方法在执行Task.Run打包新task 调用

internal static Task InternalStartNew(
            Task creatingTask, Delegate action, object state, CancellationToken cancellationToken, TaskScheduler scheduler,
            TaskCreationOptions options, InternalTaskOptions internalOptions, ref StackCrawlMark stackMark)
{
    // Validate arguments.
    if (scheduler == null)
    {
        throw new ArgumentNullException("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.PossiblyCaptureContext(ref stackMark);

    t.ScheduleAndStart(false);
    return t;
}

我们要来看看这两个方法

t.PossiblyCaptureContext(ref stackMark);
t.ScheduleAndStart(false);

PossiblyCaptureContext

internal void PossiblyCaptureContext(ref StackCrawlMark stackMark)
{
    Contract.Assert(m_contingentProperties == null || m_contingentProperties.m_capturedContext == null,
        "Captured an ExecutionContext when one was already captured.");

    // In the legacy .NET 3.5 build, we don't have the optimized overload of Capture()
    // available, so we call the parameterless overload.
#if PFX_LEGACY_3_5
            CapturedContext = ExecutionContext.Capture();
#else
    CapturedContext = ExecutionContext.Capture(
        ref stackMark,
        ExecutionContext.CaptureOptions.IgnoreSyncCtx | ExecutionContext.CaptureOptions.OptimizeDefaultCase);
#endif
}

可以发现重点是调用 ExecutionContext.Capture其底层是利用 Thread 中的 method 获取当前所在 thread 的内容并且回传。

所以变数 CapturedContext , 经过这个方法後里面存了 当前 thread 里面的资料, 待用。

换句话说, task 存下了当初创建他的 thread 的执行内容。

ScheduleAndStart

internal void ScheduleAndStart(bool needsProtection)
{
    Contract.Assert(m_taskScheduler != null, "expected a task scheduler to have been selected");
    Contract.Assert((m_stateFlags & TASK_STATE_STARTED) == 0, "task has already started");

    // 设定任务开始的标记
    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: " + ((Delegate)m_action).Method.Name, 0);
    }

    try
    {
        // 把当前任务送入 Scheduler 的队列中
        m_taskScheduler.InternalQueueTask(this);
    }
    catch (ThreadAbortException tae)
    {
        AddException(tae);
        FinishThreadAbortedTask(true, 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()
        // or from the self replicating logic, because in both cases 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()
            Contract.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;
    }
}

可以发现重点在 m_taskScheduler.InternalQueueTask(this);

其把当前任务送入 Scheduler 的队列中, 而我们之前说过, 这个 Scheduler 其实就是 TP

所以我们继续往下看

internal void InternalQueueTask(Task task)
{
    Contract.Requires(task != null);

    task.FireTaskScheduledIfNeeded(this);

    this.QueueTask(task);
}

重点在 QueueTask

他其实是一个介面, 担当了所有类型 Scheduler 推入任务的入口

而因为我们知道 Scheduler 是 TP , 所以直接看 TP 复写的 QueueTask

protected internal override void QueueTask(Task task)
{
    if ((task.Options & TaskCreationOptions.LongRunning) != 0)
    {
        // Run LongRunning tasks on their own dedicated thread.
        Thread thread = new Thread(s_longRunningThreadWork);
        thread.IsBackground = true; // Keep this thread from blocking process shutdown
        thread.Start(task);
    }
    else
    {
        // Normal handling for non-LongRunning tasks.
        bool forceToGlobalQueue = ((task.Options & TaskCreationOptions.PreferFairness) != 0);
        ThreadPool.UnsafeQueueCustomWorkItem(task, forceToGlobalQueue);
    }
}

这里可以发现两种 case ,

1. longRunning , 指工程比较浩大的任务, 会调用 Thread 的方法, 创建一条新 thread 来运行
2. Normal , 可以看到之前聊过的 UnsafeQueueCustomWorkItem 在处理连续任务区有遇到, 可以把任务放入 TP 来执行。

此外在ScheduleAndStart可以看到在例外处理有

• FinishThreadAbortedTask(true, false);
• Finish

两种 method , 他们是用来表示主要任务完成, 会触发前面提到的连续任务区。

ScheduleAndStart 要求同步运行时

实际上, 在实作场合, 创建一个 Task 但却希望他完成後别的 Task 才继续走的的情况非常常见。

其底层就是在 ScheduleAndStart 中调用 Scheduler 时不是使用预设的 TP 作为 Scheduler 而是使用 SynchronizationContextTaskScheduler类别作为 Scheduler , 以下查看其复写的InternalQueueTask。

protected internal override void QueueTask(Task task)
{
    m_synchronizationContext.Post(s_postCallback, (object)task);
}

public virtual void Post(SendOrPostCallback d, Object state)
{
    ThreadPool.QueueUserWorkItem(new WaitCallback(d), state);
}

可以看见其一样是推入 TP 来执行, 但是调用了 WaitCallback 作为参数, 该物件我往下看过, 来自外部程序码, 推测功能是与同样来自外部的 TP 底层 Scheduler 沟通, 使外部程序码操作的 workerThread 用同步方式运行这个 TP 中的任务。

明天进度

到此, .Net Task 算是看得差不多了, 我明天会进行这几天文章的整理, 算是做一个懒人包给大家。

明天见 !