Testing memory usage for a large number of events with changefeeds and eachAsync

Testing eachAsync for memory leaks

var Promise = require('bluebird');
var util = require('util');
var heapdump = require('heapdump');

function log(message) {
    console.log(new Date().toISOString() + ' - ' + message);
}

function gc() {
    log('-- Forced GC');
    global.gc();
}
var countOps = 0;
var countOpsSuccess = 0;
var countOpsErrors = 0;

var initialMemoryUsage = process.memoryUsage();
log(util.inspect(initialMemoryUsage));
function printMemoryUsage() {
    var diff = process.memoryUsage();
    var result = {};
    result.diff_rss = diff.rss - initialMemoryUsage.rss;
    result.diff_heapTotal = diff.heapTotal - initialMemoryUsage.heapTotal;
    result.diff_heapUsed = diff.heapUsed - initialMemoryUsage.heapUsed;
    log(`${countOps} tried, ${countOpsSuccess} done, ${countOpsErrors} errors; ${parseInt(result.diff_heapUsed / (1024 * 1024 ))} MiB`);
}

var INSERT_TIMERS = 100;
var DELETE_TIMERS = 20;
var UPDATE_TIMERS = 150;
var TIME_END = parseFloat(process.argv[2] || 15) * 60 * 1000;


function startTest() {
    log('-- Start test');
    var r = require('rethinkdbdash')({
        max: 1000,
        buffer: 50,
        host: 'localhost',
        port: 28015,
        // timeoutGb: 1000 * 10
    });

    setup(r).then(function () {
        var feeds = [];
        openFeeds(r, feeds);
        var timers = [];
        fireUpdates(r, timers);
        setTimeout(function () {
            tearDown(r, feeds, timers)
        }, TIME_END);
        timers.push(setInterval(() => printMemoryUsage(), 5000));
        timers.push(setInterval(() => gc(), 60000));
        return null;
    }).catch(function (err) {
        log('Could not setup the test. ' + err);
    });
}

function setup(r) {
    log('-- Setup test');
    return r.dbCreate('test').run().then(function () {
        log('-- db created');
    }).catch(function () {
        log('-- db not created');
    }).finally(function () {
        return r.db('test').tableDrop('test').run()
    }).then(function () {
        log('-- table dropped');
    }).catch(function () {
        log('-- table not dropped');
    }).finally(function () {
        return r.db('test').tableCreate('test').run()
    }).then(function () {
        log('-- table created');
    }).catch(function () {
        log('-- table not created');
    }).finally(function () {
        return Promise.resolve(true);
    });
}

function openFeeds(r, feeds) {
    printMemoryUsage();
    gc()
    printMemoryUsage();
    log('-- Open feeds');
    var count = 0;
    r.db('test').table('test').changes().run().then(function (feed) {
        feeds.push(feed);
        feed.eachAsync(function () {
            count++;
        }).catch(function (e) {
            log('Caught error?');
            log(e);
        });
    });
}

function fireUpdates(r, timers) {
    log('-- Fire updates');
    for (var i = 0; i < INSERT_TIMERS; i++) {
        timers.push(setInterval(function () {
            countOps++;
            r.db('test').table('test').insert({
                a: "a".repeat(10),
                b: "b".repeat(10),
                c: "c".repeat(10),
                d: "d".repeat(10)
            }).run().then(function () {
                countOpsSuccess++;
            }).catch(function (e) {
                countOpsErrors++;
            });
        }, 1000));
    }
    for (var i = 0; i < DELETE_TIMERS; i++) {
        timers.push(setInterval(function () {
            r.db('test').table('test').limit(1).delete().run();
        }, 1000));
    }
    for (var i = 0; i < UPDATE_TIMERS; i++) {
        timers.push(setInterval(function () {
            r.db('test').table('test').limit(1).update({
                value: Math.random()
            }).run();
        }, 1000));
    }
}

function tearDown(r, feeds, timers) {
    log(`-- countOps: ${countOps}`);
    log(`-- countOpsSuccess: ${countOpsSuccess}`);
    log(`-- countOpsErrors: ${countOpsErrors}`);
    log('-- Tear down');
    printMemoryUsage();
    gc();
    printMemoryUsage();
    for (var i = 0; i < timers.length; i++) {
        clearInterval(timers[i]);
    }
    Promise.map(feeds, function (feed) {
        return feed.close().catch(function (e) {
            log('Caught error?');
            log(e);
        });
    }).then(function () {
        log('-- Feed closed');
        feeds = [];
        timers = [];
        printMemoryUsage();
        gc()
        printMemoryUsage();
        // return r.getPoolMaster().drain();
    }).then(function () {
        setTimeout(function () {
            log('-- Pool master drained');
            printMemoryUsage();
            gc()
            printMemoryUsage();
            r = null;
            printMemoryUsage();
    log(`-- countOps: ${countOps}`);
    log(`-- countOpsSuccess: ${countOpsSuccess}`);
    log(`-- countOpsErrors: ${countOpsErrors}`);
            gc()
            printMemoryUsage();
        }, 5000);
        log('-- Pool master drained');
        printMemoryUsage();
        gc()
        printMemoryUsage();
        // r = null;
        printMemoryUsage();
    log(`-- countOps: ${countOps}`);
    log(`-- countOpsSuccess: ${countOpsSuccess}`);
    log(`-- countOpsErrors: ${countOpsErrors}`);
        gc()
        printMemoryUsage();
        // heapdump.writeSnapshot('./' + Date.now() + '.heapsnapshot');
    });
}


startTest();