Bugfix/issue 95 worker mem leak 2 (#106)

* Refactored to remove the ClusterMessages module altogether - it was adding too much complexity and memory overhead for only a slight improvement in readability over just using the Cluster module directly as necessary.

* Remove file ref count logic from CacheFS module. Memory usage was scaling with project size which eventually resulted in unacceptable memory usage. Also fixed a couple of warnings and errors in the test suite.

Author: stephen-palmer

lib/cache/cache_fs.js

@@ -9,25 +9,8 @@ const moment = require('moment');
 const pick = require('lodash').pick;
 const crypto = require('crypto');
 const fileExtensions = require('lodash').invert(consts.FILE_TYPE);
-const cluster = require('cluster');
-const cm = require('../cluster_messages');
-
-const k_incrementGuidRef = "_incrementGuidRef";
-const k_releaseGuidRef = "_releaseGuidRef";
 
 class CacheFS extends CacheBase {
-    constructor() {
-        super();
-        this._guidRefs = {};
-
-        cm.listenFor(k_incrementGuidRef, key => {
-            this._incrementGuidRef(key);
-        });
-
-        cm.listenFor(k_releaseGuidRef, key => {
-            this._releaseGuidRef(key);
-        });
-    }
 
     get _optionsPath() {
         return super._optionsPath + ".cache_fs";
@@ -65,39 +48,6 @@ class CacheFS extends CacheBase {
         return result;
     }
 
-    /**
-     *
-     * @param {String} key
-     * @private
-     */
-    _incrementGuidRef(key) {
-        if(cluster.isWorker) {
-            return cm.send(k_incrementGuidRef, key);
-        }
-
-        if(this._guidRefs.hasOwnProperty(key)) {
-            this._guidRefs[key]++;
-        }
-        else {
-            this._guidRefs[key] = 1;
-        }
-    }
-
-    /**
-     *
-     * @param {String} key
-     * @private
-     */
-    _releaseGuidRef(key) {
-        if(cluster.isWorker) {
-            return cm.send(k_releaseGuidRef, key);
-        }
-
-        if(this._guidRefs.hasOwnProperty(key)) {
-            this._guidRefs[key]--;
-        }
-    }
-
     /**
      *
      * @param {String} type
@@ -123,10 +73,6 @@ class CacheFS extends CacheBase {
     async _writeFileToCache(type, guid, hash, sourcePath) {
         const filePath = this._calcFilepath(type, guid, hash);
 
-        if(this._guidRefs.hasOwnProperty(filePath) && this._guidRefs[filePath] > 0) {
-            throw new Error(`File is busy, cannot overwrite ${key}`);
-        }
-
         await fs.move(sourcePath, filePath, { overwrite: true });
         return filePath;
     }
@@ -143,13 +89,10 @@ class CacheFS extends CacheBase {
 
         return new Promise((resolve, reject) => {
             stream.on('open', () => {
-                this._incrementGuidRef(key);
                 resolve(stream);
             }).on('error', err => {
                 helpers.log(consts.LOG_ERR, err);
                 reject(err);
-            }).on('close', () => {
-                this._releaseGuidRef(key);
             });
         });
     }

lib/cache/reliability_manager.js

@@ -1,7 +1,6 @@
 const cluster = require('cluster');
 const path = require('path');
 const helpers = require('../helpers');
-const cm = require('../cluster_messages');
 const consts = require('../constants');
 const crypto = require('crypto');
 const _ = require('lodash');
@@ -15,6 +14,8 @@ const defaultOptions = {
 
 class ReliabilityManager {
     constructor(db, cachePath, options) {
+        this._id = ++ReliabilityManager._idCounter;
+        this._kMsgUpdateReliabilityFactor = `_updateReliabilityFactorForVersion.${this._id}`;
         this._db = db;
         this._cachePath = cachePath;
         this._options = options || {};
@@ -30,9 +31,13 @@ class ReliabilityManager {
             }
         }
 
-        cm.listenFor("_updateReliabilityFactorForVersion", (data) => {
-            return this._updateReliabilityFactorForVersion(data);
-        });
+        if(cluster.isMaster) {
+            cluster.on('message', (worker, msg) => {
+                if(msg._msg === this._kMsgUpdateReliabilityFactor) {
+                    return this._updateReliabilityFactorForVersion(msg);
+                }
+            });
+        }
     }
 
     /**
@@ -43,7 +48,8 @@ class ReliabilityManager {
      */
     async _updateReliabilityFactorForVersion(params) {
         if(cluster.isWorker) {
-            return cm.send('_updateReliabilityFactorForVersion', params);
+            params._msg = this._kMsgUpdateReliabilityFactor;
+            return process.send(params);
         }
 
         const entry = this.getEntry(params.guidStr, params.hashStr, true);
@@ -160,4 +166,6 @@ ReliabilityManager.reliabilityStates = {
     Reliable: 'Reliable'
 };
 
+ReliabilityManager._idCounter = 0;
+
 module.exports = ReliabilityManager;

lib/cluster_messages.js

@@ -113,7 +113,7 @@ class CommandProcessor extends Duplex {
         if(this[kReadStream] === null || this[kSource] === null) return;
 
         let chunk;
-        while(chunk = this[kReadStream].read()) {
+        while((chunk = this[kReadStream].read()) !== null) {
             this._sendFileQueueChunkReads++;
             this._sendFileQueueReadBytes += chunk.length;
             if(!this.push(chunk, 'ascii')) break;

lib/server/command_processor.js

@@ -6,7 +6,6 @@ const loki = require('lokijs');
 const fs = require('fs-extra');
 const sleep = require('./test_utils').sleep;
 const generateCommandData = require('./test_utils').generateCommandData;
-const readStream = require('./test_utils').readStream;
 const sinon = require('sinon');
 const crypto = require('crypto');
 const consts = require('../lib/constants');
@@ -147,32 +146,6 @@ describe("Cache API", function() {
                         .then(() => { throw new Error("Expected error!"); }, err =>  assert(err));
                 });
 
-                it("should throw an error when trying to replace a file that is open for reading", async () => {
-                    const TEST_FILE_SIZE = 1024 * 64 * 2;
-
-                    const fData = generateCommandData(TEST_FILE_SIZE, TEST_FILE_SIZE);
-
-                    // Add a file to the cache (use the info data)
-                    let trx = await cache.createPutTransaction(fData.guid, fData.hash);
-                    let wStream = await trx.getWriteStream(consts.FILE_TYPE.INFO, fData.info.length);
-                    await new Promise(resolve => wStream.end(fData.info, resolve));
-                    await cache.endPutTransaction(trx);
-
-                    // Get a read stream
-                    const rStream = await cache.getFileStream(consts.FILE_TYPE.INFO, fData.guid, fData.hash);
-
-                    // Read a byte
-                    await new Promise(resolve => rStream.once('readable', () => resolve(rStream.read(1))));
-
-                    // Try to replace the file (use the resource data)
-                    trx = await cache.createPutTransaction(fData.guid, fData.hash);
-                    wStream = await trx.getWriteStream(consts.FILE_TYPE.INFO, fData.resource.length);
-                    await new Promise(resolve => wStream.end(fData.resource, resolve));
-
-                    cache.endPutTransaction(trx).then(() => { throw new Error("Expected error"); }, (err) => assert(err))
-                        .then(() => rStream.destroy());
-                });
-
                 describe("High Reliability Mode", () => {
                     before(async () => {
                         const opts = cache._options;

test/cache_api.js

@@ -48,7 +48,10 @@ describe("Cache: FS", () => {
             cache = new Cache();
         });
 
-        afterEach(() => fs.remove(cacheOpts.cachePath));
+        afterEach(async () => {
+            await cache.shutdown();
+            return fs.remove(cacheOpts.cachePath)
+        });
 
         describe("cleanup", function() {
             this.slow(500);
@@ -189,7 +192,7 @@ describe("Cache: FS", () => {
 
                 await cache.init(opts);
                 const file = await addFileToCache(moment().toDate());
-                cache.cleanup(true);
+                await cache.cleanup(true);
                 assert(await fs.pathExists(file.path));
             });
 

test/cache_fs.js

@@ -112,6 +112,33 @@ describe("Cache: RAM", function() {
                 await cache.endPutTransaction(trx);
                 assert(ok);
             });
+
+            it("should throw an error when trying to replace a file that is open for reading", async () => {
+                const TEST_FILE_SIZE = 1024 * 64 * 2;
+
+                const fData = generateCommandData(TEST_FILE_SIZE, TEST_FILE_SIZE);
+
+                // Add a file to the cache (use the info data)
+                await cache.init(opts);
+                let trx = await cache.createPutTransaction(fData.guid, fData.hash);
+                let wStream = await trx.getWriteStream(consts.FILE_TYPE.INFO, fData.info.length);
+                await new Promise(resolve => wStream.end(fData.info, resolve));
+                await cache.endPutTransaction(trx);
+
+                // Get a read stream
+                const rStream = await cache.getFileStream(consts.FILE_TYPE.INFO, fData.guid, fData.hash);
+
+                // Read a byte
+                await new Promise(resolve => rStream.once('readable', () => resolve(rStream.read(1))));
+
+                // Try to replace the file (use the resource data)
+                trx = await cache.createPutTransaction(fData.guid, fData.hash);
+                wStream = await trx.getWriteStream(consts.FILE_TYPE.INFO, fData.resource.length);
+                await new Promise(resolve => wStream.end(fData.resource, resolve));
+
+                return cache.endPutTransaction(trx).then(() => { throw new Error("Expected error"); }, (err) => assert(err))
+                    .then(() => rStream.destroy());
+            });
         });
 
         describe("shutdown", () => {
@@ -144,7 +171,8 @@ describe("Cache: RAM", function() {
 
         });
 
-        afterEach(() => {
+        afterEach(async () => {
+            await cache.shutdown();
             cache._clearCache();
             return fs.remove(opts.cachePath);
         });

test/cache_ram.js

@@ -140,7 +140,7 @@ describe("Server common", function() {
 
         it("should log an error if the error code is 'EADDRINUSE", async () => {
             return new Promise((resolve, reject) => {
-                server.errCallback = e => {}
+                server.errCallback = e => {};
                 helpers.setLogger((lvl, msg) => {
                     /already in use/.test(msg) ? resolve() : reject();
                 });

test/server.js

@@ -1,4 +1,5 @@
 const helpers = require('../lib/helpers');
+require('cluster').setMaxListeners(25);
 
 process.on('unhandledRejection', (reason) => {
     console.error(reason);

test/test_init.js

@@ -124,7 +124,7 @@ exports.readStream = function(stream, size) {
         const buffer = Buffer.alloc(size, 0, 'ascii');
         stream.on('readable', function() {
             let data;
-            while(data = this.read()) {
+            while((data = this.read()) !== null) {
                 if(pos + data.length > size) {
                     reject(new Error("Stream size exceeds buffer size allocation"));
                 }