按依赖关系执行的任务调度器

/*
 * 题目：按依赖关系执行的任务调度器
 * 要求：任务有 id、依赖列表和 run 函数。先检查缺失依赖、重复 id 和环，
 * 再在并发上限内只启动依赖全部成功的任务；失败后不再启动新任务。
 * 示例：A 完成后 B、C 可并行，D 同时依赖 B、C 时须等待两者都完成。
 */
function validateGraph(tasks) {}

function scheduleTasks(tasks, concurrency) {}

(async () => {
  const order = [];
  const results = await scheduleTasks(
    [
      {
        id: "a",
        dependencies: [],
        run: async () => {
          order.push("a");
          return 2;
        },
      },
      {
        id: "b",
        dependencies: ["a"],
        run: (values) => {
          order.push("b");
          return values.get("a") + 1;
        },
      },
      {
        id: "c",
        dependencies: ["a"],
        run: () => {
          order.push("c");
          return 4;
        },
      },
      {
        id: "d",
        dependencies: ["b", "c"],
        run: () => {
          order.push("d");
          return 5;
        },
      },
    ],
    2,
  );
  console.log(results.get("d")); // 预期：5
  console.log(order.indexOf("a") < order.indexOf("b")); // 预期：true
  console.log(order.indexOf("a") < order.indexOf("c")); // 预期：true
})().catch(console.error);

// 关键机制：先拓扑验证 DAG，运行时维护剩余依赖数与 ready 队列，并用 pump 补满并发槽位。
function validateGraph(tasks) {
  const byId = new Map();
  for (const task of tasks) {
    if (!task || typeof task.id !== "string" || typeof task.run !== "function")
      throw new TypeError("invalid task");
    if (byId.has(task.id)) throw new Error(`duplicate task id: ${task.id}`);
    byId.set(task.id, task);
  }
  const indegree = new Map();
  const successors = new Map(tasks.map((task) => [task.id, []]));
  for (const task of tasks) {
    const dependencies = task.dependencies || [];
    const unique = new Set(dependencies);
    if (unique.size !== dependencies.length)
      throw new Error(`duplicate dependency in ${task.id}`);
    indegree.set(task.id, dependencies.length);
    for (const dependency of dependencies) {
      if (!byId.has(dependency))
        throw new Error(`unknown dependency: ${dependency}`);
      successors.get(dependency).push(task.id);
    }
  }
  const copy = new Map(indegree);
  const queue = tasks
    .filter((task) => copy.get(task.id) === 0)
    .map((task) => task.id);
  let visited = 0;
  for (let head = 0; head < queue.length; head += 1) {
    const id = queue[head];
    visited += 1;
    for (const next of successors.get(id)) {
      copy.set(next, copy.get(next) - 1);
      if (copy.get(next) === 0) queue.push(next);
    }
  }
  if (visited !== tasks.length) throw new Error("dependency cycle detected");
  return { byId, indegree, successors };
}

function scheduleTasks(tasks, concurrency) {
  if (!Array.isArray(tasks))
    return Promise.reject(new TypeError("tasks must be an array"));
  if (!Number.isInteger(concurrency) || concurrency < 1)
    return Promise.reject(new RangeError("invalid concurrency"));
  let graph;
  try {
    graph = validateGraph(tasks);
  } catch (error) {
    return Promise.reject(error);
  }
  if (!tasks.length) return Promise.resolve(new Map());
  const ready = tasks
    .filter((task) => graph.indegree.get(task.id) === 0)
    .map((task) => task.id);
  let readyHead = 0;
  const results = new Map();
  let active = 0;
  let completed = 0;
  let stopped = false;

  return new Promise((resolve, reject) => {
    function pump() {
      while (!stopped && active < concurrency && readyHead < ready.length) {
        const id = ready[readyHead++];
        active += 1;
        Promise.resolve()
          .then(() => graph.byId.get(id).run(results))
          .then((value) => {
            results.set(id, value);
            completed += 1;
            for (const next of graph.successors.get(id)) {
              graph.indegree.set(next, graph.indegree.get(next) - 1);
              if (graph.indegree.get(next) === 0) ready.push(next);
            }
          })
          .catch((error) => {
            stopped = true;
            reject(error);
          })
          .finally(() => {
            active -= 1;
            if (stopped) return;
            if (completed === tasks.length) resolve(results);
            else pump();
          });
      }
    }
    pump();
  });
}

// 三组以上可直接运行的验证；异步题使用独立执行块。
(async () => {
  const order = [];
  const results = await scheduleTasks(
    [
      {
        id: "a",
        dependencies: [],
        run: async () => {
          order.push("a");
          return 2;
        },
      },
      {
        id: "b",
        dependencies: ["a"],
        run: (values) => {
          order.push("b");
          return values.get("a") + 1;
        },
      },
      {
        id: "c",
        dependencies: ["a"],
        run: () => {
          order.push("c");
          return 4;
        },
      },
      {
        id: "d",
        dependencies: ["b", "c"],
        run: () => {
          order.push("d");
          return 5;
        },
      },
    ],
    2,
  );
  console.log(results.get("d")); // 预期：5
  console.log(order.indexOf("a") < order.indexOf("b")); // 预期：true
  console.log(order.indexOf("a") < order.indexOf("c")); // 预期：true
  console.log(order.indexOf("d") > order.indexOf("b")); // 预期：true
  console.log(await Promise.resolve(scheduleTasks([{ id: "x", dependencies: ["x"], run() {} }], 1)).then(() => false, () => true)); // 预期：true（拒绝）
  const independent = Array.from({ length: 1000 }, (_, i) => ({
    id: String(i),
    dependencies: [],
    run: () => i,
  }));
  console.log((await scheduleTasks(independent, 8)).size); // 预期：independent.length


})().catch(console.error);