Allow using temporary directory to speed up (#72)

Author: liuziba

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@techgreedy/singularity",
-  "version": "0.0.14",
+  "version": "0.0.15",
   "description": "Large scale client tool with capabilities on onboard PB-scale data to Filecoin network",
   "main": "dist/source/start.js",
   "types": "dist/source/start.d.ts",
@@ -67,7 +67,7 @@
     "@goodware/task-queue": "^2.1.3",
     "@gzuidhof/go-npm": "^0.1.13",
     "@rauschma/stringio": "^1.4.0",
-    "@techgreedy/generate-car": "^1.1.1",
+    "@techgreedy/generate-car": "1.2.0",
     "async-retry": "^1.3.3",
     "await-exec": "^0.1.2",
     "axios": "^0.26.1",

package.json

@@ -197,7 +197,9 @@ export default class Datastore {
       dataCid: Schema.Types.String,
       carSize: Schema.Types.Number,
       pieceCid: Schema.Types.String,
-      pieceSize: Schema.Types.Number
+      pieceSize: Schema.Types.Number,
+      filenameOverride: Schema.Types.String,
+      tmpDir: Schema.Types.String
     }, {
       timestamps: true
     });
@@ -219,7 +221,8 @@ export default class Datastore {
         index: 1
       },
       status: Schema.Types.String,
-      errorMessage: Schema.Types.String
+      errorMessage: Schema.Types.String,
+      tmpDir: Schema.Types.String
     }, {
       timestamps: true
     });

src/common/Datastore.ts

@@ -12,5 +12,6 @@ export default interface GenerationRequest {
   carSize?: number,
   pieceCid?: string,
   pieceSize?: number,
-  filenameOverride?: string // when the car name is different from cid
+  filenameOverride?: string, // when the car name is different from cid
+  tmpDir?: string
 }

src/common/model/GenerationRequest.ts

@@ -10,5 +10,6 @@ export default interface ScanningRequest {
   maxSize: number,
   workerId?: string,
   status: 'active' | 'completed' | 'error' | 'paused',
-  errorMessage?: string
+  errorMessage?: string,
+  tmpDir?: string
 }

src/common/model/ScanningRequest.ts

@@ -5,4 +5,5 @@ export default interface CreatePreparationRequest {
   outDir: string,
   minRatio?: number,
   maxRatio?: number,
+  tmpDir?: string
 }

src/deal-preparation/CreatePreparationRequest.ts

@@ -85,6 +85,18 @@ export default class DealPreparationService extends BaseService {
         this.logger.warn(`${dir} cannot be read during cleanup.`);
       }
     }
+    let tmpDirs = (await Datastore.ScanningRequestModel.find()).map(r => r.tmpDir);
+    tmpDirs = [...new Set(tmpDirs)];
+    for (const dir of tmpDirs) {
+      if (dir) {
+        try {
+          await fs.rm(dir, { recursive: true, force: true });
+          this.logger.info(`Removing temporary folder ${dir}`);
+        } catch (e) {
+          this.logger.warn(`${dir} cannot be read during cleanup.`);
+        }
+      }
+    }
   }
 
   private async cleanupHealthCheck (): Promise<void> {
@@ -385,7 +397,8 @@ export default class DealPreparationService extends BaseService {
       dealSize,
       outDir,
       minRatio,
-      maxRatio
+      maxRatio,
+      tmpDir
     } = <CreatePreparationRequest>request.body;
     this.logger.info(`Received request to start preparing dataset.`, { name, path, dealSize });
     const dealSizeNumber = xbytes.parseSize(dealSize);
@@ -420,6 +433,9 @@ export default class DealPreparationService extends BaseService {
     try {
       await fs.access(path, constants.F_OK);
       await fs.access(outDir, constants.F_OK);
+      if (tmpDir) {
+        await fs.access(tmpDir, constants.F_OK);
+      }
     } catch (_) {
       this.sendError(response, ErrorCode.PATH_NOT_ACCESSIBLE);
       return;
@@ -432,6 +448,7 @@ export default class DealPreparationService extends BaseService {
     scanningRequest.path = path;
     scanningRequest.status = 'active';
     scanningRequest.outDir = outDir;
+    scanningRequest.tmpDir = tmpDir;
     try {
       await scanningRequest.save();
     } catch (e: any) {
@@ -448,6 +465,7 @@ export default class DealPreparationService extends BaseService {
       maxSize,
       path,
       outDir,
+      tmpDir,
       status: scanningRequest.status
     }));
   }

src/deal-preparation/DealPreparationService.ts

@@ -63,6 +63,7 @@ export default class DealPreparationWorker extends BaseService {
         datasetName: request.name,
         path: request.path,
         outDir: request.outDir,
+        tmpDir: request.tmpDir,
         index,
         status: 'created'
       });
@@ -96,15 +97,27 @@ export default class DealPreparationWorker extends BaseService {
 
   private async generate (request: GenerationRequest, input: string): Promise<[stdout: string, stderr: string, statusCode: number | null]> {
     await fs.mkdir(request.outDir, { recursive: true });
-    this.logger.debug(`Spawning generate-car.`, { outPath: request.outDir, parentPath: request.path });
-    const [stdout, stderr, exitCode] = await DealPreparationWorker.invokeGenerateCar(input, request.outDir, request.path);
+    this.logger.debug(`Spawning generate-car.`, { outPath: request.outDir, parentPath: request.path, tmpDir: request.tmpDir });
+    let tmpDir: string | undefined;
+    if (request.tmpDir) {
+      tmpDir = path.join(request.tmpDir, randomUUID());
+    }
+    const [stdout, stderr, exitCode] = await DealPreparationWorker.invokeGenerateCar(input, request.outDir, request.path, tmpDir);
+    if (tmpDir) {
+      await fs.rm(tmpDir, { recursive: true, force: true });
+    }
     this.logger.debug(`Child process finished.`, { stdout, stderr, exitCode });
     return [stdout, stderr, exitCode];
   }
 
-  public static async invokeGenerateCar (input: string, outDir: string, p: string): Promise<[stdout: string, stderr: string, statusCode: number | null]> {
+  public static async invokeGenerateCar (input: string, outDir: string, p: string, tmpDir?: string)
+    : Promise<[stdout: string, stderr: string, statusCode: number | null]> {
     const cmd = await fs.pathExists(path.join(__dirname, 'generate-car')) ? path.join(__dirname, 'generate-car') : 'generate-car';
-    const child = spawn(cmd, ['-o', outDir, '-p', p], {
+    const args = ['-o', outDir, '-p', p];
+    if (tmpDir) {
+      args.push('-t', tmpDir);
+    }
+    const child = spawn(cmd, args, {
       stdio: ['pipe', 'pipe', 'pipe']
     });
     (async () => {

src/deal-preparation/DealPreparationWorker.ts

@@ -13,6 +13,7 @@ import { GeneratedFileList } from './common/model/OutputFileList';
 // eslint-disable-next-line @typescript-eslint/ban-ts-comment
 // @ts-ignore
 import TaskQueue from '@goodware/task-queue';
+import { randomUUID } from 'crypto';
 
 const version = packageJson.version;
 const program = new Command();
@@ -24,6 +25,7 @@ program.name('singularity-prepare')
   .argument('<outDir>', 'The output Directory to save CAR files and manifest files')
   .requiredOption('-l, --url-prefix <urlPrefix>', 'The prefix of the download link, which will be followed by datacid.car, i.e. http://download.mysite.org/')
   .option('-s, --deal-size <deal_size>', 'Target deal size, i.e. 32GiB', '32 GiB')
+  .option('-t, --tmp-dir <tmp_dir>', 'Temporary directory, may be useful when dataset source is slow, such as on S3 mount or NFS')
   .addOption(new Option('-m, --min-ratio <min_ratio>', 'Min ratio of deal to sector size, i.e. 0.55').default('0.55').argParser(parseFloat))
   .addOption(new Option('-M, --max-ratio <max_ratio>', 'Max ratio of deal to sector size, i.e. 0.95').default('0.95').argParser(parseFloat))
   .addOption(new Option('-j, --parallel <parallel>', 'How many generation jobs to run at the same time').default('1'))
@@ -76,7 +78,16 @@ program.name('singularity-prepare')
           End: file.end
         })));
 
-        const [stdout, stderr, exitCode] = await DealPreparationWorker.invokeGenerateCar(input, outDir, p);
+        let tmpDir: string | undefined;
+        if (options.tmpDir) {
+          tmpDir = path.join(options.tmpDir, randomUUID());
+        }
+        const [stdout, stderr, exitCode] = await DealPreparationWorker.invokeGenerateCar(input, outDir, p, tmpDir);
+
+        if (tmpDir) {
+          await fs.rm(tmpDir, { recursive: true, force: true });
+        }
+
         if (exitCode !== 0) {
           console.error(stderr);
         }

src/singularity-prepare.ts

@@ -165,6 +165,7 @@ preparation.command('create').description('Start deal preparation for a local da
   .argument('<datasetPath>', 'Directory path to the dataset')
   .argument('<outDir>', 'The output Directory to save CAR files')
   .option('-s, --deal-size <deal_size>', 'Target deal size, i.e. 32GiB', '32 GiB')
+  .option('-t, --tmp-dir <tmp_dir>', 'Temporary directory, may be useful when dataset source is slow, such as on S3 mount or NFS')
   .addOption(new Option('-m, --min-ratio <min_ratio>', 'Min ratio of deal to sector size, i.e. 0.55').argParser(parseFloat))
   .addOption(new Option('-M, --max-ratio <max_ratio>', 'Max ratio of deal to sector size, i.e. 0.95').argParser(parseFloat))
   .action(async (name, p, outDir, options) => {
@@ -182,7 +183,8 @@ preparation.command('create').description('Start deal preparation for a local da
         dealSize: dealSize,
         outDir: path.resolve(outDir),
         minRatio: options.minRatio,
-        maxRatio: options.maxRatio
+        maxRatio: options.maxRatio,
+        tmpDir: options.tmpDir
       });
     } catch (error) {
       CliUtil.renderErrorAndExit(error);