tools and improves

Signed-off-by: Guy Margalit <[email protected]>

Author: guymguym

.eslintrc.js

@@ -138,7 +138,7 @@ module.exports = {
         'one-var': ['error', 'never'],
 
         '@stylistic/js/space-before-function-paren': ['error', {
-            'anonymous': 'never',
+            'anonymous': 'ignore',
             'named': 'never',
             'asyncArrow': 'always'
         }],

config.js

@@ -1173,7 +1173,7 @@ function _get_data_from_file(file_name) {
     try {
         data = fs.readFileSync(file_name).toString();
     } catch (e) {
-        console.warn(`Error accrued while getting the data from ${file_name}: ${e}`);
+        // console.log(`Error accrued while getting the data from ${file_name}: ${e}`);
         return;
     }
     return data;
@@ -1189,7 +1189,7 @@ function _get_config_root() {
         const data = _get_data_from_file(redirect_path);
         config_root = data.toString().trim();
     } catch (err) {
-        console.warn('config.get_config_root - could not find custom config_root, will use the default config_root ', config_root);
+        // console.log('config.get_config_root - could not find custom config_root, will use the default config_root ', config_root);
     }
     return config_root;
 }
@@ -1244,7 +1244,7 @@ function load_nsfs_nc_config() {
     try {
         if (!config.NSFS_NC_CONF_DIR) {
             config.NSFS_NC_CONF_DIR = _get_config_root();
-            console.warn('load_nsfs_nc_config.setting config.NSFS_NC_CONF_DIR', config.NSFS_NC_CONF_DIR);
+            // console.warn('load_nsfs_nc_config.setting config.NSFS_NC_CONF_DIR', config.NSFS_NC_CONF_DIR);
         }
         const config_path = path.join(config.NSFS_NC_CONF_DIR, 'config.json');
         const config_data = require(config_path);

package.json

@@ -16,7 +16,7 @@
   "scripts": {
     "install": "echo install hook override to avoid npm default hook calling node-gyp",
     "build": "npm run build:native",
-    "build:native": "node-gyp configure && node-gyp build",
+    "build:native": "node-gyp configure build",
     "rebuild": "npm run clean:install && npm run clean:build && npm install && npm run build",
     "pkg": "pkg . --public --options unhandled-rejections=warn --compress Brotli",
     "---": "------------------------------------------------------------------",

src/agent/block_store_speed.js

@@ -3,7 +3,6 @@
 
 // const _ = require('lodash');
 const argv = require('minimist')(process.argv);
-const cluster = require('cluster');
 const mongodb = require('mongodb');
 
 const api = require('../api');
@@ -26,17 +25,10 @@ argv.timeout = argv.timeout || 60000;
 
 let block_index = 0;
 
-const master_speedometer = new Speedometer('Total Speed');
 const speedometer = new Speedometer('Block Store Speed');
-
-if (argv.forks > 1 && cluster.isMaster) {
-    master_speedometer.fork(argv.forks);
-} else {
-    main();
-}
+speedometer.run_workers(argv.forks, main, argv);
 
 async function main() {
-    console.log('ARGS', argv);
     const rpc = api.new_rpc();
     const client = rpc.new_client();
     const signal_client = rpc.new_client();

src/sdk/object_sdk.js

@@ -142,25 +142,25 @@ class ObjectSDK {
             this.abort_controller.abort(err);
         });
 
-        // TODO: aborted event is being deprecated since nodejs 16
-        // https://nodejs.org/dist/latest-v16.x/docs/api/http.html#event-aborted recommends on listening to close event
-        // req.once('close', () => {
-        //     dbg.log0('request aborted1', req.url);
-
-        //     if (req.destroyed) {
-        //         dbg.log0('request aborted', req.url);
-        //         this.abort_controller.abort(new Error('request aborted ' + req.url));
-        //     }
+        // Note: aborted event is deprecated in favor of the close event
+        // https://nodejs.org/dist/latest-v16.x/docs/api/http.html#event-aborted
+        // req.once('aborted', () => {
+        //     dbg.log0('request aborted', req.url);
+        //     this.abort_controller.abort(new Error('request aborted ' + req.url));
         // });
-
-        req.once('aborted', () => {
-            dbg.log0('request aborted', req.url);
-            this.abort_controller.abort(new Error('request aborted ' + req.url));
+        req.once('close', () => {
+            // dbg.log1('request closed', req.url);
+            if (req.errored) {
+                dbg.log0('request aborted', req.url);
+                this.abort_controller.abort(new Error('request aborted ' + req.url));
+            }
         });
+
     }
 
     throw_if_aborted() {
-        if (this.abort_controller.signal.aborted) throw new Error('request aborted signal');
+        this.abort_controller.signal.throwIfAborted();
+        // if (this.abort_controller.signal.aborted) throw new Error('request aborted signal');
     }
 
     add_abort_handler(handler) {

src/tools/coding_speed.js

@@ -7,15 +7,13 @@ const _ = require('lodash');
 const argv = require('minimist')(process.argv);
 const stream = require('stream');
 const assert = require('assert');
-const cluster = require('cluster');
 const crypto = require('crypto');
 
 const config = require('../../config');
 const ChunkCoder = require('../util/chunk_coder');
 const RandStream = require('../util/rand_stream');
 const Speedometer = require('../util/speedometer');
 const ChunkEraser = require('../util/chunk_eraser');
-const stream_utils = require('../util/stream_utils');
 const ChunkSplitter = require('../util/chunk_splitter');
 const FlattenStream = require('../util/flatten_stream');
 // const CoalesceStream = require('../util/coalesce_stream');
@@ -35,17 +33,10 @@ argv.verbose = Boolean(argv.verbose); // default is false
 argv.sse_c = Boolean(argv.sse_c); // default is false
 delete argv._;
 
-const master_speedometer = new Speedometer('Total Speed');
 const speedometer = new Speedometer('Chunk Coder Speed');
-
-if (argv.forks > 1 && cluster.isMaster) {
-    master_speedometer.fork(argv.forks);
-} else {
-    main();
-}
+speedometer.run_workers(argv.forks, main, argv);
 
 function main() {
-    console.log('Arguments:', JSON.stringify(argv, null, 2));
 
     const chunk_split_config = {
         avg_chunk: config.CHUNK_SPLIT_AVG_CHUNK,
@@ -106,11 +97,10 @@ function main() {
 
     let total_size = 0;
     let num_parts = 0;
-    const reporter = new stream.Transform({
+    const reporter = new stream.Writable({
         objectMode: true,
-        allowHalfOpen: false,
         highWaterMark: 50,
-        transform(chunk, encoding, callback) {
+        write(chunk, encoding, callback) {
             if (argv.verbose) console.log({ ...chunk, data: 'ommitted' });
             if (argv.compare && chunk.original_data) {
                 assert(Buffer.concat(chunk.original_data).equals(chunk.data));
@@ -122,7 +112,7 @@ function main() {
         }
     });
 
-    /** @type {stream.Stream[]} */
+    /** @type {(stream.Readable | stream.Transform | stream.Writable)[]} */
     const transforms = [
         input,
         splitter,
@@ -137,7 +127,7 @@ function main() {
         transforms.push(new FlattenStream());
     }
     transforms.push(reporter);
-    return stream_utils.pipeline(transforms)
+    return stream.promises.pipeline(transforms)
         .then(() => {
             console.log('AVERAGE CHUNK SIZE', (total_size / num_parts).toFixed(0));
             if (splitter.md5) {

src/tools/cpu_speed.js

@@ -3,7 +3,6 @@
 
 require('../util/fips');
 const crypto = require('crypto');
-const cluster = require('cluster');
 const argv = require('minimist')(process.argv);
 const Speedometer = require('../util/speedometer');
 
@@ -13,17 +12,12 @@ argv.forks = argv.forks || 1;
 argv.size = argv.size || (10 * 1024);
 argv.hash = argv.hash || 'sha256';
 
-if (argv.forks > 1 && cluster.isMaster) {
-    const master_speedometer = new Speedometer('Total Speed');
-    master_speedometer.fork(argv.forks);
-} else {
-    main();
-}
+const speedometer = new Speedometer(`CPU(${argv.hash})`);
+speedometer.run_workers(argv.forks, main, argv);
 
 function main() {
     const hasher = crypto.createHash(argv.hash);
     const buf = crypto.randomBytes(1024 * 1024);
-    const speedometer = new Speedometer('CPU Speed');
     let size = argv.size * 1024 * 1024;
     console.log(`Crunching ${argv.size} MB with ${argv.hash}...`);
     run();

src/tools/fs_speed.js

@@ -1,11 +1,13 @@
 /* Copyright (C) 2016 NooBaa */
 'use strict';
 
+require('../util/panic');
+
 const fs = require('fs');
 const util = require('util');
 const path = require('path');
 const argv = require('minimist')(process.argv);
-const cluster = require('cluster');
+const { cluster } = require('../util/fork_utils');
 const execAsync = util.promisify(require('child_process').exec);
 const Speedometer = require('../util/speedometer');
 const RandStream = require('../util/rand_stream');
@@ -55,6 +57,7 @@ argv.file_size_units = argv.file_size_units || 'MB';
 argv.block_size_units = argv.block_size_units || 'MB';
 argv.fsync = Boolean(argv.fsync);
 argv.mode = argv.mode || 'nsfs';
+argv.backend = argv.backend || 'GPFS';
 if (argv.mode === 'dd') {
     argv.device = argv.device || '/dev/zero';
 } else {
@@ -89,18 +92,15 @@ const file_size = argv.file_size * size_units_table[argv.file_size_units];
 const block_count = Math.ceil(file_size / block_size);
 const file_size_aligned = block_count * block_size;
 const nb_native = argv.mode === 'nsfs' && require('../util/nb_native');
-const is_master = cluster.isMaster;
-const speedometer = new Speedometer(is_master ? 'Total Speed' : 'FS Speed');
+const is_master = cluster.isPrimary;
 const start_time = Date.now();
 const end_time = start_time + (argv.time * 1000);
 
-if (argv.forks > 1 && is_master) {
-    speedometer.fork(argv.forks);
-} else {
-    main();
-}
+const speedometer = new Speedometer('FS Speed');
+speedometer.run_workers(argv.forks, main, argv);
 
 async function main() {
+    // nb_native().fs.set_debug_level(5);
     const promises = [];
     fs.mkdirSync(argv.dir, { recursive: true });
     for (let i = 0; i < argv.concur; ++i) promises.push(worker(i));
@@ -127,15 +127,23 @@ async function worker(id) {
         if (file_start_time >= end_time) break;
         const file_path = path.join(dir, `file-${file_id}`);
         file_id += 1;
-        if (argv.mode === 'nsfs') {
-            await work_with_nsfs(file_path);
-        } else if (argv.mode === 'nodejs') {
-            await work_with_nodejs(file_path);
-        } else if (argv.mode === 'dd') {
-            await work_with_dd(file_path);
+        try {
+            if (argv.mode === 'nsfs') {
+                await work_with_nsfs(file_path);
+            } else if (argv.mode === 'nodejs') {
+                await work_with_nodejs(file_path);
+            } else if (argv.mode === 'dd') {
+                await work_with_dd(file_path);
+            }
+            const took_ms = Date.now() - file_start_time;
+            speedometer.add_op(took_ms);
+        } catch (err) {
+            if (argv.read && err.code === 'ENOENT') {
+                file_id = 0;
+            } else {
+                throw err;
+            }
         }
-        const took_ms = Date.now() - file_start_time;
-        speedometer.add_op(took_ms);
     }
 }
 
@@ -157,24 +165,25 @@ async function work_with_nsfs(file_path) {
     const fs_context = {
         // uid: 666,
         // gid: 666,
-        backend: 'GPFS',
-        warn_threshold_ms: 1000,
+        backend: argv.backend,
+        warn_threshold_ms: 10000,
     };
-    const file = await nb_native().fs.open(fs_context, file_path, argv.read ? 'r' : 'w', 0x660);
+    const file = await nb_native().fs.open(fs_context, file_path, argv.read ? 'r' : 'w', 0o660);
     for (let pos = 0; pos < file_size_aligned; pos += block_size) {
         const buf_start_time = Date.now();
         if (buf_start_time >= end_time) break;
         const buf = rand_stream.generator(block_size);
         if (argv.nvec > 1) {
             if (argv.read) {
-                await file.readv(fs_context, split_to_nvec(buf, argv.nvec));
+                // await file.readv(fs_context, split_to_nvec(buf, argv.nvec));
+                throw new Error('TODO: readv is not yet available in NativeFile');
             } else {
                 await file.writev(fs_context, split_to_nvec(buf, argv.nvec));
             }
         } else if (argv.read) {
             await file.read(fs_context, buf, 0, buf.length, pos);
         } else {
-            await file.write(fs_context, buf);
+            await file.write(fs_context, buf, buf.length, pos);
         }
         speedometer.update(block_size);
     }
@@ -187,7 +196,7 @@ async function work_with_nodejs(file_path) {
         highWaterMark: 2 * block_size,
         generator: argv.read ? 'noinit' : argv.generator,
     });
-    const file = await fs.promises.open(file_path, argv.read ? 'r' : 'w', 0x660);
+    const file = await fs.promises.open(file_path, argv.read ? 'r' : 'w', 0o660);
     for (let pos = 0; pos < file_size_aligned; pos += block_size) {
         const buf_start_time = Date.now();
         if (buf_start_time >= end_time) break;

src/tools/rand_speed.js

@@ -2,25 +2,21 @@
 'use strict';
 
 const zlib = require('zlib');
-const cluster = require('cluster');
 const RandStream = require('../util/rand_stream');
 const Speedometer = require('../util/speedometer');
 const argv = require('minimist')(process.argv);
 
-argv.forks = argv.forks || 1;
+argv.forks ||= 1;
+argv.buf ||= 1024 * 1024;
+argv.generator ||= 'crypto'; // see RandStream for options
 
-if (argv.forks > 1 && cluster.isMaster) {
-    const master_speedometer = new Speedometer('Total Speed');
-    master_speedometer.fork(argv.forks);
-} else {
-    main();
-}
+const speedometer = new Speedometer('RandStream');
+speedometer.run_workers(argv.forks, main, argv);
 
 function main() {
-    const speedometer = new Speedometer('Rand Speed');
     const len = (argv.len * 1024 * 1024) || Infinity;
     const input = new RandStream(len, {
-        highWaterMark: 1024 * 1024,
+        highWaterMark: argv.buf,
         generator: argv.generator,
     });
     input.on('data', data => speedometer.update(data.length));