Promiseで依存関係の並行ロードを実装する

概要

　様々な場面¹で依存関係がある複数の処理を実行したいことがあります。そのためには依存関係にある処理をモデル化し実行順を制御する必要があります。さらに依存関係にない複数の処理を並行して実行したいこともあります。この記事では JavaScript (TypeScript) で依存関係の解決と並行ロードを実装します。

タスクのモデル化

　例えばタスク A、A に依存するタスク B, C、B, C に依存するタスク D があったとします。これらの依存関係は次のようなグラフで表現できます (依存関係グラフ)。矢印は依存元から依存先へ (実行される順に) 並んでいます。

タスク	依存先
A	-
B	A
C	A
D	B, C

これらを一旦 TypeScript に落とし込みます。それぞれのタスクは 0.1 秒待って終了するもので、上に示した通りの依存関係になっています。

export type Task = {
  name: string;
  runner: () => Promise<void>;
  status: 'await' | 'running' | 'done';
  dependencies: Task[];
};

export const task = function (
  name: string,
  runner: Task['runner'],
  dependencies: Task['dependencies'],
): Task {
  const task = {
    name,
    runner: async () => {
      task.status = 'running';
      await runner();
      task.status = 'done';
    },
    status: 'await' as Task['status'],
    dependencies,
  };

  return task;
};

const seconds = (s: number) => {
  return new Promise((resolve) => setTimeout(resolve, s * 1000));
};

const taskA = task('taskA', async () => {
  console.log('TaskA start');
  await seconds(0.1);
  console.log('TaskA end');
}, []);
const taskB = task('taskB', async () => {
  console.log('TaskB start');
  await seconds(0.1);
  console.log('TaskB end');
}, [taskA]);
const taskC = task('taskC', async () => {
  console.log('TaskC start');
  await seconds(0.1);
  console.log('TaskC end');
}, [taskA]);
const taskD = task('taskD', async () => {
  console.log('TaskD start');
  await seconds(0.1);
  console.log('TaskD end');
}, [taskB, taskC]);

トポロジカルソートでスケジューリングする方法

　一般的な依存関係の解決方法としてトポロジカルソートを用いる方法があります。トポロジカルソートは有向非巡回グラフのノードの集合に全順序を作成するもので、平たく言うと依存関係のないタスクを依存関係の矛盾なく一列にならべることができます。

　トポロジカルソートは DFS (深さ優先探索) や BFS (幅優先探索) で実装することができます。以下の例は DFS を用いて実装した例です。ジェネレータ関数を用いることでシンプルに実装することができます。

const tsort = function* (tasks: Task[]): Generator<Task> {
  // 訪問済みノード
  const visited = new Set<Task>();

  // ノード以下を探索
  const search = function* (task: Task): Generator<Task> {
    // 訪問済みなら探索しない
    if (visited.has(task)) return;

    // 依存先を列挙した後、
    for (const t of task.dependencies) {
      yield* search(t);
    }

    // 自身を出力
    visited.add(task);
    yield task;
  };

  // すべてのノードに対して探索: 非連結グラフでも処理をするため
  for (const task of tasks) {
    yield* search(task);
  }
};

このようにソートされたタスクは簡単に順に実行することができます。

for (const task of tsort(tasks)) {
  await task.runner();
}

TaskA start
TaskA end
TaskB start
TaskB end
TaskC start
TaskC end
TaskD start
TaskD end

平行に実行する

　トポロジカルソートを用いることですべてのタスクを直列に並べることができました。そのため用意に 1 つずつタスクを実行することができました。一方で実際には直接依存関係にない複数のタスクを同時に実行したい場合があります。例えば決まった数の WebWorker を使い回す場合です。以下の例はトポロジカルソートしたものの先頭から最大で一定個数のタスクを同時に実行する例です。

JavaScript (TypeScript) は単一スレッドで動作するためあまり注意する点はありませんが、並列処理を行う場合 numWorkers などの共有変数の扱いに注意する必要があります。

const sortedTasks = [...tsort(tasks)];

const maxWorkers = 4;
let numWorkers = 0;
let index = 0;
const next = function () {
  while (numWorkers < maxWorkers && index < sortedTasks.length) {
    const task = sortedTasks[index];
    // 前提タスクがすべて終了している
    if (task.dependencies.some((t) => t.status !== 'done')) return;

    numWorkers += 1;
    index += 1;
    task.runner().then(next);
  }
};

next();

TaskA start
TaskA end
TaskB start
TaskC start
TaskB end
TaskC end
TaskD start
TaskD end

実行できるものから順に実行していく方法

　前述のトポロジカルソートを用いたスケジューリングでは全てのタスクを直列に並べることができました。また直接依存関係にないタスクを平行に実行することができましたが、前述の実装では不十分な点があります。

　それは十分に並行実行可能なリソースを消費しきれない場合があるということです。トポロジカルソートの実装方法は複数あり、特に DFS の場合は前後に依存関係のあるタスクが並びやすい傾向にあるため並行に実行されにくいことになります。また実行可能性の判定のために依存タスクが完了しているかを判定しているため、これを許容するのであればトポロジカルソートを行う意義は薄れると言えます。

　そこであるタスクの完了時にすべてのタスクが実行可能であるかを判定し、並行実行可能なリソースに達するまで同時に実行させるという方法を用います。

const maxWorkers = 4;
let numWorkers = 0;
const scheduledTasks = [...tasks];

const execute = function () {
  // 配列の中身が変化するため iterator ではなくインデックスでループ
  for (let i = 0; i < scheduledTasks.length; i++) {
    if (numWorkers >= maxWorkers) break;

    const task = scheduledTasks[i];
    if (task.dependencies.some((t) => t.status !== 'done')) continue;

    // scheduledTasks から削除
    scheduledTasks.splice(i, 1);
    i -= 1;

    numWorkers += 1;
    (async () => {
      task.status = 'running';
      await task.runner();
      task.status = 'done';
      numWorkers -= 1;
      execute();
    })();
  }
};

execute();

TaskA start
TaskA end
TaskB start
TaskC start
TaskB end
TaskC end
TaskD start
TaskD end

　この方法の欠点として毎回すべてのタスクが実行可能であるかを検査するため余分な処理が発生している点やこの実行可能であるかを比較する処理が $\mathcal{O}(N^2)$ である点、タスクの実行の前後処理のクロージャによりスタックオーバーフローが発生する可能性がある点が挙げられます。

Event を用いて改善する

　前述の問題を Event を用いて解決します。Task を EventTarget として実装することで依存されるタスクから依存するタスクへのフローを作成し、依存するタスクではすべての依存タスクが完了した時点で実行されるようにします。以下の例では待っているタスクの数を変数で管理し、終了イベントを受け取るごとにカウントを減らしています。これにより前述の計算量の問題が解決する他、イベントにより実行されるため環境のキャプチャによるスタックオーバーフローの問題も解消しています。

JavaScript (TypeScript) はシングルスレッドで動作し、イベントのコールバックはイベントループをブロックするため完了イベントが握りつぶされる心配はありません。

class Task extends EventTarget {
 name: string;
 runner: () => Promise<void>;
 waitingTasks: number;

 constructor(
   name: Task['name'],
   runner: Task['runner'],
   dependencies: Task[],
 ) {
   super();

   this.name = name;
   this.runner = runner;
   this.waitingTasks = dependencies.length;

   for (const task of dependencies) {
     task.addEventListener('done', () => {
       this.waitingTasks -= 1;
       this.runIfReady();
     });
   }
 }

 async runIfReady() {
   if (this.waitingTasks !== 0) return;

   await this.runner.call(this);
   this.dispatchEvent(new Event('done'));
 }
}

const seconds = (s: number) => {
 return new Promise((resolve) => setTimeout(resolve, s * 1000));
};

const taskA = new Task('taskA', async () => {
 console.log('TaskA start');
 await seconds(0.1);
 console.log('TaskA end');
}, []);
// ... (TaskB ～ TaskD も同様)

const tasks = [taskA, taskB, taskC, taskD];

for (const task of tasks) {
 task.runIfReady();
}

まとめ

　これまで依存関係のある複数のタスクを平行に実行する方法を模索してきました。私の結論としては最後の Event を用いた実装が JavaScript の処理系に適した方法だと考えています。他に良い案があったら教えていただけると幸いです。

1. 副作用のために実行順の保証を必要とする複数の処理の結合度を弱めたいとき。例えばプラグインの呼び出しや複数の機能のあるブラウザ拡張機能など。 ↩