huseyin tugrul buyukisik huseyin tugrul buyukisik - 9 days ago 4
Node.js Question

Callback was already called in async.map

Edit: "threads"+"async" modules are not async(if not thread) safe! It was being called by another async function! Thats why thread running it twice before a single async completed. There was some code leftovers calling same code asynchronously.

Here is some code that gives the error(marked with

ERROR!
):

async.map(threads, function (thread, cb_) {
zz++,
console.log(zz);
(function (id____) {

thread.msg({ cmd: "compare", data: id____ }, function (result) {
if (result)
compareCheck = false;
console.log(cb_.toString());

// ERROR!
cb_(null, thread); // why?
// return cb_ doesn't solve neighter

});
})(id__);

}, function (err, result) {

if (compareCheck) {
console.log("b:" + Session.cmpCtr);

threads[Session.cmpCtr % nThreads].msg({
cmd: "add", data: id__
}, function (response) {
setImmediate( cbCmp_);
});
}
else {

// whatsoever gives collision, retry new one,
// setImmediate to stop recursion fee on cpu
setImmediate(function () {

id__ = "";
for (var i = 0; i < 24; i++)
id__ += rndChar();
cmpId(id__, cbCmp_);
});
}
});


what this code doing is, checking if N threads (using threads module with spawn) have a session variable in their own list equal to a newly generated one then if there is no collision, adding the new variable to one of the threads list. If command to a thread is "add", then it adds variable to its list, if command is "compare" then it gives true false value on callback depending on list's values and new value.

Why async.map gives this error? That function is executed by all threads, why shouldn't it be executed more than once? It is a callback.

Maybe async.map cannot have a thread list (which is not serializable) and just copies it to have more multiple instances at each session variable generation?

Threads is being used by a wrapper that is initialized with:

threads.push(new QThread(function (input, done) {
this.data = (typeof this.data === "undefined")? [] : this.data;
if (input.cmd != null && typeof input.cmd !== "undefined") {
if (input.cmd === "add") {
data.push(input.data);
done(true);
}
else if (input.cmd === "compare")
{
for (var i = 0; i < data.length; i++) {
if (data[i] == input.data) {
done(true);
return;
}
}

done(false);
return;
}
}
}));


Edit:
return cb_(null, thread);
gives same error.

I want to use async.map because using a simple spin-lock style synchronization is cpu resource intensive and not good for nodejs.

Here is the error from output:

C:\...\async.js:985
if (fn === null) throw new Error("Callback was already called.");
^

Error: Callback was already called.
at C:\...\async.js:985:32
at C:\...\async.js:1158:13
at C:\...\session.js:79:28
at Worker.<anonymous> (C:\...\qthread.js:25:13)
at Worker.emit (C:\...\index.js:129:35)
at Worker.handleMessage (C:\Users\pc\node_modules\threads\lib\worker.node\worker.js:119:17)
at emitTwo (events.js:106:13)
at ChildProcess.emit (events.js:191:7)
at process.nextTick (internal/child_process.js:744:12)
at _combinedTickCallback (internal/process/next_tick.js:67:7)

Answer

After the cb_(null, thread); is called, the function still executes the remaining code. It is better to terminate callbacks something like this

return cb_(null, thread);

This ensures that functions calls the callback, stops execution and returns.