load_test.js

/*
ulimit -n 4096 -u 256; export ERL_MAX_PORTS=4096
node backend/_attachments/load_test.js
couchdb -d -p /tmp/couch.pid
rm /tmp/couch.* /tmp/*.couch
*/

var mc = require('multicouch'),
    f = require('fermata'),
    q = require('queue-async');

// node.js http client derps instead of queing like docs claim (Fall 2012 at least)
require('http').globalAgent.maxSockets = Infinity;
require('https').globalAgent.maxSockets = Infinity;

var TEST_PORT = 4001,
    NUM_USERS = 250,
    POLL_TIME = 15*60*1000,
    PREFIX = 'test';


var server,
    testCouch = null;
if (process.argv.length > 2) {
    server = f.json(process.argv[2]);
} else {
    testCouch = new mc({port:TEST_PORT, prefix: "/tmp"});
    server = f.json("http://localhost:"+TEST_PORT);
    testCouch.start();
}

function waitForCouch(server, callback) {
    server.get(function (e,d) {
        if (e && e.code === 'ECONNREFUSED') setTimeout(function () { waitForCouch(server, callback); }, 100);
        else callback(e, d);
    });
}

waitForCouch(server, function (e,d) {
    if (e) console.error(e);
    else main();
});

function main() {
    q(1)
        //.defer(setupDatabases)
        .defer(setupReplications)
        .defer(pollChanges, POLL_TIME)
        .defer(wait, POLL_TIME)
    .await(function (e) {
        if (e) console.error("ERROR:", e);
        else if (testCouch) testCouch.stop();          // for now, leave CouchDB running on error for easier debug
    });
}

function _ignore(code, cb) {
    return function (e,d) {
        if (e && e.status === code) e = null;
        cb(e);
    }
}

function setupDatabases(cb) {
    console.log("SETUP DATABASES");
    var queue = q(6);
    
    // setup higher limits, see http://wiki.apache.org/couchdb/Performance#Resource_Limits
    if (testCouch) queue.defer(function (cb) {
        var n = Math.max(100, NUM_USERS+3);
        server._config.couchdb.max_dbs_open.put(''+n, cb);
    });
    if (testCouch) queue.defer(function (cb) {
        var n = Math.max(2048, 42*NUM_USERS);
        server._config.httpd.max_connections.put(''+n, cb);
    });
    
    for (var i = 0; i < NUM_USERS + 1; i += 1) queue.defer(function (dbName, cb) {
        console.log("Creating database", dbName);
        server[dbName].put(_ignore(412, function (e) {
            if (e) return cb(e);
            server[dbName].post({
               _id: "_design/load_test",
               filters: {always_true: "function() { return true; }"}
            }, _ignore(409,cb));
        }));
    }, PREFIX+i);
    queue.await(cb);
}

function setupReplications(cb) {
    console.log("SETUP REPLICATIONS");
    var queue = q(6);
    for (var i = 0; i < NUM_USERS; i += 1) {
        function startReplication(src, tgt, cb) {
            console.log("Starting replication", src, "->", tgt);
            server._replicator.post({
                _id: src + "-" + tgt,
                source: (testCouch) ? src : server(src)(),
                filter: "load_test/always_true",
                query_params: {db:src+'-'+tgt},
                target: (testCouch) ? tgt : server(tgt)(),
                continuous: true,
                
                worker_processes: 1,
                http_connections: 2
            }, _ignore(409,cb));
        }
        queue.defer(startReplication, PREFIX+(i+1), PREFIX+'0');
        queue.defer(startReplication, PREFIX+'0', PREFIX+(i+1));
    }
    queue.await(cb);
}

function pollChanges(duration, cb) {
    console.log("POLL CHANGES");
    var watch = require('fermata/plugins/couchdb').watchChanges,
        watchers = [];
    for (var i = 0; i < NUM_USERS; i += 1) {
        var cancel = watch(server(PREFIX+(i+1), {filter:"load_test/always_true", timeout:1000}), 0, function (){});
        watchers.push(cancel);
    }
    function makeChange() {
        server[PREFIX+'0'].post({}, function (e,d) {
            makeChange();
        });
    }
    makeChange();
    setTimeout(function () {
        makeChange = function () {};
        watchers.forEach(function (cancel) { cancel(); });
        cb();
    }, duration);
}

function wait(t, cb) {
    console.log("WAITING");
    setTimeout(cb, t);
}