限界突破のログパイプライン：FiberとイベントループがPHPの常識を書き換える
PHPにおける「並行処理」という言葉を聞いて、未だに「pcntl_forkによるプロセスプレーンの乱立」や「cURLのマルチハンドルによる苦し紛れのI/O多重化」を思い浮かべるなら、今すぐその古いパラダイムを捨ててほしい。
PHP 8.1で導入された Fiber（ファイバー） は、言語レベルでの協調的マルチタスキング（Cooperative Multitasking）を我々に手渡した。Zend VMのコールスタックを独立したヒープ領域に退避・復元させるこのプリミティブは、I/O待ちのレイテンシを完全に隠蔽し、シングルスレッドのまま数千の並行タスクをドライブする「非同期データパイプライン」の構築を可能にする。
今回は、大規模ログ収集・分析システムという「絶えずストリームデータが流れ込み、ストレージやネットワークI/Oのボトルネックが牙を剥く極限環境」を想定し、Fiberの本質を抉るアーキテクチャと実装を解説する。
—
1. 内部構造の理解：FiberがZend VMとメモリ空間に何をもたらすか
一般的なOSスレッドやプロセスフォークは、コンテキストスイッチのたびにカーネルモードへの遷移コストとメモリ（スタック領域）の消費を伴う。これに対し、Fiberはユーザースペースでの軽量スレッドである。
コールスタックの分離とヒープ割り当て
Zend VMは通常、関数呼び出しのたびに実行コンテキスト（`zend_execute_data`）をグローバルなコールスタック上に積んでいく。しかし、`Fiber`インスタンスが生成された瞬間、PHPのランタイムは独立したヒープ領域にコールスタックを確保する。
これにより、開発者は任意の深さの関数呼び出しの途中で処理を一時停止（`Fiber::suspend()`）し、制御をイベントループに返却、外部I/Oの完了後に元の位置から再開（`Fiber::resume()`）できる。
[Main Event Loop] ──(非同期I/O監視)──> [Socket Ready]
│ │
▼ ▼
Fiber #1 (Suspending) ──────────> Fiber #1 (Resuming)
[処理A: ログ読込] [処理B: ログ書き込み]
この挙動の美しさは、コードの見た目を完全に「同期的な手続き型」に維持しながら、実行モデルだけを「非同期・ノンブロッキング」にすり替えられる点にある。コールバック地獄（コールバックヘル）に陥りがちなNode.js的非同期モデルへの、PHPからの最もエレガントな回答がここにある。
—
2. 設計指針：堅牢な非同期ログパイプラインの要件
ログ収集システムにおいて、スループットとメモリ効率のバランスを誤ると、瞬く間にOOM（Out of Memory）を引き起こすか、I/Oブロックによるスレッド枯渇に直面する。実務のコードレビューで私が必ずチェックする3つの鉄則を挙げる。
1. バックプレッシャー（流量制御）の欠如は悪
ログの生産速度が消費（ストレージ書き込み）速度を上回った際、無限にメモリ上にキューを溜め込む設計は自殺行為である。チャネル（Channel）概念を導入し、バッファ溢れ時は生産側をサスペンドさせよ。
2. 例外の伝播とFiberの寿命管理
Fiber内で未キャッチの例外が発生した場合、そのFiberは即座に破棄され、デストラクタでリソースリークが起きる。イベントループ側で例外を安全にキャッチし、パイプライン全体を巻き込まない耐障害性が必要である。
3. ノンブロッキングI/Oの徹底
Fiberを使っても、その内部でブロッキング関数（通常の `file_put_contents` や同期型TCPソケットなど）を叩けば、プロセス全体がフリーズする。ストリームは必ず非同期モード（`stream_set_blocking($stream, false)`）に設定し、`stream_select`等と組み合わせよ。
—
3. 実装：Fiber駆動型 非同期ログデータパイプライン
それでは、実務の現場にそのまま投入できるレベルまで研ぎ澄ませたリファレンスコードを提示する。ログの「読込（Tail）」「前処理（Filter/Transform）」「ストレージ出力（Batch Flush）」をFiberの協調動作で結合したパイプラインだ。
/
class AsyncLogPipelineEngine
{
/ @var \SplQueue
private \SplQueue $fiberQueue;
/ @var array
private array $readPoll = [];
/ @var bool イベントループの稼働フラグ /
private bool $isRunning = false;
public function __construct()
{
$this->fiberQueue = new \SplQueue();
}
/
- 新しいタスク（Fiber）をループに登録する
/
public function addTask(\Closure $taskLogic): void
{
$fiber = new Fiber($taskLogic);
$this->fiberQueue->enqueue($fiber);
}
/
- 非同期でストリームからのデータを待機し、読み込めたらFiberを再開する
/
public function awaitRead($stream, Fiber $fiber): void
{
$id = (int)$stream;
$this->readPoll[$id] = [$stream, $fiber];
// 制御をイベントループへ返却（サスペンド）
Fiber::suspend();
}
/
- イベントループの駆動
/
public function run(): void
{
$this->isRunning = true;
while ($this->isRunning) {
// 1. キューに溜まったFiberを順次実行（または再開）
while (!$this->fiberQueue->isEmpty()) {
$fiber = $this->fiberQueue->dequeue();
try {
if (!$fiber->isStarted()) {
$fiber->start($this);
} elseif (!$fiber->isTerminated()) {
$fiber->resume();
}
} catch (\Throwable $e) {
// 本番環境ではここで適切なロギングおよびアラート発報を行う
fprintf(STDERR, “[CRITICAL] Fiber Error: %s in %s:%d\n”, $e->getMessage(), $e->getFile(), $e->getLine());
}
// 実行権が返却された際、Fiberがまだ生きている（suspend状態）なら適切に管理
if ($fiber->isSuspended() && !$this->isManagedInPoll($fiber)) {
// 即座に再スケジュールが必要な場合
$this->fiberQueue->enqueue($fiber);
}
}
// 2. 実行すべきFiberがなく、かつI/O待ちもない場合はループ終了
if (empty($this->readPoll) && $this->fiberQueue->isEmpty()) {
break;
}
// 3. I/O多重化 (stream_selectによるノンブロッキング監視)
if (!empty($this->readPoll)) {
$readStreams = [];
foreach ($this->readPoll as $id => [$stream, $fiber]) {
$readStreams[$id] = $stream;
}
$write = $except = [];
// タイムアウトを短く設定してビジーウェイトを防ぐ (例: 10ms = 10000マイクロ秒)
$selected = @stream_select($readStreams, $write, $except, 0, 10000);
if ($selected === false) {
break; // ストリームエラー
}
if ($selected > 0) {
foreach ($readStreams as $stream) {
$id = (int)$stream;
if (isset($this->readPoll[$id])) {
[$stream, $fiber] = $this->readPoll[$id];
unset($this->readPoll[$id]);
// I/Oの準備が整ったのでFiberキューに戻す
$this->fiberQueue->enqueue($fiber);
}
}
}
}
}
}
private function isManagedInPoll(Fiber $fiber): bool
{
foreach ($this->readPoll as [$s, $f]) {
if ($f === $fiber) return true;
}
return false;
}
}
// ==========================================
// 実践：ログパイプラインの構築と実行
// ==========================================
// テスト用の一時ファイル（ログソース）を作成
$logFile = sys_get_temp_dir() . ‘/access_stream.log’;
if (!file_exists($logFile)) {
file_put_contents($logFile, “”);
}
$engine = new AsyncLogPipelineEngine();
// 【タスクA】ログファイルを擬似的にTailし続けるプロデューサーFiber
$engine->addTask(function (AsyncLogPipelineEngine $engine) use ($logFile) {
// 実際の実装ではここで非同期TCPサーバーやtail -f相当のストリームを開く
$stream = fopen($logFile, ‘r’);
stream_set_blocking($stream, false);
echo “[Producer] ログ監視ストリームを開始しました。\n”;
$counter = 0;
while ($counter < 5) {
$line = fgets($stream);
if ($line === false) {
// データがない場合はI/Oイベントを待つためにサスペンド
$engine->awaitRead($stream, Fiber::getCurrent());
continue;
}
$line = trim($line);
if ($line !== ”) {
echo “[Producer] ログを検知: {$line}\n”;
$counter++;
}
}
fclose($stream);
echo “[Producer] 規定のログ収集を完了しました。\n”;
});
// 【タスクB】定期的にバックグラウンドで集計・フラッシュを行うコンシューマーFiber
$engine->addTask(function (AsyncLogPipelineEngine $engine) {
echo “[Consumer] バッファ監視ワーカー起動。\n”;
for ($i = 0; $i < 3; $i++) {
// 処理を0.5秒間手放す（疑似的なタイマー協調動作）
$start = microtime(true);
while (microtime(true) - $start < 0.5) {
Fiber::suspend(); // イベントループに制御を戻し、CPUを解放
}
echo "[Consumer] ストレージへのバッファフラッシュ実行 (Tick {$i})\n";
}
echo "[Consumer] ワーカー終了。\n";
});
// 外部プロセスからテストデータを書き込むための別スレッド的アプローチ（動作確認用）
// 実際の運用では別プロセスやFluentd等からのストリームインジェクションを想定
register_shutdown_function(function() use ($logFile) {
@unlink($logFile);
});
// バックグラウンドで非同期にファイルへ書き込みを行うダミー処理を別プロセスでなく、
// タイマー代わりに動かすか、あるいはループ開始直後にファイルへ追記する
// ここではシンプルにループをキック
$pid = pcntl_fork();
if ($pid === 0) {
// 子プロセス：100msごとにログを書き込む
$fp = fopen($logFile, 'a');
for ($i = 1; $i <= 5; $i++) {
usleep(200000); // 0.2秒待機
fwrite($fp, json_encode(['level' => ‘INFO’, ‘message’ => “User action #{$i}”, ‘time’ => microtime(true)]) . “\n”);
fflush($fp);
}
fclose($fp);
exit(0);
}
// メインのイベントループ開始
$engine->run();
// 子プロセスの終了を待つ
pcntl_wait($status);
コードの解説と内部挙動のポイント
1. `stream_set_blocking($stream, false)` と `stream_select` の組み合わせ
PHPにおける真の非同期I/Oの要である。OSレベルでのソケットやファイル記述子のステータス変化をポーリングし、CPUコアを100%占有（ビジーウェイト）させることなく、イベント駆動を実現している。
2. 協調的マルチタスキングの極意
`Fiber::suspend()` を呼び出すことで、Zend VMはその時点のスタックフレームを安全に保存し、呼び出し元の `run()` メソッドへ制御を戻す。これにより、重いデータ処理の合間に他のタスク（HTTPリクエストの処理や別ストリームの読み込みなど）を挟み込むことが可能になる。
—
4. チーフアーキテクトからの警句：本番運用における罠
Fiberは万能の銀の弾丸ではない。以下のアンチパターンに踏み込んだ瞬間、あなたのシステムは破綻する。
- ブロッキングなサードパーティ製ライブラリの利用
内部で同期的な外部APIコール（例えば古いバージョンの `PDO` による重いクエリや、同期型の `guzzlehttp/guzzle`）をFiber内で実行した場合、その瞬間にプロセス全体がブロックされ、他のすべてのFiberが停止する。非同期化するならエコシステム全体が非同期対応（AmpやReactPHPなどのエコシステム、あるいは完全なノンブロッキングドライバ）で統一されている必要がある。
- デバッグの難易度上昇
非同期処理の常として、スタックトレースが非線形になる。例外発生時に「どのFiberの、どのコンテキストで起きたエラーか」を追跡できるよう、Fiberインスタンス生成時に必ず一意なID（UUIDや連番）を付与し、ログのコンテキストにバインドする設計を義務づけよ。
PHPはもはや、単なる「リクエストごとに死んで生まれ変わるスクリプト言語」ではない。Fiberを使いこなすことで、高スループットなデーモンプロセスやリアルタイムデータパイプラインを極めて高い安全性と保守性で実装できる。
この知見を武器に、あなたのプロダクトのアーキテクチャを次のステージへ引き上げてほしい。