Mini Heap

minHeap.spec.ts

ts
import { describe, expect, it } from "vitest";
import { peek, push, pop, Heap, Node } from "../src/SchedulerMinHeap";

function createNode(val: number): Node {
  return { sortIndex: val, id: val };
}

describe("test min heap", () => {
  it("empty heap return null", () => {
    const tasks: Heap = [];
    expect(peek(tasks)).toBe(null);
  });

  it("heap length === 1", () => {
    const tasks: Heap = [createNode(1)];
    expect(peek(tasks)).toEqual(createNode(1));
  });

  it("heap length === 1", () => {
    const tasks: Heap = [createNode(1)];
    push(tasks, createNode(2));
    push(tasks, createNode(3));
    expect(peek(tasks)).toEqual(createNode(1));
    push(tasks, createNode(0));
    expect(peek(tasks)).toEqual(createNode(0));
    pop(tasks);
    expect(peek(tasks)).toEqual(createNode(1));
  });
});

ScheduleMiniHeap.ts

ts
export type Heap = Array<Node>;

export type Node = {
  id: number;
  sortIndex: number;
};

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

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

export function pop(heap: Heap): Node | 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(heap: Heap, node: Node, i: number) {
  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(heap: Heap, node: Node, i: number) {
  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;
    }
  }
}

// sort from small to big
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;
}

Implement Scheduler Skeleton

在React 运行的过程中有一些任务要执行，这些任务分别有不同的priority，比如从高优先级到低优先级分别为 立即执行、用户阻塞级别、普通优先级、低优先级、闲置。

ts
export type PriorityLevel = 0 | 1 | 2 | 3 | 4 | 5;

// 任务优先级
// 优先级越高，值越小
export const NoPriority = 0;
export const ImmediatePriority = 1;
export const UserBlockingPriority = 2;
export const NormalPriority = 3;
export const LowPriority = 4;
export const IdlePriority = 5;

调度策略（React如何执行这些任务）

scheduler.js其实是一个独立的包，它和React没有直接的关系，可以理解为用js实现了一个单线程任务调度器

scheduler.js骨架

ts
type Callback = (args: boolean) => Callback | void | undefined;

// 任务存储，最小堆
const taskQueue: Array<Task> = [];

let currentTask: Task | null = null;
// let currentPriorityLevel: PriorityLevel = NormalPriority;
// 获取当前任务优先级
export function getCurrentPriorityLevel(): PriorityLevel {
  return currentTask?.priorityLevel || NormalPriority;
}

// !任务调度器的入口函数
export function scheduleCallback(
  priorityLevel: PriorityLevel,
  callback: Callback,
) {
  // todo
}

// 取消任务
export 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;
}

搭建好了骨架，我们来查一下Scheduler的测试用例

Scheduler Unit Test

scheduler.spec.ts

ts
import {describe, it, test, beforeEach, afterEach, expect, vi} from "vitest";
import {
  scheduleCallback,
  NormalPriority,
  UserBlockingPriority,
  ImmediatePriority,
} from "../index";

describe("任务", () => {
  it("2个相同优先级的任务", () => {
    let eventTasks: Array<string> = [];

    scheduleCallback(NormalPriority, () => {
      eventTasks.push("Task1");

      expect(eventTasks).toEqual(["Task1"]);
    });

    scheduleCallback(NormalPriority, () => {
      eventTasks.push("Task2");
      expect(eventTasks).toEqual(["Task1", "Task2"]);
    });
  });

  it("3个不同优先级的任务", () => {
    let eventTasks: Array<string> = [];

    scheduleCallback(NormalPriority, () => {
      eventTasks.push("Task1");
      expect(eventTasks).toEqual(["Task3", "Task2", "Task1"]);
    });

    scheduleCallback(UserBlockingPriority, () => {
      eventTasks.push("Task2");
      expect(eventTasks).toEqual(["Task3", "Task2"]);
    });

    scheduleCallback(ImmediatePriority, () => {
      eventTasks.push("Task3");
      expect(eventTasks).toEqual(["Task3"]);
    });
  });

  it("4个不同优先级的任务", () => {
    let eventTasks: Array<string> = [];

    scheduleCallback(NormalPriority, () => {
      eventTasks.push("Task1");
      expect(eventTasks).toEqual(["Task3", "Task2", "Task1"]);
    });

    scheduleCallback(UserBlockingPriority, () => {
      eventTasks.push("Task2");
      expect(eventTasks).toEqual(["Task3", "Task2"]);
    });

    scheduleCallback(ImmediatePriority, () => {
      eventTasks.push("Task3");
      expect(eventTasks).toEqual(["Task3"]);
    });

    scheduleCallback(NormalPriority, () => {
      eventTasks.push("Task4");

      expect(eventTasks).toEqual(["Task3", "Task2", "Task1", "Task4"]);
    });
  });
});

Time Slice

`callback`,`task`,`work`

callback 是任务的初始值
task 是 scheduler封装后的task
work 是一个时间切片内的工作单元

ts
export interface Task {
  id: number;
  callback: Callback;
  priorityLevel: PriorityLevel;
  startTime: number;
  expirationTime: number;
  sortIndex: number; // 根据前面三个值计算得到
}

何时交还控制权给主线程

ts
let startTime = -1;
let frameInterval = 5; // frameYieldMs;
function shouldYieldToHost() {
  const timeElapsed = performance.now() - startTime;
  if (timeElapsed < frameInterval) {
    // The main thread has only been blocked for a really short amount of time;
    // smaller than a single frame. Don't yield yet.
    return false;
  }
  return true;
}

workLoop

ts
// This is set while performing work, to prevent re-entrance.
let isPerformingWork = false;

// 有很多task，每个task都有一个callback, 
// callback执行完了就执行下一个task
// 一个work就是一个时间切片内要执行的一些task
// 时间切片要循环
// 返回true，说明还有任务待执行
function workLoop(initialTime: number) {
  let currentTime = initialTime;
  currentTask = peek(taskQueue) as Task;
  while (currentTask !== null) {
    if (
      currentTask.expirationTime > currentTime && shouldYieldToHost())
    ) {
      // 当前任务还没有过期，并且没有剩余时间了
      break;
    }
    const callback = currentTask.callback;
    if (isFn(callback)) {
      currentTask.callback = null;
      currentPriorityLevel = currentTask.priorityLevel;
      const didUserCallbackTimeout = currentTask.expirationTime <= currentTime;

      const continuationCallback = callback(didUserCallbackTimeout);
      if (isFn(continuationCallback)) {
        // 任务没有执行完
        currentTask.callback = continuationCallback;
        return true;
      } else {
        if (currentTask === peek(taskQueue)) {
          pop(taskQueue);
        }
      }
    } else {
      // currentTask不是有效任务
      pop(taskQueue);
    }
    currentTask = peek(taskQueue) as Task;
  }
  // 判断还有没有其他的任务
  return currentTask !== null;
}

任务调度器入口函数

ts
// 在调度任务
let isHostCallbackScheduled = false;
function scheduleCallback(
  priorityLevel: PriorityLevel,
  callback: Callback,
) {
  const startTime = performance.now();

  let timeout: number;

  switch (priorityLevel) {
    case ImmediatePriority:
      // -1
      timeout = IMMEDIATE_PRIORITY_TIMEOUT; 
      break;
    case UserBlockingPriority:
      // 250
      timeout = USER_BLOCKING_PRIORITY_TIMEOUT;
      break;
    case IdlePriority:
      timeout = IDLE_PRIORITY_TIMEOUT;
      break;
    case LowPriority:
      // 10000
      timeout = LOW_PRIORITY_TIMEOUT;
      break;
    case NormalPriority:
    default:
      // 5000
      timeout = NORMAL_PRIORITY_TIMEOUT;
      break;
  }
  const expirationTime = startTime + timeout;

  const newTask = {
    id: taskIdCounter++,
    callback,
    priorityLevel,
    startTime, //任务开始调度的理论时间
    expirationTime, //过期时间
    sortIndex: -1,
  };

  newTask.sortIndex = expirationTime;
  push(taskQueue, newTask);

  if (!isHostCallbackScheduled && !isPerformingWork) {
    isHostCallbackScheduled = true;
    requestHostCallback();
  }
}

function requestHostCallback() {
  // todo
}

requestIdleCallback

如何实现时间切片，浏览器有一个API requestIdleCallback

js
function workLoop(deadline) {
  let shouldYield = false;
  while(nextUnitOfWork && !shouldYield) {
    nextUnitOfWork = performUnitOfWork(nextUnitOfWork);
    shouldYield = deadline.timeRemaining() < 1;
  }
  if(!nextUnitOfWork && wipRoot) {
    commitRoot();
  }
  requestIdleCallback(workloop);
}
requestIdleCallback(workloop);

但是它存在兼容性问题，以及不支持任务优先级

React基于MessageChannel实现了一个宏任务的时间切片

实现时间切片

ts
// ! requestHostCallback 是由 schedulerCallback函数触发
let isMessageLoopRunning = false;
function requestHostCallback() {
  if (!isMessageLoopRunning) {
    isMessageLoopRunning = true;
    port.postMessage(null); // 1. 发送宏任务
  }
}
const channel = new MessageChannel();
const port = channel.port2;

// 2. 接收宏任务
channel.port1.onmessage = performWorkUntilDeadline;

// 3. 执行宏任务
function performWorkUntilDeadline() {
  if(isMessageLoopRunning) {
    const currentTime = performance.now();
    let startTime = currentTime;
    let hasMoreWork= true;
    try {
      hasMoreWork = flushWork(currentTime);
    } finally {
      if (hasMoreWork) {
        schedulePerformWorkUntilDeadline();
      } else {
        isMessageLoopRunning = false;
      }
    }
  }
}

// 4. 真正执行宏任务
function flushWork(initialTime: number) {
  isHostCallbackScheduled = false;
  isPerformingWork = true;

  let previousPriorityLevel = currentPriorityLevel;
  try {
    // 5. 循环执行任务
    return workLoop(hasTimeRemaining, initialTime);
  } finally {
    currentTask = null;
    currentPriorityLevel = previousPriorityLevel;
    isPerformingWork = false;
  }
}

// 执行完workLoop， 下一次宏任务还是由schedulerCallback触发 1-2-3-4-5

TODO react延时任务调度器，源码中有，但实际没有用到

目录