Batch stat and readdir calls

Author: code-asher

packages/protocol/src/browser/modules/fs.ts

@@ -1,12 +1,32 @@
 import * as fs from "fs";
 import { callbackify } from "util";
-import { ClientProxy } from "../../common/proxy";
+import { ClientProxy, Batch } from "../../common/proxy";
 import { IEncodingOptions, IEncodingOptionsCallback } from "../../common/util";
 import { FsModuleProxy, Stats as IStats, WatcherProxy, WriteStreamProxy } from "../../node/modules/fs";
 import { Writable  } from "./stream";
 
 // tslint:disable no-any
 
+class StatBatch extends Batch<IStats, { path: fs.PathLike }> {
+	public constructor(private readonly proxy: FsModuleProxy) {
+		super();
+	}
+
+	protected remoteCall(batch: { path: fs.PathLike }[]): Promise<(IStats | Error)[]> {
+		return this.proxy.statBatch(batch);
+	}
+}
+
+class ReaddirBatch extends Batch<Buffer[] | fs.Dirent[] | string[], { path: fs.PathLike, options: IEncodingOptions }> {
+	public constructor(private readonly proxy: FsModuleProxy) {
+		super();
+	}
+
+	protected remoteCall(queue: { path: fs.PathLike, options: IEncodingOptions }[]): Promise<(Buffer[] | fs.Dirent[] | string[] | Error)[]> {
+		return this.proxy.readdirBatch(queue);
+	}
+}
+
 class Watcher extends ClientProxy<WatcherProxy> implements fs.FSWatcher {
 	public close(): void {
 		this.proxy.close();
@@ -28,7 +48,13 @@ class WriteStream extends Writable<WriteStreamProxy> implements fs.WriteStream {
 }
 
 export class FsModule {
-	public constructor(private readonly proxy: FsModuleProxy) {}
+	private readonly statBatch: StatBatch;
+	private readonly readdirBatch: ReaddirBatch;
+
+	public constructor(private readonly proxy: FsModuleProxy) {
+		this.statBatch = new StatBatch(this.proxy);
+		this.readdirBatch = new ReaddirBatch(this.proxy);
+	}
 
 	public access = (path: fs.PathLike, mode: number | undefined | ((err: NodeJS.ErrnoException) => void), callback?: (err: NodeJS.ErrnoException) => void): void => {
 		if (typeof mode === "function") {
@@ -175,7 +201,7 @@ export class FsModule {
 			callback = options;
 			options = undefined;
 		}
-		callbackify(this.proxy.readdir)(path, options, callback!);
+		callbackify(this.readdirBatch.add)({ path, options }, callback!);
 	}
 
 	public readlink = (path: fs.PathLike, options: IEncodingOptionsCallback, callback?: (err: NodeJS.ErrnoException, linkString: string | Buffer) => void): void => {
@@ -203,7 +229,7 @@ export class FsModule {
 	}
 
 	public stat = (path: fs.PathLike, callback: (err: NodeJS.ErrnoException, stats: fs.Stats) => void): void => {
-		callbackify(this.proxy.stat)(path, (error, stats) => {
+		callbackify(this.statBatch.add)({ path }, (error, stats) => {
 			callback(error, stats && new Stats(stats));
 		});
 	}

packages/protocol/src/common/proxy.ts

@@ -81,3 +81,66 @@ export enum Module {
 	NodePty = "node-pty",
 	Trash = "trash",
 }
+
+interface BatchItem<T, A> {
+	args: A;
+	resolve: (t: T) => void;
+	reject: (e: Error) => void;
+}
+
+/**
+ * Batch remote calls.
+ */
+export abstract class Batch<T, A> {
+	/**
+	 * Flush after reaching this count.
+	 */
+	private maxCount = 100;
+
+	/**
+	 * Flush after not receiving more requests for this amount of time.
+	 */
+	private maxTime = 100;
+
+	private flushTimeout: number | NodeJS.Timer | undefined;
+
+	private batch = <BatchItem<T, A>[]>[];
+
+	public add = (args: A): Promise<T> => {
+		return new Promise((resolve, reject) => {
+			this.batch.push({
+				args,
+				resolve,
+				reject,
+			});
+			if (this.batch.length >= this.maxCount) {
+				this.flush();
+			} else {
+				clearTimeout(this.flushTimeout as any);
+				this.flushTimeout = setTimeout(this.flush, this.maxTime);
+			}
+		});
+	}
+
+	protected abstract remoteCall(batch: A[]): Promise<(T | Error)[]>;
+
+	private flush = (): void => {
+		clearTimeout(this.flushTimeout as any);
+
+		const batch = this.batch;
+		this.batch = [];
+
+		this.remoteCall(batch.map((q) => q.args)).then((results) => {
+			batch.forEach((item, i) => {
+				const result = results[i];
+				if (!result) {
+					item.reject(new Error("no response"));
+				} else if (result instanceof Error) {
+					item.reject(result);
+				} else {
+					item.resolve(result);
+				}
+			});
+		}).catch((error) => batch.forEach((item) => item.reject(error)));
+	}
+}

packages/protocol/src/node/modules/fs.ts

@@ -182,6 +182,10 @@ export class FsModuleProxy {
 		return promisify(fs.readdir)(path, options);
 	}
 
+	public readdirBatch(args: { path: fs.PathLike, options: IEncodingOptions }[]): Promise<(Buffer[] | fs.Dirent[] | string[] | Error)[]> {
+		return Promise.all(args.map((args) => this.readdir(args.path, args.options).catch((e) => e)));
+	}
+
 	public readlink(path: fs.PathLike, options: IEncodingOptions): Promise<string | Buffer> {
 		return promisify(fs.readlink)(path, options);
 	}
@@ -202,6 +206,10 @@ export class FsModuleProxy {
 		return this.makeStatsSerializable(await promisify(fs.stat)(path));
 	}
 
+	public async statBatch(args: { path: fs.PathLike }[]): Promise<(Stats | Error)[]> {
+		return Promise.all(args.map((args) => this.stat(args.path).catch((e) => e)));
+	}
+
 	public symlink(target: fs.PathLike, path: fs.PathLike, type?: fs.symlink.Type | null): Promise<void> {
 		return promisify(fs.symlink)(target, path, type);
 	}