memory leak with worker thread and parentPort.postMessage

[8HoLoN]

Version

v17.4.0

Platform

Microsoft Windows NT 10.0.22000.0 x64

Subsystem

No response

What steps will reproduce the bug?

node test.js

• test.js :

const {
    Worker, isMainThread, parentPort
} = require('worker_threads');

if (isMainThread) {
    function parseJSAsync(script) {
        return new Promise((resolve, reject) => {
            const worker = new Worker(__filename);
            //worker.on('message', resolve);
            worker.on('message', (i)=>{
                //console.log('plop' + i);
            });
            worker.on('error', reject);
            worker.on('exit', (code) => {
                if (code !== 0)
                    reject(new Error(`Worker stopped with exit code ${code}`));
            });
        });
    };
    parseJSAsync()// no leak if called only once
    parseJSAsync()// leak appears if called twice or more
    parseJSAsync()
} else {
    let i = 0;
    while (i<1000000000000) {
        parentPort.postMessage(i);// no leak if commented
        i++;
    }
}

How often does it reproduce? Is there a required condition?

Always, if more than 1 worker is instantiated and parentPort.postMessage is used.

What is the expected behavior?

constant memory usage

What do you see instead?

a linear increase of memory usage over time (up to 16Gio+ in less than 1min)

Additional information

No response

[addaleax]

It sounds like you are calling postMessage faster than the receiving thread can handle messages. In that case, memory growth doesn't seem like unexpected behavior to me.

[8HoLoN]

hm, it's unexpected to me because how can the worker know that you are calling postMessage faster than the receiving thread can handle messages. Is there any queue length to check within the worker, like worker.postMessageQueueLength ?

[addaleax]

@8HoLoN No, I don't think the Worker can know that for sure, i.e. you need some kind of active collaboration from the receiving side if you want to keep track of the number of outstanding messages. (It's definitely possible though -- as an example, you could share a SharedArrayBuffer containing a counter of the outstanding messages, and use Atomics.add() to increase the counter when you queue an additional message/decrease it when you receive and process a new message.)

[8HoLoN]

@addaleax ok, and use Atomics.compareExchange because value could change between a if(Atomics.load(lock, 0) === 0) and Atomics.add(lock, 0, 1) ?

const lockSnap = Atomics.load(lock, 0);
if (lockSnap === 0) {
    const oldSnap = Atomics.compareExchange(lock, 0, lockSnap, lockSnap + 1);
    if (oldSnap === lockSnap) {
        parentPort.postMessage({ i, id });
    }
}

instead of

if (Atomics.load(lock, 0) === 0) {
    Atomics.add(lock, 0, 1);
    parentPort.postMessage({ i, id });
}