Fiberを用いたバッチ処理の並列化とリソース管理：Zend VMの深淵と協調的マルチタスクの極意
PHPにおける並行処理のパラダイムは、PHP 8.1で導入された`Fiber`によって決定的な転換期を迎えた。
往年の多重プロセス（`pcntl_fork`）やノンブロッキングI/O（`stream_select`等）の泥臭い実装、あるいは外部拡張（SwooleやReactPHP）への依存から解放され、言語コアレベルの協調的マルチタスク（Cooperative Multitasking）が手に入ったのだ。
しかし、Zend VMのメモリ管理、コールスタックの動的割り当て、そしてOPcacheの挙動を理解せずに`Fiber`を乱用すれば、アプリケーションは容易にメモリリークやCPUキャッシュミスの嵐に飲み込まれる。
本稿では、数百万件規模のレコードを処理するバッチジョブを題材に、Fiberを用いた並列化とリソース管理の極限を、Zend VMの低レイヤの挙動から紐解いていく。
—
1. Zend VMにおけるFiberの物理構造とコンテキストスイッチの代償
まず、C言語レベル（Zend Engine）でFiberがどのように扱われているかを理解しなければならない。
従来のPHP関数呼び出しは、Zend VMのコールスタック（`zend_execute_data`のスタックフレーム）上で直列に処理される。関数が中断されることはなく、return または例外スローによってのみフレームが破棄される。
これに対し、`Fiber::suspend()` と `Fiber::resume()` は、以下の極めて重厚な処理を裏で行っている。
1. コールスタックの退避: 現在の `zend_execute_data` チェーン、および実行状態（OPコードのポインタ `opline` など）をヒープ上に切り出された専用のスタック領域に退避する。
2. レジスタ・コンテキストの切り替え: CPUの実行コンテキストを別のFiberのスタック領域へとスイッチする。
ここで重要なのは、Fiberはプリエンプティブ（占有型）ではなく、協調的（協調型）であるという点だ。つまり、各Fiberが明示的に `Fiber::suspend()` を呼び出さない限り、CPU権は解放されない。I/O待ちや重い演算の途中で勝手にコンテキストスイッチは発生しないため、スケジューラ側の設計が稚拙であれば、ただの「重い同期処理のラッパー」と成り下がる。
—
2. 大量データバッチ処理におけるメモリ空間（HashTable）の罠
数百万件のレコードを一度にメモリ上にロードせず、チャンク（分割）単位で処理するのはバッチ処理の定石だ。しかし、Fiberで並列タスクを多数走らせた場合、各Fiberが保持するローカル変数やPDOのステートメントがZendの `HashTable` を激しく肥大化させる。
PHPの配列はすべて `HashTable` であり、内部的にバケットの動的拡張（rehashing）を行う。並列実行される複数のFiberがそれぞれ数万件のオブジェクトを生成・保持すると、Zendのメモリマネージャ（zend_mm）は細切れのヒープ領域を割り当て、メモリ断片化（Fragmentation）を引き起こす。結果として、RSS（Resident Set Size）が暴騰し、OSのOOM Killerの標的となる。
これを防ぐためには、Fiberの数（並行度）を厳格に制限する「トークンバケツ・スケジューラ」の概念が必要不可欠である。
—
3. 実装：リソース制限を伴うFiberバッチスケジューラ
以下に、Zend VMのメモリ消費とCPU負荷を制御しながら、大量のタスクを効率的に並列処理するスケジューラのプロダクションコードを示す。
/
private array $queue = [];
public function __construct(int $maxConcurrency = 10)
{
$this->maxConcurrency = $maxConcurrency;
}
/
- タスク（ジェネレータを返すクロージャ）をキューに追加する
/
public function add(callable $taskFactory): void
{
$this->queue[] = new Fiber(function () use ($taskFactory) {
$generator = $taskFactory();
if (!$generator instanceof Generator) {
throw new \RuntimeException(‘Task must return a Generator.’);
}
foreach ($generator as $item) {
// 各チャンク処理の境界でCPU/I/Oの制御権をスケジューラへ返却（協調的マルチタスク）
Fiber::suspend($item);
}
});
}
/
- 全タスクの実行とリソース管理を行う
/
public function run(): void
{
$running = [];
while (count($this->queue) > 0 || count($running) > 0) {
// 最大並行数に達するまで、キューからFiberを取り出して起動
while (count($running) < $this->maxConcurrency && count($this->queue) > 0) {
$fiber = array_shift($this->queue);
$running[] = $fiber;
try {
// 初回スタート
$value = $fiber->start();
$this->handleYield($fiber, $value);
} catch (Throwable $e) {
$this->handleException($fiber, $e);
// 異常終了したFiberをrunningから除外
$running = array_filter($running, fn($f) => $f !== $fiber);
}
}
// 稼働中のFiberを順次再開させる（ラウンドロビン的実行）
foreach ($running as $index => $fiber) {
if ($fiber->isTerminated()) {
unset($running[$index]);
continue;
}
try {
// 前回の suspend から処理を再開
$value = $fiber->resume();
$this->handleYield($fiber, $value);
} catch (Throwable $e) {
$this->handleException($fiber, $e);
unset($running[$index]);
}
}
// CPUのスパイクを防ぎ、I/OイベントループやOSのスケジューリングに配慮する微小なウェイト
// ※実運用ではイベントループ（Stream Select等）と統合する
usleep(1000);
// 配列のインデックスを詰める
$running = array_values($running);
}
}
private function handleYield(Fiber $fiber, mixed $value): void
{
// ここで yield されたデータに対する非同期処理や進捗管理を行う
if ($value !== null) {
// 例: ログ出力やメトリクスの計測
// echo “Processed chunk data: ” . gettype($value) . PHP_EOL;
}
}
private function handleException(Fiber $fiber, Throwable $e): void
{
// ログ出力およびトランザクションのロールバック等のクリーンアップ処理
error_log(“Fiber execution failed: ” . $e->getMessage());
}
}
// ==========================================
// 実行例：100万件のデータを1000件ずつのチャンクに分割し、
Pdo等を経由して並列処理するシミュレーション
// ==========================================
$scheduler = new FiberBatchScheduler(maxConcurrency: 4);
for ($i = 1; $i <= 5; $i++) {
$batchId = $i;
$scheduler->add(function () use ($batchId) {
// 模擬的な大量データフェッチ（実際にはCURSORやLIMIT/OFFSET、あるいはKeyset Paginationを使用）
for ($chunk = 1; $chunk <= 3; $chunk++) {
// 重いDBクエリや外部APIリクエストのシミュレーション
usleep(500000); // 0.5秒のI/O待ちを想定
$data = [
'batch' => $batchId,
‘chunk’ => $chunk,
‘memory’ => memory_get_usage(true),
];
// 処理したチャンクデータをメインスケジューラへ返す
yield $data;
}
});
}
// 実行開始
$startTime = microtime(true);
$scheduler->run();
$endTime = microtime(true);
echo “Batch processing completed in ” . round($endTime – $startTime, 4) . ” seconds.\n”;
echo “Peak Memory Usage: ” . round(memory_get_peak_usage(true) / 1024 / 1024, 2) . ” MB\n”;
—
4. セキュリティ・ガバナンス：非同期環境における「オブジェクトインジェクション」の脅威
Fiberを用いた並列処理や、シリアライズされたタスクキューを扱うシステムにおいて、アーキテクトが絶対に見落してはならないのがPHPオブジェクトインジェクション（PHP Object Injection）の脅威だ。
分散バッチ処理では、ジョブのペイロードをキュー（Redisやデータベース等）に格納するためによく `serialize()` / `unserialize()` が使われる。もし、入力値検証が不十分な状態で外部からのデータを `unserialize()` した場合、悪意ある攻撃者は Gadget Chain（既存のクラス群のデストラクタやマジックメソッド `__wakeup()` / `__destruct()` を連鎖させる攻撃パターン）を構築し、RCE（リモートコード実行）に至る。
対策の鉄則
1. `unserialize()` の使用禁止: 代わりに JSON（`json_encode` / `json_decode`）を使用する。JSONであればオブジェクト構造ではなくただのデータ構造に強制変換されるため、マジックメソッドが勝手に発火する余地はない。
2. どうしてもシリアライズが必要な場合: `unserialize($data, [‘allowed_classes’ => [MyJobPayload::class]])` のように、許可されたクラスホワイトリストを厳格に指定する。
Zend VMのメモリ空間や関数ポインタの書き換え（UAF等）に至る脆弱性はPHPコアのアップデートで塞がれているが、アプリケーション層での不適切なデータハンドリングによる脆弱性は、アーキテクトの設計ミス以外の何ものでもない。
—
5. チーフアーキテクトからの提言
FiberはPHPをリアクティブかつ非同期な次世代言語へと押し上げる強力なプリミティブである。しかし、それは「魔法の弾丸」ではない。
- CPUバウンドな処理をFiberで並列化しても、PHPのシングルスレッド実行モデル（Zend VMの制約）の枠内であるため、真の並列計算（マルチコア活用）にはならない。CPUバウンドには `pcntl_fork` や外部ワーカー（RoadRunner, Swoole, あるいはGoによるマイクロサービス）を組み合わせるべきである。
- I/Oバウンドな処理（DBクエリの並列実行、HTTPリクエストの並行送信など）においては、Fiberはその真価を発揮し、メモリを極限まで節約しながらスループットを何倍にも跳ね上げる。
低レイヤのメモリ構造とZend VMの挙動を脳内に描くこと。それこそが、真にスケーラブルなWebシステムを構築する唯一の道である。