Task timeout functionallity

[ehsanghorbani190]

I decided to open a separate issue for this problem. A good feature to have is task timeout functionality. The signal option provided has a logical bug, that it applies to tasks that are in queue. But the correct logic is that it applies to the task after it starts to run. That was the nearest thing I found to the task timeout in piscina. Currently, I'm trying something with the messagedChannels to abort the controller of a task when I get a message. But I've got no results so far. I need to make sure that each task in the running phase doesn't take more than 10 seconds. Any ideas on this?

[ehsanghorbani190]

Update: I've managed to achieve what i want with AbortController and MessagedChannel. This is my code:

export default function getAbortChannel() {
  const channel = new MessageChannel();
  const controller = new AbortController();

  channel.port2.on("message", (message) => {
    if (message === 504) {
      controller.abort();
    }
    channel.port2.close();
  });

  const { port1, port2 } = channel;
  const signal = controller.signal;

  return { port1, port2, signal };
}

i close the port2 at my spawning function and port1 at my worker.
Although my functionality works, it's not efficient as I'm creating a MessagedChannel and AbortControlller for each task. the proposed timeout functionality on the returned promise of run function could be more useful.

[metcoder95]

What you say is true, currently the signal is attached to the task regardless of whether has been executed or enqueued; this is as it is complicated to fully know the moment at which the task will start to be executed given the workloads at hand.

AbortSignal.timeout will effectively account for the time the task is sitting idle in the queue.

A task timeout can come handy, at the Piscina#run can bring good ergonomics. Would you like to send a PR for it?

[ehsanghorbani190]

It would be an honor to do this my self :), but I don't know when I'll have time to do it. Also, my current implementation of AbortChannel is not useful enough.
I don't know if I got this right or not as I'm not so experienced in open source projects :), but I should do the task and make a PR, right? Or can I open a PR for anyone to do it ?

[metcoder95]

As you wish; there's no rush into have it done quickly, if starts getting traction either someone or myself can take it over later.

You can start with a PR that attempts to implement it and we take it from there :)

[ehsanghorbani190]

@metcoder95 Great! I'll open an PR later this day or tommorw :D

[ehsanghorbani190]

@metcoder95 I've found a bug in the Pool.run function. When testing my AbortChannel, I found out that no matter what the value of maxQueue is, no task would become queued! the problem is from the signal option of pool.run. After removing the signal, tasks are queued correctly.

[metcoder95]

That's intentional; when a task contains an abort signal, is offloaded to a different queue when not processed right away, this as currently when a task is aborted we do not communicate this directly to the worker but rather we immediately teardown the worker.

For instance, Piscina attempts that, if a task is abortable, a worker should just handle only that task so it can be preempted if task is aborted (this helps cleaning up possible tangling resources, etc.).

Is this giving wrong numbers for the tasks enqueued?

[ehsanghorbani190]

@metcoder95
Thanks for details. I think that makes sense for why this is working like this. But the problem is, in this situation, queueSize is not updated at all. If the number of tasks that are queued are not accessible through the queueSize property, what's the meaning of even setting the maxQueue option? This would make backpressure handling very hard.

[metcoder95]

Yeah, I'm accounting for that in #872