Api allows workers to do initialization but it is not clear how to do cleanup

[bhovhannes]

Hi there,

Docs says that it is possible to do initialization inside worker. I am using that feature to create a database connection like this:

async function initialize() {
  const conn = await connectToDb();
  return (query) => conn.exec(query);
}

module.exports = initialize();

During its lifetime, pool may decide to close inactive threads. In that case I need to close database connection.
But currently I didn't found any api which allows to do that.

Currently I set minThreads and maxThreads to the same value and have the following as a workaround:

const Piscina = require("piscina")

const workerCount = 10;

const pool = new Piscina({
    filename: __dirname + "/searchMessagesWorker.cjs",
    minThreads: workerCount,
    maxThreads: workerCount,
    idleTimeout: Integer.MAX_SAFE_INTEGER
})

// this will be set to `false` each time user calls `pool.runTask`
let poolIsEmpty = true
pool.on('drain', () => {
    poolIsEmpty = true
})
    
function shutdown() {
    // drainPromise will be resolved as soon pool will be drained
    const drainPromise = new Promise((resolve) => {
        if (poolIsEmpty) {
           resolve()
        }
        else {
            pool.once('drain', resolve)
        }
    })

    return drainPromise.then(
        () => {
            // no tasks are scheduled on the pool at this moment.
            // Schedule "workerCount" amount "close" tasks in a single loop,
            // assuming Piscina will assign a single task to each thread.
            const promises = []
            for (let i = 0; i < workerCount; ++i) {
                // this tells worker to close DB connection
                promises.push(pool.runTask({
                    type: 'close'
                }))
            }
            return Promise.all(promises)
        }
    ).then(
        () => pool.destroy()
    )
}

Not only it is complicated, but also fragile (depending on how Piscina schedules tasks under the hood, setting idleTimeout to very high number). Also, this way it is not possible to have dynamic number of threads in the pool.

Interesting, but cleanup functionality is something which is missing in all worker pool implementations I can find on npm. I am wondering why?
Maybe there is a completely different way to do cleanup inside worker thread?

I'll be grateful for any explanation/reading resource on this topic.
If this is something which is in scope of Piscina package I am ready to work on PR as well.

[jasnell]

Cleanup is difficult because it is possible for the thread the worker is running on to be terminated abruptly. And it can be difficult to reason about when the cleanup logic should run.

@addaleax ... what do you think? Understanding that the mechanism would not be perfect, a mechanism to send a message to each worker that it should run cleanup logic after it completes it's current job (if any) should be possible.

[bhovhannes]

@jasnell thank you for quick reply!

Cleanup is difficult because it is possible for the thread the worker is running on to be terminated abruptly.

Are you meaning process.exit in the main thread or some kind of abnormal program termination?
I am more concerned about memory leaks during normal program flow.

And it can be difficult to reason about when the cleanup logic should run.

What do you mean? Inside message pool we know when we're going to terminate a thread.

What if at WorkerInfo.destroy(https://github.com/piscinajs/piscina/blob/current/src/index.ts#L368), just before this.worker.terminate(); line we'll mark thread as "being terminated" and then postMessage to it saying "cleanup yourself".
Thread pool should skip threads in "being terminated" status during assigning tasks.

[addaleax]

Cleanup is difficult because it is possible for the thread the worker is running on to be terminated abruptly.

Are you meaning process.exit in the main thread or some kind of abnormal program termination?
I am more concerned about memory leaks during normal program flow.

It’s referring to worker.terminate(), which is equivalent to calling process.exit() inside the Worker thread.

And it can be difficult to reason about when the cleanup logic should run.

What do you mean? Inside message pool we know when we're going to terminate a thread.

What if at WorkerInfo.destroy(https://github.com/piscinajs/piscina/blob/current/src/index.ts#L368), just before this.worker.terminate(); line we'll mark thread as "being terminated" and then postMessage to it saying "cleanup yourself".
Thread pool should skip threads in "being terminated" status during assigning tasks.

That would change the behavior here significantly, because if we allow graceful shutdown, then terminating the Worker thread is no longer an option, and we would have to rely on it cleaning up properly, even after an uncaught exception.

But I’m not 100 % clear on what your concern is here. If you’re establishing database connections – network sockets are shut down when a Worker that owns them is being terminated, so there’s probably more to it than that?

[bhovhannes]

Thanks for explanation!

But I’m not 100 % clear on what your concern is here. If you’re establishing database connections – network sockets are shut down when a Worker that owns them is being terminated, so there’s probably more to it than that?

I don't have deep knowledge of which resources clean automatically and which one don't in worker threads, so yes, maybe cleanup is not that necessary.
In my case I am opening sqlite connection, which I assume result in opening file under the hood (because sqlite is file-based). Can I be sure that not closing connection in that case will not cause any issues in a long run?

[addaleax]

So … yeah. Files are a more complicated topic: The FileHandle API, i.e. what fs.promises.open() returns, does clean up automatically, but when working with raw file descriptors, i.e. what fs.open() returns, Node.js can’t perform automatic cleanup because those values are just raw integers.

So yes, maybe it makes sense to provide some explicit cleanup hook here, even if that changes the behavior a bit.

[jasnell]

I definitely think we should implement something but we have to be explicit that the semantics are Best Effort in that we cannot guarantee that the cleanup logic will run every time for every worker instance. Sometimes it will, other times it won't.

For something like a raw file descriptor, I'd expect that it would be better for the main thread to handle those regardless of what we do here. Specifically, I'm wondering if there's a way in core (not in piscina) that we could track raw file descriptors opened in a worker thread and have those automatically closed (if necessary) or reported with a warning on the main thread after the worker terminates.

[addaleax]

Here’s a PR to implement that: nodejs/node#34303

I think we could disable that in Piscina by default without further issues, and maybe also flip the default in Node.js in a semver-major follow-up.

[jasnell]

Just an update on this: We've landed and shipped the support for tracking unmanaged FDs.

In Node.js core, we'll hopefully soon land an implementation of BroadcastChannel(nodejs/node#36271) which can be used to signal a global cleanup.