Nodejs Error: EMFILE

Question

I need to open a directory containing files.. open readstream for each and write the data from all the files into a single file. But I keep getting Error: EMFILE, open 'chunks/piece96.data'

My ulimit was 256 and I increased it to 1024. I have 127 files in the directory to open, read and write into a single file.

My code is below

    var DIR='chunks/';
    var files=fs.readdirSync(DIR);
    var filename='bach.mp3';

    files.forEach(function(singlebit){
        //console.log(files);
        var bit=fs.createReadStream(DIR+singlebit);
        var resultfile=fs.createWriteStream(filename,{
            flags:'r+',
            encoding:null,
            mode:0666
        });
        bit.on('data',function(bitdata){ 
                resultfile.write(bitdata); 
                console.log(bitdata);
            }).on('end',function(){
                resultfile.end();
            });
        });
    console.log('file complete');

How can I prevent the EMI file error. I am not opening many files at once since I am using readdirSync and not opening all of them at once. I need a way to read all the files and write to a single file.

Answer 1

I just finished writing a little snippet of code to solve this problem myself, I don't use streams, but this should be adaptable to your needs:

// Queuing reads and writes, so your nodejs script doesn't overwhelm system limits catastrophically
global.maxFilesInFlight = 100; // Set this value to some number safeish for your system
var origRead = fs.readFile;
var origWrite = fs.writeFile;

var activeCount = 0;
var pending = [];

var wrapCallback = function(cb){
    return function(){
        activeCount--;
        cb.apply(this,Array.prototype.slice.call(arguments));
        if (activeCount < global.maxFilesInFlight && pending.length){
            console.log("Processing Pending read/write");
            pending.shift()();
        }
    };
};
fs.readFile = function(){
    var args = Array.prototype.slice.call(arguments);
    if (activeCount < global.maxFilesInFlight){
        if (args[1] instanceof Function){
            args[1] = wrapCallback(args[1]);
        } else if (args[2] instanceof Function) {
            args[2] = wrapCallback(args[2]);
        }
        activeCount++;
        origRead.apply(fs,args);
    } else {
        console.log("Delaying read:",args[0]);
        pending.push(function(){
            fs.readFile.apply(fs,args);
        });
    }
};

fs.writeFile = function(){
    var args = Array.prototype.slice.call(arguments);
    if (activeCount < global.maxFilesInFlight){
        if (args[1] instanceof Function){
            args[1] = wrapCallback(args[1]);
        } else if (args[2] instanceof Function) {
            args[2] = wrapCallback(args[2]);
        }
        activeCount++;
        origWrite.apply(fs,args);
    } else {
        console.log("Delaying write:",args[0]);
        pending.push(function(){
            fs.writeFile.apply(fs,args);
        });
    }
};