NSFS | Replace ChunkFS with FileWriter

Signed-off-by: Romy <[email protected]>

Author: romayalon

config.js

@@ -882,6 +882,8 @@ config.NSFS_LOW_FREE_SPACE_PERCENT_UNLEASH = 0.10;
 // anonymous account name
 config.ANONYMOUS_ACCOUNT_NAME = 'anonymous';
 
+config.NSFS_UPLOAD_STREAM_MAX_BUFFER_SIZE = 64 * 1024 * 1024;
+
 ////////////////////////////
 // NSFS NON CONTAINERIZED //
 ////////////////////////////

src/sdk/namespace_fs.js

@@ -19,7 +19,7 @@ const stream_utils = require('../util/stream_utils');
 const buffer_utils = require('../util/buffer_utils');
 const size_utils = require('../util/size_utils');
 const native_fs_utils = require('../util/native_fs_utils');
-const ChunkFS = require('../util/chunk_fs');
+const FileWriter = require('../util/file_writer');
 const LRUCache = require('../util/lru_cache');
 const nb_native = require('../util/nb_native');
 const RpcError = require('../rpc/rpc_error');
@@ -1564,36 +1564,46 @@ class NamespaceFS {
     // Can be finetuned further on if needed and inserting the Semaphore logic inside
     // Instead of wrapping the whole _upload_stream function (q_buffers lives outside of the data scope of the stream)
     async _upload_stream({ fs_context, params, target_file, object_sdk, offset }) {
-        const { source_stream, copy_source } = params;
+        const { copy_source } = params;
         try {
-            // Not using async iterators with ReadableStreams due to unsettled promises issues on abort/destroy
-            const md5_enabled = this._is_force_md5_enabled(object_sdk);
-            const chunk_fs = new ChunkFS({
+            const md5_enabled =
+                config.NSFS_CALCULATE_MD5 ||
+                this.force_md5_etag ||
+                object_sdk?.requesting_account?.force_md5_etag;
+            const file_writer = new FileWriter({
                 target_file,
                 fs_context,
-                stats: this.stats,
-                namespace_resource_id: this.namespace_resource_id,
-                md5_enabled,
                 offset,
+                md5_enabled,
+                stats: this.stats,
                 bucket: params.bucket,
-                large_buf_size: multi_buffer_pool.get_buffers_pool(undefined).buf_size
+                large_buf_size: multi_buffer_pool.get_buffers_pool(undefined).buf_size,
+                namespace_resource_id: this.namespace_resource_id,
             });
-            chunk_fs.on('error', err1 => dbg.error('namespace_fs._upload_stream: error occured on stream ChunkFS: ', err1));
-            chunk_fs.on('finish', arg => dbg.error('namespace_fs._upload_stream: finish occured on stream ChunkFS: ', arg));
-            chunk_fs.on('close', arg => dbg.error('namespace_fs._upload_stream: close occured on stream ChunkFS: ', arg));
+            file_writer.on('error', err => dbg.error('namespace_fs._upload_stream: error occured on FileWriter: ', err));
+            file_writer.on('finish', arg => dbg.error('namespace_fs._upload_stream: finish occured on stream FileWriter: ', arg));
+            file_writer.on('close', arg => dbg.error('namespace_fs._upload_stream: close occured on stream FileWriter: ', arg));
+
+            const parser = params.source_stream?.socket?.parser;
+            if (parser) {
+                const { HTTPParser } = process.binding('http_parser');
+                parser[HTTPParser.kOnStreamAlloc] = length => {
+                    const bp = multi_buffer_pool.get_buffers_pool(length);
+                    const { buffer, callback } = bp.get_buffer_nowait();
+                    // @ts-ignore
+                    buffer.buffer_pool_release_callback = callback;
+                    return buffer;
+                };
+            }
             if (copy_source) {
-                // ChunkFS is a Transform stream, however read_object_stream expects a write stream. call resume to close the read part
-                // we need to close both read and write parts for Transform stream to properly close and release resorces
-                chunk_fs.resume();
-                await this.read_object_stream(copy_source, object_sdk, chunk_fs);
+                await this.read_object_stream(copy_source, object_sdk, file_writer);
             } else if (params.source_params) {
-                chunk_fs.resume();
-                await params.source_ns.read_object_stream(params.source_params, object_sdk, chunk_fs);
+                await params.source_ns.read_object_stream(params.source_params, object_sdk, file_writer);
             } else {
-                await stream_utils.pipeline([source_stream, chunk_fs]);
-                await stream_utils.wait_finished(chunk_fs);
+                await stream_utils.pipeline([params.source_stream, file_writer]);
+                await stream_utils.wait_finished(file_writer);
             }
-            return { digest: chunk_fs.digest, total_bytes: chunk_fs.total_bytes };
+            return { digest: file_writer.digest, total_bytes: file_writer.total_bytes };
         } catch (error) {
             dbg.error('_upload_stream had error: ', error);
             throw error;

src/test/unit_tests/index.js

@@ -57,7 +57,7 @@ require('./test_bucket_chunks_builder');
 require('./test_mirror_writer');
 require('./test_namespace_fs');
 require('./test_ns_list_objects');
-require('./test_chunk_fs');
+require('./test_file_writer');
 require('./test_namespace_fs_mpu');
 require('./test_nb_native_fs');
 require('./test_s3select');

src/test/unit_tests/nc_index.js

@@ -7,7 +7,7 @@ coretest.setup();
 
 require('./test_namespace_fs');
 require('./test_ns_list_objects');
-require('./test_chunk_fs');
+require('./test_file_writer');
 require('./test_namespace_fs_mpu');
 require('./test_nb_native_fs');
 require('./test_nc_nsfs_cli');

src/test/unit_tests/test_chunk_fs.js renamed to src/test/unit_tests/test_file_writer.js

@@ -3,24 +3,24 @@
 'use strict';
 
 const mocha = require('mocha');
-const chunk_fs_hashing = require('../../tools/chunk_fs_hashing');
+const file_writer_hashing = require('../../tools/file_writer_hashing');
 
-mocha.describe('ChunkFS', function() {
+mocha.describe('FileWriter', function() {
     const RUN_TIMEOUT = 10 * 60 * 1000;
 
-    mocha.it('Concurrent ChunkFS with hash target', async function() {
+    mocha.it('Concurrent FileWriter with hash target', async function() {
         const self = this;
         self.timeout(RUN_TIMEOUT);
-        await chunk_fs_hashing.hash_target();
+        await file_writer_hashing.hash_target();
     });
 
-    mocha.it('Concurrent ChunkFS with file target', async function() {
+    mocha.it('Concurrent FileWriter with file target', async function() {
         const self = this;
         self.timeout(RUN_TIMEOUT);
-        await chunk_fs_hashing.file_target();
+        await file_writer_hashing.file_target();
     });
 
-    mocha.it('Concurrent ChunkFS with file target - produce num_chunks > 1024 && total_chunks_size < config.NSFS_BUF_SIZE_L', async function() {
+    mocha.it('Concurrent FileWriter with file target - produce num_chunks > 1024 && total_chunks_size < config.NSFS_BUF_SIZE_L', async function() {
         const self = this;
         self.timeout(RUN_TIMEOUT);
         // The goal of this test is to produce num_chunks > 1024 && total_chunks_size < config.NSFS_BUF_SIZE_L
@@ -29,6 +29,6 @@ mocha.describe('ChunkFS', function() {
         // chunk size = 100, total_chunks_size after having 1024 chunks is = 100 * 1024 < config.NSFS_BUF_SIZE_L
         const chunk_size = 100;
         const parts_s = 50;
-        await chunk_fs_hashing.file_target(chunk_size, parts_s);
+        await file_writer_hashing.file_target(chunk_size, parts_s);
     });
 });

src/tools/chunk_fs_hashing.js renamed to src/tools/file_writer_hashing.js

@@ -3,7 +3,7 @@
 
 const crypto = require('crypto');
 const assert = require('assert');
-const ChunkFS = require('../util/chunk_fs');
+const FileWriter = require('../util/file_writer');
 const config = require('../../config');
 const nb_native = require('../util/nb_native');
 const stream_utils = require('../util/stream_utils');
@@ -19,7 +19,7 @@ const PARTS = Number(argv.parts) || 1000;
 const CONCURRENCY = Number(argv.concurrency) || 20;
 const CHUNK = Number(argv.chunk) || 16 * 1024;
 const PART_SIZE = Number(argv.part_size) || 20 * 1024 * 1024;
-const F_PREFIX = argv.dst_folder || '/tmp/chunk_fs_hashing/';
+const F_PREFIX = argv.dst_folder || '/tmp/file_writer_hashing/';
 
 const DEFAULT_FS_CONFIG = {
     uid: Number(argv.uid) || process.getuid(),
@@ -28,12 +28,6 @@ const DEFAULT_FS_CONFIG = {
     warn_threshold_ms: 100,
 };
 
-const DUMMY_RPC = {
-    object: {
-        update_endpoint_stats: (...params) => null
-    }
-};
-
 const XATTR_USER_PREFIX = 'user.';
 // TODO: In order to verify validity add content_md5_mtime as well
 const XATTR_MD5_KEY = XATTR_USER_PREFIX + 'content_md5';
@@ -75,25 +69,24 @@ async function hash_target() {
             }
         }());
         const target = new TargetHash();
-        const chunk_fs = new ChunkFS({
+        const file_writer = new FileWriter({
             target_file: target,
             fs_context: DEFAULT_FS_CONFIG,
-            rpc_client: DUMMY_RPC,
             namespace_resource_id: 'MajesticSloth'
         });
-        await stream_utils.pipeline([source_stream, chunk_fs]);
-        await stream_utils.wait_finished(chunk_fs);
+        await stream_utils.pipeline([source_stream, file_writer]);
+        await stream_utils.wait_finished(file_writer);
         const write_hash = target.digest();
         console.log(
             'Hash target',
-            `NativeMD5=${chunk_fs.digest}`,
+            `NativeMD5=${file_writer.digest}`,
             `DataWriteCryptoMD5=${write_hash}`,
             `DataOriginMD5=${content_md5}`,
         );
         assert.strictEqual(content_md5, write_hash);
         if (config.NSFS_CALCULATE_MD5) {
-            assert.strictEqual(chunk_fs.digest, content_md5);
-            assert.strictEqual(chunk_fs.digest, write_hash);
+            assert.strictEqual(file_writer.digest, content_md5);
+            assert.strictEqual(file_writer.digest, write_hash);
         }
     });
 }
@@ -113,32 +106,31 @@ async function file_target(chunk_size = CHUNK, parts = PARTS) {
                     yield data.slice(i, i + chunk_size);
                 }
             }());
-            const chunk_fs = new ChunkFS({
+            const file_writer = new FileWriter({
                 target_file,
                 fs_context: DEFAULT_FS_CONFIG,
-                rpc_client: DUMMY_RPC,
                 namespace_resource_id: 'MajesticSloth'
             });
-            await stream_utils.pipeline([source_stream, chunk_fs]);
-            await stream_utils.wait_finished(chunk_fs);
+            await stream_utils.pipeline([source_stream, file_writer]);
+            await stream_utils.wait_finished(file_writer);
             if (XATTR) {
                 await target_file.replacexattr(
                     DEFAULT_FS_CONFIG,
-                    assign_md5_to_fs_xattr(chunk_fs.digest, {})
+                    assign_md5_to_fs_xattr(file_writer.digest, {})
                 );
             }
             if (FSYNC) await target_file.fsync(DEFAULT_FS_CONFIG);
             const write_hash = crypto.createHash('md5').update(fs.readFileSync(F_TARGET)).digest('hex');
             console.log(
                 'File target',
-                `NativeMD5=${chunk_fs.digest}`,
+                `NativeMD5=${file_writer.digest}`,
                 `DataWriteMD5=${write_hash}`,
                 `DataOriginMD5=${content_md5}`,
             );
             assert.strictEqual(content_md5, write_hash);
             if (config.NSFS_CALCULATE_MD5) {
-                assert.strictEqual(chunk_fs.digest, content_md5);
-                assert.strictEqual(chunk_fs.digest, write_hash);
+                assert.strictEqual(file_writer.digest, content_md5);
+                assert.strictEqual(file_writer.digest, write_hash);
             }
             // Leave parts on error
             fs.rmSync(F_TARGET);