chore(screencast): respect i/o backpressure when writing into ffmpeg (#4164)

Author: pavelfeldman

src/server/chromium/videoRecorder.ts

@@ -16,7 +16,7 @@
 
 import { ChildProcess } from 'child_process';
 import { ffmpegExecutable } from '../../utils/binaryPaths';
-import { assert } from '../../utils/utils';
+import { assert, monotonicTime } from '../../utils/utils';
 import { launchProcess } from '../processLauncher';
 import { Progress, ProgressController } from '../progress';
 import * as types from '../types';
@@ -26,11 +26,13 @@ const fps = 25;
 export class VideoRecorder {
   private _process: ChildProcess | null = null;
   private _gracefullyClose: (() => Promise<void>) | null = null;
-  private _lastWritePromise: Promise<void> | undefined;
+  private _lastWritePromise: Promise<void> = Promise.resolve();
   private _lastFrameTimestamp: number = 0;
   private _lastFrameBuffer: Buffer | null = null;
   private _lastWriteTimestamp: number = 0;
   private readonly _progress: Progress;
+  private _frameQueue: Buffer[] = [];
+  private _isStopped = false;
 
   static async launch(options: types.PageScreencastOptions): Promise<VideoRecorder> {
     if (!options.outputFile.endsWith('.webm'))
@@ -50,7 +52,6 @@ export class VideoRecorder {
   }
 
   private async _launch(options: types.PageScreencastOptions) {
-    assert(!this._isRunning());
     const w = options.width;
     const h = options.height;
     const args = `-loglevel error -f image2pipe -c:v mjpeg -i - -y -an -r ${fps} -c:v vp8 -qmin 0 -qmax 50 -crf 8 -b:v 1M -vf pad=${w}:${h}:0:0:gray,crop=${w}:${h}:0:0`.split(' ');
@@ -84,58 +85,43 @@ export class VideoRecorder {
     this._gracefullyClose = gracefullyClose;
   }
 
-  async writeFrame(frame: Buffer, timestamp: number) {
+  writeFrame(frame: Buffer, timestamp: number) {
     assert(this._process);
-    if (!this._isRunning())
+    if (this._isStopped)
       return;
     this._progress.log(`writing frame ` + timestamp);
-    if (this._lastFrameBuffer)
-      this._lastWritePromise = this._flushLastFrame(timestamp - this._lastFrameTimestamp).catch(e => this._progress.log('Error while writing frame: ' + e));
+
+    if (this._lastFrameBuffer) {
+      const durationSec = timestamp - this._lastFrameTimestamp;
+      const repeatCount = Math.max(1, Math.round(fps * durationSec));
+      for (let i = 0; i < repeatCount; ++i)
+        this._frameQueue.push(this._lastFrameBuffer);
+      this._lastWritePromise = this._lastWritePromise.then(() => this._sendFrames());
+    }
+
     this._lastFrameBuffer = frame;
     this._lastFrameTimestamp = timestamp;
-    this._lastWriteTimestamp = Date.now();
+    this._lastWriteTimestamp = monotonicTime();
   }
 
-  private async _flushLastFrame(durationSec: number): Promise<void> {
-    assert(this._process);
-    const frame = this._lastFrameBuffer;
-    if (!frame)
-      return;
-    const previousWrites = this._lastWritePromise;
-    let finishedWriting: () => void;
-    const writePromise = new Promise<void>(fulfill => finishedWriting = fulfill);
-    const repeatCount = Math.max(1, Math.round(fps * durationSec));
-    this._progress.log(`flushing ${repeatCount} frame(s)`);
-    await previousWrites;
-    for (let i = 0; i < repeatCount; i++) {
-      const callFinish = i === (repeatCount - 1);
-      this._process.stdin.write(frame, (error: Error | null | undefined) => {
-        if (error)
-          this._progress.log(`ffmpeg failed to write: ${error}`);
-        if (callFinish)
-          finishedWriting();
-      });
-    }
-    return writePromise;
+  private async _sendFrames() {
+    while (this._frameQueue.length)
+      await this._sendFrame(this._frameQueue.shift()!);
+  }
+
+  private async _sendFrame(frame: Buffer) {
+    return new Promise(f => this._process!.stdin.write(frame, f)).then(error => {
+      if (error)
+        this._progress.log(`ffmpeg failed to write: ${error}`);
+    });
   }
 
   async stop() {
-    if (!this._gracefullyClose)
+    if (this._isStopped)
       return;
-
-    if (this._lastWriteTimestamp) {
-      const durationSec = (Date.now() - this._lastWriteTimestamp) / 1000;
-      if (!this._lastWritePromise || durationSec > 1 / fps)
-        this._flushLastFrame(durationSec).catch(e => this._progress.log('Error while writing frame: ' + e));
-    }
-
-    const close = this._gracefullyClose;
-    this._gracefullyClose = null;
+    this.writeFrame(Buffer.from([]), this._lastFrameTimestamp + (monotonicTime() - this._lastWriteTimestamp) / 1000);
+    this._isStopped = true;
     await this._lastWritePromise;
-    await close();
-  }
-
-  private _isRunning(): boolean {
-    return !!this._gracefullyClose;
+    await this._gracefullyClose!();
   }
 }