I have a module in Node.js which repeatedly pick a document from MongoDB and process it. One document should be processed only once. I also want to use multiple processes concept. I want to run the same module(process) on different processors, which run independently.
The problem is, there might be a scenario where the same document picked and processed by two different workers. How multiple processes can know that, a particular document is processed by some other worker so I should not touch it. And there is no way that my independent processes can communicate. I cannot use a parent which forks multiple processes and acts as a bridge between them. How to avoid this kind of problems in Node.js?
One way to do it is to assign an unique numeric ID to each of your MongoDB documents, and to assign an unique numeric identifier to each of your node.js workers.
For example, have an env var called NUM_WORKERS, and then in your node.js module:
var NumWorkers = process.env.NUM_WORKERS || 1;
You then need to assign an unique, contiguous instance number id (in the range 0 to NumWorkers-1) to each of your workers (e.g. via a command line parameter read by your node.js process when it initializes). You can store that in a variable called MyWorkerInstanceNum.
When you pick a document from MongoDB, call the following function (passing the document's unique documentId as a parameter):
function isMine(documentId){
//
// Example: documentId=10
// NumWorkers= 4
// (10 % 4) = 2
// If MyWorkerInstanceNum is 2, return true, else return false.
return ((documentId % NumWorkers) === MyWorkerInstanceNum);
}
Only continue to actually process the document if isMine() returns true. So, multiple workers may "pick" a document, but only one worker will actually process it.
Simply keep a transaction log of the document being processed by its unique ID. In the transaction log table for the processed documents, write the status as one of the following (for example):
requested
initiated
processed
failed
You may also want a column in that table for stderr/stdout in case you want to know why something failed or succeeded, and timestamps - that sort of thing.
When you initialize the processing of the document in your Node app, look up the document by ID and check its status. If it doesn't exist, then you're free to process it.
Pseudo-code (sorry, I'm not a Mongo guy!):
db.collection.list('collectionName', function(err, doc) {
db.collection.find(doc.id, 'transactions', function(err, trx) {
if (trx === undefined || trx.status === 'failed') {
DocProcessor.child.process(doc)
} else {
// don't need to process it, it's already been done
}
})
})
You'll also want to enable concurrency locking on the transactions log collection so that you ensure a row (and subsequent job) can't be duplicated. If this becomes a challenge to ensure docs are being queued properly, consider adding in an AMQP service to handle queuing of the docs. Set up a handler to manage distribution of the child processes and transaction logging. Flow would be something like:
MQ ⇢ Log ⇢ Handler ⇢ Doc processor children