33

I'm trying to write a small node application that will search through and parse a large number of files on the file system. In order to speed up the search, we are attempting to use some sort of map reduce. The plan would be the following simplified scenario:

  • Web request comes in with a search query
  • 3 processes are started that each get assigned 1000 (different) files
  • once a process completes, it would 'return' it's results back to the main thread
  • once all processes complete, the main thread would continue by returning the combined result as a JSON result

The questions I have with this are: Is this doable in Node? What is the recommended way of doing it?

I've been fiddling, but come no further then following example using Process:

initiator:

function Worker() { 
    return child_process.fork("myProcess.js"); 
}

for(var i = 0; i < require('os').cpus().length; i++){
        var process = new Worker();
        process.send(workItems.slice(i * itemsPerProcess, (i+1) * itemsPerProcess));
}

myProcess.js

process.on('message', function(msg) {
    var valuesToReturn = [];
    // Do file reading here
    //How would I return valuesToReturn?
    process.exit(0);
}

Few sidenotes:

  • I'm aware the number of processes should be dependent of the number of CPU's on the server
  • I'm also aware of speed restrictions in a file system. Consider it a proof of concept before we move this to a database or Lucene instance :-)
1
  • Assuming optimising a proof-of-concept is the right use of time, then I wouldn't do it like this. I'd build an index and watch the file system for changes if necessary. That way you'd get much better performance. If this is just an exercise in learning worker processes then go ahead. Commented Nov 15, 2013 at 17:47

2 Answers 2

28

Should be doable. As a simple example:

// parent.js
var child_process = require('child_process');

var numchild  = require('os').cpus().length;
var done      = 0;

for (var i = 0; i < numchild; i++){
  var child = child_process.fork('./child');
  child.send((i + 1) * 1000);
  child.on('message', function(message) {
    console.log('[parent] received message from child:', message);
    done++;
    if (done === numchild) {
      console.log('[parent] received all results');
      ...
    }
  });
}

// child.js
process.on('message', function(message) {
  console.log('[child] received message from server:', message);
  setTimeout(function() {
    process.send({
      child   : process.pid,
      result  : message + 1
    });
    process.disconnect();
  }, (0.5 + Math.random()) * 5000);
});

So the parent process spawns an X number of child processes and passes them a message. It also installs an event handler to listen for any messages sent back from the child (with the result, for instance).

The child process waits for messages from the parent, and starts processing (in this case, it just starts a timer with a random timeout to simulate some work being done). Once it's done, it sends the result back to the parent process and uses process.disconnect() to disconnect itself from the parent (basically stopping the child process).

The parent process keeps track of the number of child processes started, and the number of them that have sent back a result. When those numbers are equal, the parent received all results from the child processes so it can combine all results and return the JSON result.

5
  • When the child process does a process.send, isn't there a chance that it's own event handler will catch the on('message') as well as the parent's event handler? Commented Nov 18, 2013 at 15:25
  • 1
    @BartVangeneugden no, process.send() is meant for child-to-parent communications, and child.send() is for the other way around (from parent to child). See also the official documentation.
    – robertklep
    Commented Nov 18, 2013 at 15:32
  • Hello @robertklep I have used worker-thread similar to your examples. But I have checked CPU utilization is too high. Please suggest the best way
    – Sudhir
    Commented Jan 2, 2023 at 13:22
  • @Sudhir perhaps you should post a new question explaining your specific problems and the code that you're using.
    – robertklep
    Commented Jan 2, 2023 at 14:16
  • @robertklep I have added can you please look into it stackoverflow.com/questions/74993486/…
    – Sudhir
    Commented Jan 3, 2023 at 13:52
7

For a distributed problem like this, I've used zmq and it has worked really well. I'll give you a similar problem that I ran into, and attempted to solve via processes (but failed.) and then turned towards zmq.

Using bcrypt, or an expensive hashing algorith, is wise, but it blocks the node process for around 0.5 seconds. We had to offload this to a different server, and as a quick fix, I used essentially exactly what you did. Run a child process and send messages to it and get it to respond. The only issue we found is for whatever reason our child process would pin an entire core when it was doing absolutely no work.(I still haven't figured out why this happened, we ran a trace and it appeared that epoll was failing on stdout/stdin streams. It would also only happen on our Linux boxes and would work fine on OSX.)

edit:

The pinning of the core was fixed in https://github.com/joyent/libuv/commit/12210fe and was related to https://github.com/joyent/node/issues/5504, so if you run into the issue and you're using centos + kernel v2.6.32: update node, or update your kernel!

Regardless of the issues I had with child_process.fork(), here's a nifty pattern I always use

client:

var child_process = require('child_process');

function FileParser() {

    this.__callbackById = [];
    this.__callbackIdIncrement = 0;
    this.__process = child_process.fork('./child');
    this.__process.on('message', this.handleMessage.bind(this));

}

FileParser.prototype.handleMessage = function handleMessage(message) {

    var error = message.error;
    var result = message.result;
    var callbackId = message.callbackId;
    var callback = this.__callbackById[callbackId];

    if (! callback) {
        return;
    }
    callback(error, result);
    delete this.__callbackById[callbackId];

};

FileParser.prototype.parse = function parse(data, callback) {

    this.__callbackIdIncrement = (this.__callbackIdIncrement + 1) % 10000000;
    this.__callbackById[this.__callbackIdIncrement] = callback;
    this.__process.send({
        data: data, // optionally you could pass in the path of the file, and open it in the child process.
        callbackId: this.__callbackIdIncrement
    });

};

module.exports = FileParser;

child process:

process.on('message', function(message) {

    var callbackId = message.callbackId;
    var data = message.data;
    function respond(error, response) {
        process.send({
            callbackId: callbackId,
            error: error,
            result: response
        });
    }
    // parse data..
    respond(undefined, "computed data");
});

We also need a pattern to synchronize the different processes, when each process finishes its task, it will respond to us, and we'll increment a count for each process that finishes, and then call the callback of the Semaphore when we've hit the count we want.

function Semaphore(wait, callback) {

    this.callback = callback;
    this.wait = wait;
    this.counted = 0;

}

Semaphore.prototype.signal = function signal() {
    this.counted++;
    if (this.counted >= this.wait) {
        this.callback();
    }
}

module.exports = Semaphore;

here's a use case that ties all the above patterns together:

var FileParser = require('./FileParser');
var Semaphore  = require('./Semaphore');

var arrFileParsers = [];
for(var i = 0; i < require('os').cpus().length; i++){
    var fileParser = new FileParser();
    arrFileParsers.push(fileParser);
}

function getFiles() {
    return ["file", "file"];
}

var arrResults = [];
function onAllFilesParsed() {
    console.log('all results completed', JSON.stringify(arrResults));
}

var lock = new Semaphore(arrFileParsers.length, onAllFilesParsed);
arrFileParsers.forEach(function(fileParser) {
    var arrFiles  = getFiles(); // you need to decide how to split the files into 1k chunks
    fileParser.parse(arrFiles, function (error, result) {
        arrResults.push(result);
        lock.signal();
    });
});

Eventually I used http://zguide.zeromq.org/page:all#The-Load-Balancing-Pattern, where the client was using the nodejs zmq client, and the workers/broker were written in C. This allowed us to scale this across multiple machines, instead of just a local machine with sub processes.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.