ChannelTask can be terminated - #22
Conversation
ChannelTask can be terminated and the whole worker is closed and removed from the pool. The pool will open a new worker afterwards. One ChannelTask is only allowed per WebWorker, so termination has no effect on other tasks.
Termination only needs read var access, because the controlling moved to a cloneable struct. This allows to terminate the whole webworker from any place at any time.
| ChannelTask::new(channel, result_rx) | ||
| let worker = self.worker.clone(); | ||
| let open_tasks = Rc::clone(&self.open_tasks); | ||
| let on_terminate = Box::new(move || { |
There was a problem hiding this comment.
Blocking: for a direct (non-pool) WebWorker, this terminates the underlying JS worker while the caller still holds the WebWorker handle. Any later run() posts to a dead worker and hangs forever (the oneshot sender inserted into open_tasks by send_request is never resolved). Since ChannelTask::drop auto-terminates, merely dropping an unfinished direct-worker task now silently bricks the worker — previously this was harmless. Suggestion: set a terminated flag on WebWorker and fail fast on subsequent use, or only auto-terminate on drop for pool tasks (where the lease/replacement machinery needs it).
| } | ||
|
|
||
| /// Terminate this worker and fail all tasks currently assigned to it. | ||
| pub(crate) fn terminate(&self) { |
There was a problem hiding this comment.
Blocking: "One ChannelTask is only allowed per WebWorker" is only enforced for pools. On a direct WebWorker, regular tasks can run concurrently with a channel task, and clearing open_tasks here drops their result senders — pending run() futures then panic via expect_throw("WebWorker gone") in send_request. Either enforce exclusivity for direct workers too, or document that terminating a direct-worker channel task fails all in-flight tasks on that worker.
| .expect("WebWorker result sender dropped"); | ||
| from_bytes(&bytes) | ||
| .take() | ||
| .ok_or(TaskError::ResultAlreadyConsumed)?; |
There was a problem hiding this comment.
this variant is unreachable.
result(mut self) consumes self, so result_rx can only be taken once (the Option exists only because Drop is implemented). Consider .expect("result_rx is only taken here") and dropping ResultAlreadyConsumed from the public TaskError enum, so callers don't have to handle an impossible error.
| } | ||
| } | ||
|
|
||
| impl<R> Drop for ChannelTask<R> { |
There was a problem hiding this comment.
Dropping a task whose function already completed (result unconsumed) still kills and replaces a healthy worker. A result_rx.try_recv() check here could avoid the wasted replacement.
| .slots | ||
| .iter() | ||
| .map(|slot| match &*slot.borrow() { | ||
| WorkerSlot::Active { worker, .. } if worker.current_load() == 0 => Some(0), |
There was a problem hiding this comment.
this requires a fully idle worker, so channel tasks queue behind each other instead of sharing workers as before. This is a behavior change for existing users: apps running more concurrent channel tasks than num_workers can now block where they previously worked.
|
Hey, I just revisited the PR today. I'm not yet fully happy with the motivation and the complexity this introduces. |
2 participants
ChannelTask can be terminated and the whole worker is closed and removed from the pool. The pool will open a new worker afterwards. One ChannelTask is only allowed per WebWorker, so termination has no effect on other tasks.
This is needed for cases where the ChannelTask is running complex logic and spawn additional asynchronous functions endlessly or we don't want to answer to request anymore and close the ChannelTask.