react18.2调度器scheduler源码分析

核心算法 - 最小堆

packages\scheduler\src\SchedulerMinHeap.js

type Heap<T: Node> = Array<T>;
type Node = {
  id: number,
  sortIndex: number,
  ...
};

export function push<T: Node>(heap: Heap<T>, node: T): void {
  const index = heap.length;
  heap.push(node);
  siftUp(heap, node, index);
}

export function peek<T: Node>(heap: Heap<T>): T | null {
  return heap.length === 0 ? null : heap[0];
}

export function pop<T: Node>(heap: Heap<T>): T | null {
  if (heap.length === 0) {
    return null;
  }
  const first = heap[0];
  const last = heap.pop();
  if (last !== first) {
    heap[0] = last;
    siftDown(heap, last, 0);
  }
  return first;
}

function siftUp<T: Node>(heap: Heap<T>, node: T, i: number): void {
  let index = i;
  while (index > 0) {
    const parentIndex = (index - 1) >>> 1;
    const parent = heap[parentIndex];
    if (compare(parent, node) > 0) {
      // The parent is larger. Swap positions.
      heap[parentIndex] = node;
      heap[index] = parent;
      index = parentIndex;
    } else {
      // The parent is smaller. Exit.
      return;
    }
  }
}

function siftDown<T: Node>(heap: Heap<T>, node: T, i: number): void {
  let index = i;
  const length = heap.length;
  const halfLength = length >>> 1;
  while (index < halfLength) {
    const leftIndex = (index + 1) * 2 - 1;
    const left = heap[leftIndex];
    const rightIndex = leftIndex + 1;
    const right = heap[rightIndex];

    // If the left or right node is smaller, swap with the smaller of those.
    if (compare(left, node) < 0) {
      if (rightIndex < length && compare(right, left) < 0) {
        heap[index] = right;
        heap[rightIndex] = node;
        index = rightIndex;
      } else {
        heap[index] = left;
        heap[leftIndex] = node;
        index = leftIndex;
      }
    } else if (rightIndex < length && compare(right, node) < 0) {
      heap[index] = right;
      heap[rightIndex] = node;
      index = rightIndex;
    } else {
      // Neither child is smaller. Exit.
      return;
    }
  }
}

function compare(a: Node, b: Node) {
  // Compare sort index first, then task id.
  const diff = a.sortIndex - b.sortIndex;
  return diff !== 0 ? diff : a.id - b.id;
}

任务调度器

通过调用 scheduleCallback 方法进入调度器

const newCallbackNode = scheduleCallback(
    schedulerPriorityLevel,
    performConcurrentWorkOnRoot.bind(null, root),
);

从下面开始，我们将深入scheduler的源码：

1. 定义全局变量

packages\scheduler\src\forks\Scheduler.js

import {push, pop, peek} from '../SchedulerMinHeap';
import {
  ImmediatePriority,
  UserBlockingPriority,
  NormalPriority,
  LowPriority,
  IdlePriority,
} from '../SchedulerPriorities';

import {
  enableIsInputPending,
  enableIsInputPendingContinuous,
  frameYieldMs,
  continuousYieldMs,
  maxYieldMs,
  userBlockingPriorityTimeout,
  lowPriorityTimeout,
  normalPriorityTimeout,
} from '../SchedulerFeatureFlags';

import {
  markTaskRun,
  markTaskYield,
  markTaskCompleted,
  markTaskCanceled,
  markTaskErrored,
  markSchedulerSuspended,
  markSchedulerUnsuspended,
  markTaskStart,
  stopLoggingProfilingEvents,
  startLoggingProfilingEvents,
} from '../SchedulerProfiling';

export type Callback = boolean => ?Callback;

export opaque type Task = {
  id: number,
  callback: Callback | null,
  priorityLevel: PriorityLevel,
  startTime: number,
  expirationTime: number,
  sortIndex: number,
  isQueued?: boolean,
};

// 获取当前时间，通过performance或者Date
let getCurrentTime: () => number | DOMHighResTimeStamp;
const hasPerformanceNow =
  typeof performance === 'object' && typeof performance.now === 'function';

if (hasPerformanceNow) {
  const localPerformance = performance;
  getCurrentTime = () => localPerformance.now();
} else {
  const localDate = Date;
  const initialTime = localDate.now();
  getCurrentTime = () => localDate.now() - initialTime;
}

// Max 31 bit integer. The max integer size in V8 for 32-bit systems.
// 最大的31位整数。V8中32位系统的最大整数。
// Math.pow(2, 30) - 1
// 0b111111111111111111111111111111
var maxSigned31BitInt = 1073741823;

// Tasks are stored on a min heap
var taskQueue: Array<Task> = []; // 没有延迟的任务
var timerQueue: Array<Task> = []; // 有延迟的任务

// 自增id，标记task唯一性
// Incrementing id counter. Used to maintain insertion order.
var taskIdCounter = 1;

// Pausing the scheduler is useful for debugging.
var isSchedulerPaused = false;

var currentTask = null;
var currentPriorityLevel = NormalPriority;

// 锁
// 是否有work在执行
// This is set while performing work, to prevent re-entrance.
var isPerformingWork = false;

// 主线程是否在调度
var isHostCallbackScheduled = false;

let isMessageLoopRunning = false;

// 是否有任务在倒计时
var isHostTimeoutScheduled = false;

let taskTimeoutID: TimeoutID = (-1: any);

let needsPaint = false;

// Capture local references to native APIs, in case a polyfill overrides them.
const localSetTimeout = typeof setTimeout === 'function' ? setTimeout : null;
const localClearTimeout =
  typeof clearTimeout === 'function' ? clearTimeout : null;
const localSetImmediate =
  typeof setImmediate !== 'undefined' ? setImmediate : null; // IE and Node.js + jsdom

// 判断当前是否有所有类型的输入事件，包括按键、鼠标、滚轮触控等DOM UI事件
const isInputPending =
  typeof navigator !== 'undefined' &&
  // $FlowFixMe[prop-missing]
  navigator.scheduling !== undefined &&
  // $FlowFixMe[incompatible-type]
  navigator.scheduling.isInputPending !== undefined
    ? navigator.scheduling.isInputPending.bind(navigator.scheduling)
    : null;

// 是否应该把控制权交换给主线程
function shouldYieldToHost(): boolean {
  // 当前程序运行时间
  const timeElapsed = getCurrentTime() - startTime;
  // 如果运行时间小于帧间隔时间5ms
  if (timeElapsed < frameInterval) {
    return false;
  }

  // The main thread has been blocked for a non-negligible amount of time. We
  // may want to yield control of the main thread, so the browser can perform
  // high priority tasks. The main ones are painting and user input. If there's
  // a pending paint or a pending input, then we should yield. But if there's
  // neither, then we can yield less often while remaining responsive. We'll
  // eventually yield regardless, since there could be a pending paint that
  // wasn't accompanied by a call to `requestPaint`, or other main thread tasks
  // like network events.
  if (enableIsInputPending) {
    if (needsPaint) {
      // There's a pending paint (signaled by `requestPaint`). Yield now.
      return true;
    }
    if (timeElapsed < continuousInputInterval) {
      // We haven't blocked the thread for that long. Only yield if there's a
      // pending discrete input (e.g. click). It's OK if there's pending
      // continuous input (e.g. mouseover).
      if (isInputPending !== null) {
        return isInputPending();
      }
    } else if (timeElapsed < maxInterval) {
      // Yield if there's either a pending discrete or continuous input.
      if (isInputPending !== null) {
        return isInputPending(continuousOptions);
      }
    } else {
      // We've blocked the thread for a long time. Even if there's no pending
      // input, there may be some other scheduled work that we don't know about,
      // like a network event. Yield now.
      return true;
    }
  }

  // `isInputPending` isn't available. Yield now.
  return true;
}

2. 根据环境设置生成宏任务调度函数

let schedulePerformWorkUntilDeadline;
if (typeof localSetImmediate === 'function') {
  // Node.js and old IE.
  // There's a few reasons for why we prefer setImmediate.
  //
  // Unlike MessageChannel, it doesn't prevent a Node.js process from exiting.
  // (Even though this is a DOM fork of the Scheduler, you could get here
  // with a mix of Node.js 15+, which has a MessageChannel, and jsdom.)
  // https://github.com/facebook/react/issues/20756
  //
  // But also, it runs earlier which is the semantic we want.
  // If other browsers ever implement it, it's better to use it.
  // Although both of these would be inferior to native scheduling.
  schedulePerformWorkUntilDeadline = () => {
    localSetImmediate(performWorkUntilDeadline);
  };
} else if (typeof MessageChannel !== 'undefined') {
  // DOM and Worker environments.
  // We prefer MessageChannel because of the 4ms setTimeout clamping.
  const channel = new MessageChannel();
  const port = channel.port2;
  channel.port1.onmessage = performWorkUntilDeadline;
  schedulePerformWorkUntilDeadline = () => {
    port.postMessage(null);
  };
} else {
  // We should only fallback here in non-browser environments.
  schedulePerformWorkUntilDeadline = () => {
    localSetTimeout(performWorkUntilDeadline, 0);
  };
}

3. 任务调度器的入口函数

// 任务调度器的入口函数：并发模式下调度一个回调函数 【这里传入的callback就是performConcurrentWorkOnRoot】
function unstable_scheduleCallback(
  priorityLevel: PriorityLevel,
  callback: Callback,
  options?: {delay: number},
): Task {
  var currentTime = getCurrentTime(); // 获取当前程序执行时间，performance.now()

  var startTime; // 定义任务开始时间，不是执行时间
  if (typeof options === 'object' && options !== null) {
    var delay = options.delay;
    if (typeof delay === 'number' && delay > 0) {
      // 如果存在延期，则开始时间 = 当前时间 + 延期时间
      startTime = currentTime + delay;
    } else {
      // 否则，开始时间 = 当前时间
      startTime = currentTime;
    }
  } else {
    // 开始时间直接等于currentTime
    startTime = currentTime;
  }

  // 定义超时时间 【根据优先级，设置不同的超时时间】
  var timeout;
  switch (priorityLevel) {
    case ImmediatePriority:
      // Times out immediately，立即超时
      timeout = -1;
      break;
    case UserBlockingPriority:
      // Eventually times out，最终超时
      timeout = userBlockingPriorityTimeout; // 250
      break;
    case IdlePriority:
      // Never times out，永不超时
      timeout = maxSigned31BitInt; // Math.pow(2, 30) - 1， 最大的31位整数。V8中32位系统的最大整数。
      break;
    case LowPriority:
      // Eventually times out，最终超时
      timeout = lowPriorityTimeout; // 10000
      break;
    case NormalPriority:
    default:
      // Eventually times out，最终超时
      timeout = normalPriorityTimeout; // 5000
      break;
  }

  // 过期时间，也是理论上的任务执行时间，值越小，说明优先级越高，需要优先执行
  var expirationTime = startTime + timeout;

  var newTask: Task = {
    id: taskIdCounter++,
    callback,
    priorityLevel,
    startTime,
    expirationTime,
    sortIndex: -1, // 调度执行任务的依据
  };

  // 开始时间 大于当前时间：说明是延期任务，先加入到延时队列timerQueue
  if (startTime > currentTime) {
    // 有delay的任务
    newTask.sortIndex = startTime; // sortIndex是把任务从timerQueue中取出来放到taskQueue中的依据
    push(timerQueue, newTask); // 暂时存到timerQueue，等晚点到了执行时间，再放到taskQueue，再执行任务
    if (peek(taskQueue) === null && newTask === peek(timerQueue)) {
      // All tasks are delayed, and this is the task with the earliest delay.
      // 所有任务都延迟了，而这是延迟时间最短的任务。
      if (isHostTimeoutScheduled) {
        // Cancel an existing timeout.
        // 取消现有的setTimeout
        cancelHostTimeout();
      } else {
        isHostTimeoutScheduled = true;
      }
      // Schedule a timeout.
      // setTimeout
      requestHostTimeout(handleTimeout, startTime - currentTime);
    }
  } else {
    // 没有delay的任务，直接加入任务队列
    newTask.sortIndex = expirationTime;
    push(taskQueue, newTask);
    // Schedule a host callback, if needed. If we're already performing work,
    // wait until the next time we yield.
    // 如果需要的话，调度一个HostCallback。如果我们已经在执行work，就等到下次我们让出控制权的时候。
    // 判断host回调任务是否已经被调度，以及是否正在工作中，只有host回调任务还没有被调度 且 当前并未在工作中；才会开启一个新的host回调任务
    // 【首次加载时，需要调度一个host回调任务】
    if (!isHostCallbackScheduled && !isPerformingWork) {
      isHostCallbackScheduled = true;
      // 设置host回调任务，触发localSetImmediate、MessageChannel或localSetTimeout，生成新的宏任务，在宏任务中执行工作循环workLoop
      requestHostCallback();
    }
  }

  return newTask;
}

4. 调度异步执行，创建新的宏任务

function requestHostCallback() {
  if (!isMessageLoopRunning) {
    isMessageLoopRunning = true;
    // 调度异步执行，创建新的宏任务
    schedulePerformWorkUntilDeadline();
  }
}

5. 异步执行宏任务

// 在有效时间内执行工作
const performWorkUntilDeadline = () => {
  if (isMessageLoopRunning) {
    const currentTime = getCurrentTime();
    // Keep track of the start time so we can measure how long the main thread
    // has been blocked.
    // 记录了一个work的起始时间，其实就是一个时间切片的起始时间，是个时间戳
    startTime = currentTime;

    // If a scheduler task throws, exit the current browser task so the
    // error can be observed.
    //
    // Intentionally not using a try-catch, since that makes some debugging
    // techniques harder. Instead, if `flushWork` errors, then `hasMoreWork` will
    // remain true, and we'll continue the work loop.
    let hasMoreWork = true;
    try {
      // 根据返回判断是否还有工作
      hasMoreWork = flushWork(currentTime);
    } finally {
      if (hasMoreWork) {
        // 如果还有任务,则又触发调度宏任务事件，生成新的宏任务，即在下一个event loop中继续执行任务
        schedulePerformWorkUntilDeadline();
      } else {
        isMessageLoopRunning = false;
      }
    }
  }
  // Yielding to the browser will give it a chance to paint, so we can
  // reset this.
  needsPaint = false;
};

function flushWork(initialTime: number) {
  // We'll need a host callback the next time work is scheduled.
  isHostCallbackScheduled = false;
  if (isHostTimeoutScheduled) {
    // We scheduled a timeout but it's no longer needed. Cancel it.
    isHostTimeoutScheduled = false;
    cancelHostTimeout();
  }

  isPerformingWork = true;
  const previousPriorityLevel = currentPriorityLevel;
  try {
    return workLoop(initialTime);
  } finally {
    currentTask = null;
    currentPriorityLevel = previousPriorityLevel;
    isPerformingWork = false;
  }
}

// 有很多task，每个task都有一个callback，callback执行完了，就执行下一个task
// 一个work就是一个时间切片内执行的一些task
// 时间切片要循环，就是work要循环
// 返回为true，表示还有任务没有执行完，需要继续执行
function workLoop(initialTime: number) {
  let currentTime = initialTime;
  // 如果timerQueue中有*有效任务*到达执行时间，就放到taskQueue中
  advanceTimers(currentTime);
  // 从任务队列中取出队列第一个任务【注意：taskQueue中是按任务的到期时间expirationTime排序的，越小越先执行】
  currentTask = peek(taskQueue);
  // 循环从taskQueue中取出任务
  while (currentTask !== null) {
    /**
     * 【重点判断】
     * 1，如果当前任务到期时间 大于 当前时间，说明任务还未过期
     * 2，shouldYieldToHost为true应该暂停
     * 总结：如果同时满足这两个条件，即任务还没过期，但是没有剩余可执行时间了，就应该跳出本次工作循环，
     * 让出主线程，交给渲染流水线，等待下一个宏任务执行task
     */
    if (currentTask.expirationTime > currentTime && shouldYieldToHost()) {
      break;
    }
    // 执行任务
    const callback = currentTask.callback;
    if (typeof callback === 'function') {
      // 有效的任务
      currentTask.callback = null;
      currentPriorityLevel = currentTask.priorityLevel;
      // 是否属于过期的任务，可能存在还没过期的任务。
      const didUserCallbackTimeout = currentTask.expirationTime <= currentTime;
      // 执行任务
      const continuationCallback = callback(didUserCallbackTimeout);
      currentTime = getCurrentTime();
      if (typeof continuationCallback === 'function') {
        // 如果执行完后又返回了 function，赋值给当前任务的callback
        currentTask.callback = continuationCallback;
        advanceTimers(currentTime);
        return true;
      } else {
        // 否则的话，将当前任务移除。中断在这个位置发生，高优先任务会把低优先任务的callback置空。
        if (currentTask === peek(taskQueue)) {
          pop(taskQueue);
        }
        advanceTimers(currentTime);
      }
    } else {
      pop(taskQueue);
    }
    currentTask = peek(taskQueue);
  }
  // Return whether there's additional work
  if (currentTask !== null) {
    // 返回是否还有任务
    return true;
  } else {
    // 说明 currentTask 执行完了
    const firstTimer = peek(timerQueue);
    if (firstTimer !== null) {
      // 处理 timerQueue
      requestHostTimeout(handleTimeout, firstTimer.startTime - currentTime);
    }
    return false;
  }
}

// Scheduler periodically yields in case there is other work on the main
// thread, like user events. By default, it yields multiple times per frame.
// It does not attempt to align with frame boundaries, since most tasks don't
// need to be frame aligned; for those that do, use requestAnimationFrame.
let frameInterval = frameYieldMs;
const continuousInputInterval = continuousYieldMs;
const maxInterval = maxYieldMs;
let startTime = -1;

const continuousOptions = {includeContinuous: enableIsInputPendingContinuous};

function advanceTimers(currentTime: number) {
  // Check for tasks that are no longer delayed and add them to the queue.
  let timer = peek(timerQueue);
  while (timer !== null) {
    if (timer.callback === null) {
      // 无效的任务
      // Timer was cancelled.
      pop(timerQueue);
    } else if (timer.startTime <= currentTime) {
      // 有效的任务
      // 任务已到达开始时间，转入taskQueue中
      // Timer fired. Transfer to the task queue.
      pop(timerQueue);
      timer.sortIndex = timer.expirationTime;
      push(taskQueue, timer);
    } else {
      // Remaining timers are pending.
      return;
    }
    timer = peek(timerQueue);
  }
}

function handleTimeout(currentTime: number) {
  isHostTimeoutScheduled = false;
  // 把延迟任务从timerQueue中推入taskQueue
  advanceTimers(currentTime);

  if (!isHostCallbackScheduled) {
    if (peek(taskQueue) !== null) {
      isHostCallbackScheduled = true;
      requestHostCallback();
    } else {
      const firstTimer = peek(timerQueue);
      if (firstTimer !== null) {
        requestHostTimeout(handleTimeout, firstTimer.startTime - currentTime);
      }
    }
  }
}

function cancelCallback(task: Task) {
  // Null out the callback to indicate the task has been canceled. (Can't
  // remove from the queue because you can't remove arbitrary nodes from an
  // array based heap, only the first one.)
  task.callback = null;
}

function getCurrentPriorityLevel(): PriorityLevel {
  return currentPriorityLevel;
}

function requestHostTimeout(
  callback: (currentTime: number) => void,
  ms: number,
) {
  taskTimeoutID = localSetTimeout(() => {
    callback(getCurrentTime());
  }, ms);
}

function cancelHostTimeout() {
  localClearTimeout(taskTimeoutID);
  taskTimeoutID = ((-1: any): TimeoutID);
}

export {
  ImmediatePriority,
  UserBlockingPriority,
  NormalPriority,
  IdlePriority,
  LowPriority,
  scheduleCallback,
  cancelCallback,
  getCurrentPriorityLevel,
  shouldYieldToHost as unstable_shouldYield,
  getCurrentTime,
};

核心流程图

1. 使用了 Scheduler 任务调度的流程图（Concurrent模式）

2. 没有使用 Scheduler 任务调度的流程图（legacy模式）

总结

一个比较泛的流程示例：

• 在 React 中宏观来看，针对浏览器、Scheduler 、Reconciler 其实是有3层 Loop。浏览器级别的 eventLoop，Scheduler 级别的 workLoop，Reconciler 级别 workLoopConcurrent 。
  • 浏览器的 eventLoop 与 Scheduler 的关系
    • 每次 eventLoop 会执行宏任务的队列的宏任务，而 React 中的 Scheduler 就是用宏任务 setImmediate等 触发的。
    • 当 eventLoop 开始执行跟 Scheduler 有关的宏任务时，Scheduler 会启动一次 workLoop，就是在遍历执行 Scheduler 中已存在的 taskQueue 队列的每个 task。
  • Scheduler 与 Reconciler 的关系
    • Scheduler中的 workLoop 中每执行一次 task，是通过调用 Reconciler 中的 performConcurrentWorkOnRoot 方法，即每一个 task 可以理解为是一个 performConcurrentWorkOnRoot 方法的调用。
    • performConcurrentWorkOnRoot 方法每次调用，其本质是在执行 workLoopConcurrent 方法，这个方法是在循环 performUnitOfWork 这个构建 Fiber 树中每个 Fiber 的方法。

因此可以梳理出来，3个大循环，从最开始的 eventLoop 的单个宏任务执行，会逐步触发 Scheduler 和 Reconciler 的任务循环执行。

• 任务的中断与恢复，实现中断与恢复的逻辑分了2个部分，第一个是 Scheduler 中正在执行的 workLoop 的任务中断，第二个是 Reconciler 中正在执行的 workLoopConcurrent 的任务中断
  • Reconciler 中的任务中断与恢复：在 workLoopConcurrent 的 while 循环中，通过 shouldYield() 方法来判断当前构建 fiber 树的执行过程是否超时，如果超时，则中断当前的 while 循环。由于每次 while 执行的 fiber 构建方法，即 performUnitOfWork 是按照每个 fiberNode 来遍历的，也就是说每完成一次 fiberNode 的 beginWork + completeWork 树的构建过程，会设置下一次 nextNode 的值 ，可以理解为中断时已经保留了下一次要构建的 fiberNode 指针，以至于不会下一次不知道从哪里继续。
  • Scheduler 中的任务中断与恢复：当执行任务时间超时后，如果 Reconciler 中的 performConcurrentWorkOnRoot 方法没有执行完成，会返回其自身。在 Scheduler 中，发现当前任务还有下一个任务没有执行完，则不会将当前任务从 taskQueue 中取出，同时会把 reconciler 中返回的待执行的回调函数继续赋值给当前任务，于是下一次继续启动 Scheduler 的任务时，也就连接上了。同时退出这次中断的任务前，会通过 messageChannel 向 eventLoop 的宏任务队列放入一个新的宏任务。
  • 所以任务的恢复，其实就是从下一次 eventLoop 开始执行 Scheduler 相关的宏任务，而执行的宏任务也是 Reconciler 中断前赋值的 fiberNode，也就实现了整体的任务恢复。