🔒

非同期処理版mutex

rentaro

2026/07/27に公開

mutex, Inc. でエンジニアをしている加藤です。業務では TypeScript・Dart を書くことが多いです。

はじめに

当社の名前でもある mutex をタイトルに掲げるのは少し気が引けるのですが、非同期関数の競合問題を、tony の知見を借りて mutex の考え方で解決できたので、このタイトルで記事にします。詳しい方には既知の話かもしれませんが、実務でこの構造に出会った一例の紹介です。

前提：この記事で扱う状況

きっかけは、Flutter アプリで record パッケージ（当時 ^6.2.0）を使った録音機能の開発でした。この記事ではこれを題材にします。この問題は特定の言語ではなく async/await というパターンを持つ言語に共通で見られますが、コード例は題材に合わせて Dart で書きます。録音機能は、次の性質を持っています。

実体が1つしかないリソース（マイク）を扱う：今回の構成では、録音処理を二重に開始するとネイティブ側で競合し、アプリがクラッシュします
イベントの通知は同期的：割り込みやバックグラウンド移行は、OS から同期コールバックで届きます
状態は同期的に読める：コールバックでは、その瞬間のマイクの状態を読み取れます
状態の変更は非同期：停止・再開はマイクの解放・取得や OS のセッション操作を伴うため、呼び出してから完了するまでに await を挟みます

マイク起動（録音中）
  ↓
イベント発生 …… ユーザーのタップ、電話の割り込み、バックグラウンド移行など様々な操作によって発生
  ↓
通知・確認（同期）…… コールバックが呼ばれ、mic.state のような状態をその場で読む
  ↓
停止・再開に伴う処理（非同期）…… 録音の開始・停止、ファイルやセッションの操作。await を跨ぐ

つまり、状態はその場で読めるのに、状態を変える操作はその場では完了しません。ここに「チェックした瞬間は正しいが、操作が完了する頃には古くなっているかもしれない」という隙間が生まれます。しかも、マイクの状態を変えるきっかけはユーザー操作、割り込み、バックグラウンド移行など複数あります。そのため競合を前提に、競合したときにどうするかを決めることになります。

コードで書くと、「割り込みの開始でマイクを停止し、割り込みの終了やアプリ復帰で再開する」という処理です。以下のコードは DartPad に貼り付けてあるので、Run を何度か押して実行してみてください（TypeScript 版は TypeScript Playground にあります）。通知がランダムなタイミングで発生するため、成功することもあれば、エラーログが出ることもあります。

デモコード全文（クリックで展開）

import 'dart:math';

enum MicState { on, off }

// マイクの実体（デバイスに1つ）は OS の中。アプリは同期で読むか、「切り替え」を依頼するかしかできない
class Mic {
  MicState _state = MicState.on;

  Future<void> _deviceIO() =>
      Future.delayed(const Duration(milliseconds: 5)); // 時間がかかるハードウェア操作

  MicState get state => _state; // 検知は同期。読めるのは「その瞬間」の値だけ

  Future<void> release() async {
    await _deviceIO();
    if (_state == MicState.off) throw Exception('停止済みのマイクを停止した'); // クラッシュ
    _state = MicState.off;
  }

  Future<void> acquire() async {
    await _deviceIO();
    if (_state == MicState.on) throw Exception('使用中のマイクを二重取得した'); // クラッシュ
    _state = MicState.on;
  }
}

final mic = Mic(); // デバイスに1つだけ

// 割り込みの開始で呼ばれる：録音中なら停止する
Future<void> stopForInterruption() async {
  if (mic.state == MicState.on) { // 停止済みのマイクを二重に停止しないためのチェック
    await mic.release(); // 解放。完了して初めて off になる
    print('停止した');
  }
}

// 割り込みの終了やアプリ復帰で呼ばれる：止まっていたら再開する
Future<void> restartRecording() async {
  if (mic.state == MicState.off) { // 使用中のマイクを二重に取得しないためのチェック
    await mic.acquire(); // 取得。完了して初めて on になる
    print('再開した');
  }
}

// 実機ならアプリごと落ちる。ここではログで見えるようにする
void logCrash(Object e) => print('💥 クラッシュ: $e');

// OS の通知は同期コールバックで届く
// 停止・再開の非同期処理は、完了を待たれないまま走る
void onOsNotification(String event) {
  if (event == '割り込み開始' || event == 'バックグラウンド移行') {
    stopForInterruption().catchError(logCrash);
    return;
  }
  restartRecording().catchError(logCrash);
}

// 通知はいつ・何が届くかわからない。電話では「開始」と「移行」、「終了」と「復帰」がほぼ同時に届く
void main() {
  final random = Random();
  onOsNotification('割り込み開始');
  Future.delayed(Duration(milliseconds: random.nextInt(10)),
      () => onOsNotification('バックグラウンド移行'));
  Future.delayed(Duration(milliseconds: 15 + random.nextInt(30)),
      () => onOsNotification('割り込み終了'));
  Future.delayed(Duration(milliseconds: 15 + random.nextInt(30)),
      () => onOsNotification('アプリ復帰'));
}

// 出力例: 停止した
//         💥 クラッシュ: Exception: 停止済みのマイクを停止した
//         再開した

壊れる原因は、if (mic.state == MicState.on) のようなチェックが実行中に古びることです。await はスレッドを止めて待つのではなく、続き（次の行以降）を Future^[1] に登録して呼び出し元に制御を返します。制御を手放している隙間では別の処理が走れるので、チェックした瞬間は正しくても、操作の完了を待っている間に別の呼び出しが同じチェックを通過します。ある時点で状態を確認して決めたことは、次の await を越えた後も正しいとは限りません。

この記事の Dart の例では、コードは1つのイベントループ（isolate）上で動きます。同期コードの実行中に別のコールバックが割り込むことはなく、await で制御を手放した後に他の処理が進みます。仕組みの詳細はこちらの記事を（JavaScript のイベントループは Jake Archibald の名記事を）読んでください。

「await stopForInterruption() にすれば直るのでは」と思うかもしれません。しかし、呼び出し元の onOsNotification は OS から呼ばれる同期コールバックです。ここから開始した非同期処理の完了を、通知元に待たせることはできません。停止・再開は投げっぱなしで走ることになり、独立に開始された複数の処理が重なります。

つまり競合の正体は、投げっぱなしで独立に走る複数の非同期処理が、await の隙間で互いのチェックを無効化し合うことです。ここからは対処法の話をします。題材は録音機能ですが、async/await というパターンそのものに潜む話です。

対策は2系統：諦めるか、待つか

ここでタイトルの mutex が登場します。mutex（mutual exclusion、相互排他）は、共有リソースに「使用中」の札を掛ける仕組みです。リソースを触りたい処理は、まず札を確認する。誰も使っていなければ札を掛けて（lock）作業し、終わったら外す（unlock）。マルチスレッドで共有メモリを守る定番の道具です。mutex そのものの詳しい解説は、こちらを読んでください。

そして、札が掛かっているときに来た処理をどう扱うかが、そのまま対策の2系統になります。諦めて引き返す（mutex でいう tryLock：取れなければ失敗を返す）か、外れるまで待つ（lock：獲得できるまでブロックする）かです。Java の Lock インターフェースにおける両者の対比は、こちらの記事がわかりやすいです。

諦める対処法

1つ目の「諦める対処法」は、遷移中フラグと try/finally で実装できます。非同期の状態遷移を始める前に「遷移中」フラグを同期的に立て、遷移中に来た他の呼び出しは実行せずに捨て、finally でフラグを戻す。競合の検出だけして、負けた側は成立を諦める発想です。

冒頭のコードの stopForInterruption と restartRecording の2つを次のコードに差し替えると（他はそのまま）、実行してもクラッシュが出なくなったことを確認できます。

bool isTransitioning = false; // 「いま状態遷移中か」を表すフラグ

Future<void> stopForInterruption() async {
  if (isTransitioning) return; // 遷移中なら諦める。ここと次の行の間に await はない
  isTransitioning = true;      // 「使用中」の札を掛ける。以降に来た呼び出しは冒頭の return で引き返す
  try {
    if (mic.state == MicState.on) { // 同じフラグを守る処理とは競合しない。停止済みならここで降りる
      await mic.release();
      print('停止した');
    }
  } finally {
    isTransitioning = false;   // どの経路で抜けても必ず戻す
  }
}

Future<void> restartRecording() async {
  if (isTransitioning) return;
  isTransitioning = true;
  try {
    if (mic.state == MicState.off) {
      await mic.acquire();
      print('再開した');
    }
  } finally {
    isTransitioning = false;
  }
}

// 出力: 停止した
//       再開した   （何度実行してもクラッシュは出ない）

ただし捨てた呼び出しは戻りません。「停止の途中でユーザーが再度ボタンを押した」操作は消えます。さらにマイクを触るすべての非同期関数で同じフラグを守る必要があります。フラグを戻す finally を書き忘れた関数で例外が発生すると、フラグが立ちっぱなしになります。

とはいえ、捨てても目的が達成されるなら「諦める」が正解です。保存ボタンの連打（2回目を律儀に実行すると二重保存）、定期同期（前回が走行中ならスキップしないとキューが際限なく伸びる）などがこちらです。isLoading でボタンを無効化するのも、諦める対処法の UI 版と言えます。

待つ対処法

2つ目の「待つ対処法」は、操作の直列化で実装します。前の操作が完全に終わるまで、次の操作を開始しません。呼び出された処理は捨てず、順番に実行します。

実装は Future の .then チェーン数行です。こちらも、冒頭のコードの stopForInterruption と restartRecording の2つを synchronized ごと次のコードに差し替えると（他はそのまま）、実行してもクラッシュが出なくなったことを確認できます。

Future<void> last = Future.value();

Future<T> synchronized<T>(Future<T> Function() task) {
  final next = last.then((_) => task());     // 前の完了に「つなぐ」
  last = next.then((_) {}, onError: (_) {}); // チェーン用のコピー（下の details 参照）
  return next;
}

Future<void> stopForInterruption() => synchronized(() async {
  if (mic.state == MicState.on) { // 同じキューの前のタスクが完了してから状態を読み直す
    await mic.release();
    print('停止した');
  }
});

Future<void> restartRecording() => synchronized(() async {
  if (mic.state == MicState.off) {
    await mic.acquire();
    print('再開した');
  }
});

// 出力: 停止した
//       再開した   （何度実行してもクラッシュは出ない）

新しいタスクは最後尾の Future の完了に繋がり、自分が新しい最後尾になります（以降「キューに並ぶ」と呼びます）。待つのは await の隙間ではなく前のタスク全体の完了なので、同じキューを通る後続タスクは、前のタスクが何回制御を手放しても途中に割り込めません。後続の呼び出しは自分の番が来たときに状態を確認し直します。そのため、2本目の停止は "off" を、2本目の再開は "on" を見て何もせずに終わります。結果として、マイクの解放も取得も1回ずつになります。実プロダクトでは synchronized パッケージの Lock が同じ仕組みを提供しており、今回はこちらを使いました。

注意点は1つ。この保証が効くのは synchronized を通したコード同士だけです。1箇所でも synchronized を通さずに mic.release() や mic.acquire() を呼べば、キューを素通りして元どおり割り込めてしまいます。

なお、この Future チェーンによる簡易実装は、last の読み書きが同じイベントループ上で行われることを前提にしています。複数の isolate・スレッド・プロセスから共有リソースを操作する場合は、その実行環境に対応した同期機構が別途必要です。

実装の細部（空ハンドラの理由・呼び出しが割り込まれない理由）

2行目の空ハンドラは省略ではない：onError を渡さない p.then(task) の task は p が成功で終わったときだけ実行されます（基礎は Future.then のドキュメント、JavaScript なら MDN のプロミスの使用）。last = next; と生で繋ぐと、1つの throw で last がエラーの Future に固定され、キューが永久に詰まります。空ハンドラ付きの last は必ず成功で終わるのでキューは進み、呼び出し元に返す next からエラーは届きます
この実装は同じイベントループ上で呼ばれることを前提にしている：final next = ... と last = ... の2行に await はなく、別のコールバックに割り込まれず完了します
同じキューへの再入はできない：synchronized で実行中のタスクが、同じ synchronized をもう一度呼んでその完了を await すると、内側のタスクは外側の後ろに並び、互いに待ち合って進まなくなります

順番が来たら「まだやるべきか」を確認し直す

直列化が保証するのは順番だけで、順番が来た処理が「まだやるべきこと」かどうかは保証しません。ユーザーが停止した直後に「割り込みが終わったから再開」がキューに入っていれば、止めたのに録音が再開します。作法は1つ：順番が来たら、まず「まだやるべきか」を確認し直す。

bool userWantsRecording = false;

Future<void> stop() {
  userWantsRecording = false;   // ① ユーザーの意思はキューに並ばせず、即時反映する
  return synchronized(() async { /* 停止処理 */ });
}

Future<void> autoResume() {
  return synchronized(() async {
    // ② キューに並んだ処理は、冒頭で前提を再チェックする
    if (!userWantsRecording) return;
    /* 再開処理 */
  });
}

これで、ユーザーが停止した直後に autoResume の番が来ても、②の再チェックで userWantsRecording が false だと分かり、何もせずに降ります。

①の書き込みだけキューを通していないのは、キューに並ぶ処理がどのみち②で再チェックするからです。フラグがいつ書き換わっても、次に番が来た処理には最新値として見えます。唯一のすり抜けは「再開処理が②を通過した後、実行中にフラグが false へ変わる」ケースですが、そのときも stop() が積んだ停止処理が後ろに並んでいるので、最終状態は必ず「停止」で終わります。キューの外に置いていいのは、この種の「読み違えても後続のキュー処理が回収してくれる」フラグだけです。

ロックは守りたいリソースごとに持つ

記事のコードでは last をトップレベルに置きましたが、実コードではロックをリソースの所有者のフィールドとして持ちます。筆者の Flutter 実装ではこの形です^[2]。

mixin RecordingLifecycle {
  final _transitionLock = Lock();

  // マイクを触る遷移は、すべてこれを通す
  Future<T> serializeTransition<T>(Future<T> Function() action) =>
      _transitionLock.synchronized(action);
}

キューはグローバルに1本ではなく、守りたいリソース1つにつき1本です。この持ち方なら「このロックはマイクの遷移用」というラベリングがコードの構造で表現でき、無関係な操作まで直列化して遅くする事故も防げます。

まとめ

ある時点で状態を確認して決めたことは、次の await を越えた後も正しいとは限らない。直列化した処理の中で、自分の番が来たときに状態を確認し直す
対策は mutex と同じ2系統。競合したら諦めるか、前の操作の完了を待つか
選ぶ基準は「その呼び出しを捨てていいか」。捨てても目的が達成されるなら諦める（連打、リフレッシュ、定期同期）。1つ1つの呼び出しに固有の意味があり、取りこぼすと状態がずれるなら待つ（録音の停止・再開）

mutex というとマルチスレッドの道具という印象がありますが、シングルスレッドの async/await でも、実体が1つのリソースに複数のイベントが触りに来れば、同じ考え方ができます。録音に限らず、カメラ・BLE・WebSocket も同じ構造です。それがタイトルの「非同期処理版 mutex」です。

参考

Dart asynchronous programming: Isolates and event loops：Dart のイベントループと isolate の解説
Tasks, microtasks, queues and schedules：JavaScript のイベントループの解説
マルチスレッド・プログラミングの道具箱：mutex をはじめとする同期プリミティブの解説
Java Concurrency: Lock and tryLock：lock と tryLock の対比
プロミスの使用（MDN）：Promise チェーンとエラー処理の基礎
Future.then（Dart API）：then と onError の仕様
synchronized（pub.dev）：Dart で直列化ロックを提供するパッケージ
record（pub.dev）：題材にした録音パッケージ

脚注

JavaScript/TypeScript の Promise に相当します。以降は Future と表記して話を進めます。 ↩︎
mixin は Dart でクラスに実装を混ぜ込む仕組みです。ここでは「録音を扱うクラスに組み込んで使う部品」程度の理解で大丈夫です。 ↩︎

mutex Tech BlogPublication

東京大学発のスタートアップ株式会社mutex の Tech Blogです！