Node.jsでCloud Storageのファイルをストリーミングしてffmpegに渡したかったけどffmpegはFIFO非対応っぽい

所用で、Cloud Storageにある動画ファイルをopus(音声)形式にエンコードしたかった(Node.jsとffmpegで)。

動画ファイルがデカいのと、生成されるopusは小さいので、いちいちストレージに保存するのが手間だと考えた。そこで、間にmkfifoして作った名前付きパイプを挟んで効率化しようと思ったメモ。

ちなみにうまく動かなかったので識者の方がいらしたらコッソリ教えてください。

Cloud Storage

Cloud StorageというのはGoogleのサービス。要するにGoogle版のS3。これのNode.js版SDKにはfile.createReadStreamというメソッドがあり、バケットのファイルの中身をストリーミングしてReadable形式で流してくれる。

これを使うと、例えばfs.createWritableStreamなどにpipeさせることで、ファイルにストリーミング保存できる。デカいファイルを保存中に色々他の作業ができて高効率だ。

`mkfifo(1)`

mkfifoとは、名前付きパイプを作るUNIX系のコマンド。どのディストリビューションにも入っている基礎的なコマンドだ。

ja.wikipedia.org

mkfifo 名前すると、そこにパイプとして振る舞うファイルが生成され、リダイレクト先として使うことができる。

振る舞いはパイプそのもので、例えば書き込もうとしてもパイプを読み出してくれるプロセスがやってくるまではブロックするし、またその逆に、読み出そうとしてもそのパイプに書き込まれるまではブロックする。

% mkfifo p
% echo 'Hello, Pipe!' > p
# ここでブロックする。別のシェルで・・・
% cat p | tr '[:lower:]' '[:upper:]'
HELLO, PIPE!
# ここで両者のブロックが解除される

これをうまく使うと無駄な一時ファイルを作らずに済むので、かなり便利だ。

Node.jsでpipeに書き込みたかった

さて、Node.jsはシングルスレッドモデルを採用しているので、何も考えずにブロッキングな処理を行うと厄介なこと、つまりデッドロックが生じることになる。マルチスレッドではないので、asyncを付けていてもデッドロックは起こる。

それに比べるとffmpegをspawnする操作はブロックしないはずなので(というのも、プロセスが分かれるから)、先にffmpegを起動して、その後で名前付きパイプにストリーミング書き込みすればいいじゃ〜んという発想である。

結果、なんかどこかでデッドロックしてダウンロードが進まなくなり、止まってしまった。

import { Storage } from '@google-cloud/storage';
import * as fs from 'fs';
import { spawn, spawnSync } from 'node:child_process';

const prepareFifo = async () => {
  const mktemp = spawnSync("mktemp", ["-d"]);
  const tempDir = mktemp.stdout.toString().trim();
  console.log(`Created temp dir: ${tempDir}`);

  const fifoFile = `${tempDir}/fifo`;
  const mkfifo = spawnSync("mkfifo", [fifoFile]);

  return [fifoFile, tempDir];
}

const fetchVideo = async (cloudStorageUrl: URL, targetPath: string) => {
  const bucketName = cloudStorageUrl.host;
  const path = cloudStorageUrl.pathname.replace('/', ''); // 先頭の/を削る必要がある

  const bucket = storage.bucket(bucketName);
  const p = new Promise((resolve, reject) => {
    bucket.file(path).createReadStream().pipe( fs.createWriteStream(outputPath)).on('finish', resolve).on('error', reject);
  };

  return p;
};

const convert = async (inPath: string, outPath: string) => {
  // エラー処理などは省いて書いています
  return spawn(
    "ffmpeg", ["-i", inPath, "-vn", "-acodec", "libopus", "-b:a", "64k", "-y", outPath],
  );
};

const gsUri = new URL(...);
const [fifo, fifoDir] = await prepareFifo();

const command = await convert(fifo, 'example.opus');
const video = await fetchVideo(gsUri, fifo);

await Promise.all([command, video]);
// ...