Merge pull request #8728 from jackyalbo/jacky-fix5

Fix to break semaphore circular dependency

Author: jackyalbo

src/agent/agent_cli.js

@@ -20,7 +20,7 @@ const dbg = require('../util/debug_module')(__filename);
 const Agent = require('./agent');
 const fs_utils = require('../util/fs_utils');
 const os_utils = require('../util/os_utils');
-const Semaphore = require('../util/semaphore');
+const semaphore = require('../util/semaphore');
 const json_utils = require('../util/json_utils');
 const addr_utils = require('../util/addr_utils');
 const debug_config = require('../util/debug_config');
@@ -494,7 +494,7 @@ class AgentCLI {
         if (n === 0) {
             return self.create(0, paths_to_work_on);
         } else {
-            const sem = new Semaphore(5);
+            const sem = new semaphore.Semaphore(5);
             return P.all(_.times(n, function() {
                 return sem.surround(function() {
                     return self.create(n, paths_to_work_on);

src/agent/block_store_services/block_store_mongo.js

@@ -9,15 +9,15 @@ const size_utils = require('../../util/size_utils');
 const mongo_client = require('../../util/mongo_client');
 const buffer_utils = require('../../util/buffer_utils');
 const BlockStoreBase = require('./block_store_base').BlockStoreBase;
-const Semaphore = require('../../util/semaphore');
+const semaphore = require('../../util/semaphore');
 
 // limiting the IO concurrency on mongo
 // we use a very low limit since this is used only for temporary internal storage
 // which is meant for quick onboarding, and not for performance/production.
-const sem_head = new Semaphore(3);
-const sem_read = new Semaphore(1);
-const sem_write = new Semaphore(1);
-const sem_delete = new Semaphore(1);
+const sem_head = new semaphore.Semaphore(3);
+const sem_read = new semaphore.Semaphore(1);
+const sem_write = new semaphore.Semaphore(1);
+const sem_delete = new semaphore.Semaphore(1);
 
 const GRID_FS_BUCKET_NAME = 'mongo_internal_agent';
 const GRID_FS_BUCKET_NAME_FILES = `${GRID_FS_BUCKET_NAME}.files`;

src/agent/func_services/func_node.js

@@ -9,7 +9,7 @@ const P = require('../../util/promise');
 const dbg = require('../../util/debug_module')(__filename);
 const { RpcError, RPC_BUFFERS } = require('../../rpc');
 const fs_utils = require('../../util/fs_utils');
-const Semaphore = require('../../util/semaphore');
+const semaphore = require('../../util/semaphore');
 const zip_utils = require('../../util/zip_utils');
 
 const FUNC_PROC_PATH = path.resolve(__dirname, 'func_proc.js');
@@ -22,7 +22,7 @@ class FuncNode {
         this.storage_path = params.storage_path || '.';
         this.functions_path = path.join(this.storage_path, 'functions');
         this.functions_loading_path = path.join(this.storage_path, 'functions_loading');
-        this.loading_serial = new Semaphore(1);
+        this.loading_serial = new semaphore.Semaphore(1);
     }
 
     async invoke_func(req) {

src/rpc/ice.js

@@ -6,6 +6,7 @@ module.exports = Ice;
 
 const _ = require('lodash');
 const P = require('../util/promise');
+const Defer = require('../util/defer');
 const os = require('os');
 const net = require('net');
 const tls = require('tls');
@@ -1341,7 +1342,7 @@ function IceSession(local, remote, packet, udp) {
     self.state = 'init';
     js_utils.self_bind(self, 'run_udp_request_loop');
     js_utils.self_bind(self, 'run_udp_indication_loop');
-    self.defer = new P.Defer();
+    self.defer = new Defer();
     self.defer.promise.catch(_.noop); // to ignore 'Unhandled rejection' printouts
     // set session timeout
     self.ready_timeout = setTimeout(function() {

src/rpc/rpc.js

@@ -11,6 +11,7 @@ const assert = require('assert');
 const EventEmitter = require('events').EventEmitter;
 
 const P = require('../util/promise');
+const Defer = require('../util/defer');
 const js_utils = require('../util/js_utils');
 const dbg = require('../util/debug_module')(__filename);
 const config = require('../../config');
@@ -191,7 +192,7 @@ class RPC extends EventEmitter {
 
             // initialize the request
             req._new_request(api, method_api, params, options.auth_token);
-            req._response_defer = new P.Defer();
+            req._response_defer = new Defer();
             req._response_defer.promise.catch(_.noop); // to prevent error log of unhandled rejection
 
             if (options.tracker) {

src/rpc/rpc_base_conn.js

@@ -7,6 +7,8 @@ const _ = require('lodash');
 const events = require('events');
 
 const P = require('../util/promise');
+const Defer = require('../util/defer');
+
 const dbg = require('../util/debug_module')(__filename);
 const config = require('../../config');
 const time_utils = require('../util/time_utils');
@@ -38,7 +40,7 @@ class RpcBaseConnection extends events.EventEmitter {
         this._state = STATE_INIT;
 
         // the connecting_defer is used by connect() to wait for the connected event
-        this.connecting_defer = new P.Defer();
+        this.connecting_defer = new Defer();
         this._connect_promise = events.once(this, 'connect');
         this._connect_promise.catch(_.noop); // to prevent error log of unhandled rejection
 

src/rpc/rpc_http.js

@@ -5,7 +5,7 @@ const _ = require('lodash');
 const http = require('http');
 const https = require('https');
 
-const P = require('../util/promise');
+const Defer = require('../util/defer');
 const dbg = require('../util/debug_module')(__filename);
 const buffer_utils = require('../util/buffer_utils');
 const http_utils = require('../util/http_utils');
@@ -145,7 +145,7 @@ class RpcHttpConnection extends RpcBaseConnection {
 
         dbg.log3('HTTP request', http_req.method, http_req.path, http_req._headers);
 
-        const send_defer = new P.Defer();
+        const send_defer = new Defer();
 
         // reject on send errors
         http_req.on('error', send_defer.reject);

src/rpc/rpc_request.js

@@ -2,7 +2,7 @@
 'use strict';
 
 /** @typedef {import('./rpc_base_conn')} RpcBaseConnection */
-/** @typedef {import('../util/promise').Defer} Defer */
+/** @typedef {import('../util/defer')} Defer */
 
 const _ = require('lodash');
 const RpcMessage = require('./rpc_message');

src/sdk/map_client.js

@@ -16,7 +16,7 @@ const LRUCache = require('../util/lru_cache');
 const s3_utils = require('../endpoint/s3/s3_utils');
 const db_client = require('../util/db_client');
 const nb_native = require('../util/nb_native');
-const Semaphore = require('../util/semaphore');
+const semaphore = require('../util/semaphore');
 const KeysSemaphore = require('../util/keys_semaphore');
 const block_store_client = require('../agent/block_store_services/block_store_client').instance();
 const system_store = require('../server/system_services/system_store').get_instance();
@@ -25,9 +25,9 @@ const { ChunkAPI } = require('./map_api_types');
 const { RpcError, RPC_BUFFERS } = require('../rpc');
 
 // semphores global to the client
-const block_write_sem_global = new Semaphore(config.IO_WRITE_CONCURRENCY_GLOBAL);
-const block_replicate_sem_global = new Semaphore(config.IO_REPLICATE_CONCURRENCY_GLOBAL);
-const block_read_sem_global = new Semaphore(config.IO_READ_CONCURRENCY_GLOBAL);
+const block_write_sem_global = new semaphore.Semaphore(config.IO_WRITE_CONCURRENCY_GLOBAL);
+const block_replicate_sem_global = new semaphore.Semaphore(config.IO_REPLICATE_CONCURRENCY_GLOBAL);
+const block_read_sem_global = new semaphore.Semaphore(config.IO_READ_CONCURRENCY_GLOBAL);
 
 // semphores specific to an agent
 const block_write_sem_agent = new KeysSemaphore(config.IO_WRITE_CONCURRENCY_AGENT);
@@ -652,7 +652,7 @@ class MapClient {
                 });
                 dbg.log1('MapClient: move_blocks_to_storage_class SUCCEEDED', 'ADDR:', agent_address, 'MOVED:', moved);
             } catch (err) {
-                dbg.error('MapClient: move_blocks_to_storage_class FAILED', 'ADDR:', agent_address, 'ERROR', err,);
+                dbg.error('MapClient: move_blocks_to_storage_class FAILED', 'ADDR:', agent_address, 'ERROR', err);
             }
         });
     }

src/sdk/namespace_cache.js

@@ -11,13 +11,13 @@ const RangeStream = require('../util/range_stream');
 const P = require('../util/promise');
 const buffer_utils = require('../util/buffer_utils');
 const stream_utils = require('../util/stream_utils');
-const Semaphore = require('../util/semaphore');
+const semaphore = require('../util/semaphore');
 const S3Error = require('../endpoint/s3/s3_errors').S3Error;
 const s3_utils = require('../endpoint/s3/s3_utils');
 const config = require('../../config');
 const size_utils = require('../util/size_utils');
 
-const _global_cache_uploader = new Semaphore(cache_config.UPLOAD_SEMAPHORE_CAP, {
+const _global_cache_uploader = new semaphore.Semaphore(cache_config.UPLOAD_SEMAPHORE_CAP, {
     timeout: cache_config.UPLOAD_SEMAPHORE_TIMEOUT,
     timeout_error_code: 'NAMESPACE_CACHE_UPLOAD_TIMEOUT'
 });
@@ -36,7 +36,7 @@ class NamespaceCache {
      *      stats: import('./endpoint_stats_collector').EndpointStatsCollector,
      * }} params
      */
-   constructor({ namespace_hub, namespace_nb, caching, active_triggers, stats }) {
+    constructor({ namespace_hub, namespace_nb, caching, active_triggers, stats }) {
         this.namespace_hub = namespace_hub;
         this.namespace_nb = namespace_nb;
         this.active_triggers = active_triggers;