关于yarn:Monorepo-中的任务调度机制

前言

Monorepo 中的一个我的项目称为 project，对 project 进行的具体操作称为工作 task，比方 build 或 test，能够广义地了解为 npm scripts 中注册的操作，Monorepo 管理工具该当有能力调度这些工作。

先看一个 🌰，如上图所示，存在一个依赖关系较为简单的 Monorepo，此时须要执行某个工作，例如 build，如何同时保障工作执行程序以及工作执行效率（假如最大工作并行数为 N）？

接下来就是枯燥乏味的做题过程，咱们先把下面那张我的项目依赖图形象成代码。

问题

interface Project {
  name: string;
  actions: {name: string; fn: () => Promise<void> }[];
  dependencyProjects: Project[];}

const sleep = (s: number): Promise<void> =>
  new Promise((r) => setTimeout(r, s));

// Monorepo 中注册的所有我的项目
const projects: Project[] = [
  "@monorepo/a",
  "@monorepo/b",
  "@monorepo/c",
  "@monorepo/d",
  "@monorepo/x",
  "@monorepo/y",
  "@monorepo/z",
].map((name) => ({
  name,
  actions: [{name: "build", fn: () => sleep(Math.random() * 1000) }],
  dependencyProjects: [],}));

const [A, B, C, D, X, Y, Z] = projects;

A.dependencyProjects = [B];
B.dependencyProjects = [D];
C.dependencyProjects = [D, X, Y];
X.dependencyProjects = [Y, Z];

/**
 * 实现本办法，使得 build 行为依照正确的程序执行，且保障执行效率
 * @param projects 须要执行工作的 project 汇合
 * @param actionName 具体操作名称
 * @param limit 工作最大并行数
 */
function run(projects: Project[], actionName: string, limit: number) {// todo}

run(projects, "build", 12);

解题

很显著，project 之间存在依赖关系，那么工作之间也存在依赖关系，那么能够失去以下论断：

1. 当前任务作为上游工作时，当前任务实现后，须要更新其上游工作的依赖工作，从其内移除当前任务。
2. 当前任务作为上游工作时，只有当前任务的上游工作都被清空（实现）时，当前任务才能够执行。

于是 task 定义如下：

interface Task {// 工作名 `${projectName}:{actionName}`
  name: string;
  // 当前任务依赖的工作，即当前任务的上游工作，当该 dependenciesSet 被清空，阐明当前任务能够被执行
  dependencies: Set<Task>;
  // 依赖当前任务的工作，即当前任务的上游工作，当前任务实现后，须要更新其上游工作的 dependenciesSet（从其内移除当前任务）dependents: Set<Task>;
  // 具体任务执行函数
  fn: () => Promise<void>;}

初始化工作

依据 projects 参数，结构出我的项目对应的工作。

function run(projects: Project[], actionName: string, limit: number) {
  // 工作名与工作的映射
  const tasks = new Map<string, Task>();
  projects.forEach((project) =>
    tasks.set(getTaskName(project, actionName), {name: getTaskName(project, actionName),
      dependencies: new Set(),
      dependents: new Set(),
      fn: project.actions.find((a) => a.name === actionName)?.fn ?? noop,
    })
  );
}

// 获取工作名
function getTaskName(project: Project, actionName: string) {return `${project.name}:${actionName}`;
}

function noop(): Promise<void> {return new Promise((r) => r());
}

补充 dependencies 与 dependents

假如存在 project1，对其进行以下操作：

1. 取到以后我的项目对应的工作 task1
2. 获取当前任务对应的上游工作名 dependencyTaskNames（基于 project1.dependencyProjects）
3. 遍历上游工作名 dependencyTaskName
4. 取到上游工作（上一步初始化而来）dependencyTask
5. 补充 task1 的 dependencies
6. 补充 dependencyTask 的 dependents

function run(projects: Project[], actionName: string, limit: number) {
  // ...
  // project 与 project 对应 task 的上游工作名称
  function getDependencyTaskNames(project: Project): Set<string> {const dependencyTaskNames: Set<string> = new Set();
    // 遍历上游我的项目
    for (const dep of project.dependencyProjects) {
      // 收集上游工作名
      dependencyTaskNames.add(getTaskName(dep, actionName));
    }

    return dependencyTaskNames;
  }

  projects.forEach((project) => {
    // 1. 获取以后我的项目对应的工作
    const task = tasks.get(getTaskName(project, actionName))!;
    // 2. 获取当前任务对应的上游工作名
    const dependencyTaskNames = getDependencyTaskNames(project);
    // 3. 遍历上游工作名
    for (const dependencyName of dependencyTaskNames) {
      // 4. 取到上游工作（上一步初始化而来）const dependency: Task = tasks.get(dependencyName)!;
      // 5. 补充当前任务的 dependencies
      task.dependencies.add(dependency);
      // 6. 补充上游工作的 dependents
      dependency.dependents.add(task);
    }
  });
}

并行执行工作

function run(projects: Project[], actionName: string, limit: number) {
  // ...
  const taskQueue: Task[] = [];
  for (const [, task] of tasks) {taskQueue.push(task);
  }
  runTasks(taskQueue, limit);
}

async function runTasks(taskQueue: Task[], limit: number) {
  let currentActiveTasks = 0;
  function getNextTask() {for (let i = 0; i < taskQueue.length; i++) {const task: Task = taskQueue[i];
      // 返回筹备好执行的工作
      if (task.dependencies.size === 0) {return taskQueue.splice(i, 1)[0];
      }
    }
    return null;
  }

  function _run(task: Task): Promise<void> {return task.fn().then(() => {console.log("act success", task.name);
      currentActiveTasks--;
      // 当前任务执行实现，从其上游工作的 dependencies 中移除当前任务
      task.dependents.forEach((dependent: Task) => {dependent.dependencies.delete(task);
      });
      // 继续执行
      start();});
  }

  async function start() {
    let ctask: Task | null = null;
    const taskPromises: Promise<void>[] = [];
    while (currentActiveTasks < limit && (ctask = getNextTask())) {
      currentActiveTasks++;
      const task: Task = ctask;
      taskPromises.push(_run(task));
    }

    await Promise.all(taskPromises);
  }

  start();}

执行 run(projects, "build", 12)，能够依照正确程序输入后果。

act success @monorepo/z:build
act success @monorepo/y:build
act success @monorepo/x:build
act success @monorepo/d:build
act success @monorepo/b:build
act success @monorepo/a:build
act success @monorepo/c:build

要害门路长度

上文中的实现使得工作能够依照正确的程序执行，然而在理论工作执行过程中，最长的工作链限度了整个工作树的执行速度，效率不能失去保障。

要害门路长度：工作间隔最远的根节点的间隔。

interface Task {
  name: string;
  dependencies: Set<Task>;
  dependents: Set<Task>;
  // 关联门路长度
  criticalPathLength?: number;
  fn: () => Promise<void>;}

function run(projects: Project[], actionName: string, limit: number) {
  // ...
  const taskQueue: Task[] = [];
  for (const [, task] of tasks) {
    // 计算要害门路长度
    task.criticalPathLength = calculateCriticalPaths(task);
    taskQueue.push(task);
  }
  // 基于要害门路长度对工作进行降序排序
  taskQueue.sort((a, b) => b.criticalPathLength! - a.criticalPathLength!);
  runTasks(taskQueue, limit);
}

// 计算要害门路长度
function calculateCriticalPaths(task: Task): number {
  // 反复走到某一个工作了 间接返回值
  if (task.criticalPathLength !== undefined) {return task.criticalPathLength;}

  // 如果没有 dependents, 阐明咱们是 "root"，即 app 此类不被依赖的 project
  if (task.dependents.size === 0) {
    task.criticalPathLength = 0;
    return task.criticalPathLength;
  }

  // 递归向上取最大值 每次 +1
  const depsLengths: number[] = [];
  task.dependents.forEach((dep) =>
    depsLengths.push(calculateCriticalPaths(dep))
  );
  task.criticalPathLength = Math.max(...depsLengths) + 1;
  return task.criticalPathLength;
}

Selecting subsets of projects

理论业务开发中，个别不须要构建 Monorepo 内全副的我的项目，在 利用级 Monorepo 优化计划 一文中介绍了应用 Monorepo 形式治理业务我的项目可能遇到的一些坑点以及相干解决方案，其中有这样一个问题：

公布速度慢

若须要公布 app1，则所有 package 均会被构建，而非仅有 app1 依赖的 package1 与 package2 的 build 脚本被执行。

最终通过 Rush 提供的 Selecting subsets of projects 能力解决了以上问题。

具体命令如下：

# 只会执行 @monorepo/app1 及其依赖 package 的 build script
$ rush build -t @monorepo/app1

-t PROJECT即 --to PROJECT，前面PROJECT 参数为此次工作的起点我的项目包名，若不想蕴含起点我的项目，能够改为 -T 参数，即--to-except PROJECT，与之类似的可挑选出我的项目子集的参数还有：

• -f PROJECT, --from PROJECT
• -F PROJECT, --from-except PROJECT
• -i PROJECT, --impacted-by PROJECT
• -I PROJECT, --impacted-by-except PROJECT
• -o PROJECT, --only PROJECT

如果不指定这些参数，那么默认会执行所有我的项目（rush.json 中注册过的我的项目）的对应的 npm scripts。

当然，也能够指定多个参数，最差状况获取到的 subsets 是 projects 自身，与不指定参数体现统一（个别不会）。

理论利用场景：在 CI 阶段，会筛选出所有产生变更的我的项目及其会影响到的我的项目进行一次 build check，比方一次提交改变了 @monorepo/a 以及 @monorepo/b的源码，CI 就会执行以下命令：

$ rush build -t @monorepo/a -f @monorepo/a -t @monorepo/b -f @monorepo/b

不必放心某些 project 的工作会被反复执行，这种工作只是图里的一个入度不为零的点，挑选出须要执行工作的 subsets 后，依照后面的任务调度机制执行工作即可。

除了内置的 rush build 等命令反对以上参数（默认执行 package.json 中的 build script），Rush 也将此能力凋谢给了自定义命令，也就是说你能够自定义一个 rush foo 命令，用于调用指定范畴我的项目 package.json 中的 foo script，配合 Rush 的任务调度机制，工作能够保障执行程序（如果需要的话），具体能够参考 custom_commands 一节。

Selecting subsets of projects 的具体实现不在本文探讨范畴内。

结语

拓扑排序与要害门路，做了道题。🌚