Force quitting application

[imneedle]

Possibility to force quit a Textual application

I developing a Textual application that I am attaching to a bigger main application. The Textual application is quite complex and makes many calls to workers in both Tasks and Threads.

The problem is that the way things currently stand, when I try to quit the Textual application to go back to the main one, the textual application will wait for all its worker threads to quit before completely quitting.

I would like a way to force quit the application and also force quit all its workers.

This Textual thread briefly speaks about the problem I am currently facing.

Example of problem

For an illustration of the problem, consider this application:

import asyncio

from textual.app import App, ComposeResult
from textual.widgets import Static

class DummyApp(App):
    def compose(self) -> ComposeResult:
        yield Static("This is a dummy Test GUI")

    async def on_mount(self) -> None:
        self.call_after_refresh(self._some_very_expensive_task)

    async def _some_very_expensive_task(self) -> None:
        await asyncio.sleep(10)

if __name__ == "__main__":
    app = DummyApp()
    ret_val = app.run()
    print(f"Application exited with {ret_val=!r}")

The problem is that when launching the application, if you quit the application via ^C (ctrl_c), the application will clear from the screen, however, the background task is still running, and you have to hit ^C again to quit the application completely.

I am searching for a way to force quit the application, and all of its worker threads and tasks.

Attempted solution

I tried looking at the documentation but couldn't find anything, so I attempted to look at parts of Textual's code itself.

I tried overriding the action_quit_app function to cancel all the workers via self.workers.cancel_all().

Additionally, I wanted to force empty the message queue with a self._close_messages(False). I'm pretty sure this function is not supposed to be called by the user, but I gave it a shot.

    # inside the DummyApp
    async def action_quit(self) -> None:
        await self._close_messages(False)  # empty message queue ?
        self.workers.cancel_all()  # force cancel all workers
        self.app.exit()  # exit application

Unfortunately, this still doesn't work and why I'm opening this discussion asking for help.

[willmcgugan]

Are all your tasks and threads created using the workers API?

Tasks can be cancelled, and will receive a CancelledError on the next await. Threads can't be forced to cancel like that. You need to implement a mechanism to gracefully exit the thread. You could do this by polling get_current_worker().is_cancelled but there are other options.

Exiting an app when there is concurrency involved can require some additional work. This isn't something exclusive to Textual, or even Python.

Your example could be fixed by adding @work to your very expensive task.

[imneedle]

Hi @willmcgugan

Thanks for the quick answer.

All my tasks and threads are created using the workers API.

However, what I've noticed is that using the @work decorator is usually what seems to be causing the problem. Take this example using @work with a thread worker:

class DummyApp(App):
    def compose(self) -> ComposeResult:
        yield Static("This is a dummy Test GUI")

    def on_mount(self):
        self.call_after_refresh(self._some_very_expensive_task)

    @work(thread=True)
    def _some_very_expensive_task(self):
        from time import sleep
        sleep(10)

When exiting the application via ^C, we face the same problem, and have to wait for the end of the worker.

However, if I don't use @work(thread=True), and instead use a threading.Thread, I don't have to join the threads so I can quit the application immediately. This is quite a dirty solution though:

import threading

class DummyApp(App):
    def compose(self) -> ComposeResult:
        yield Static("This is a dummy Test GUI")

    def on_mount(self):
        self.call_after_refresh(self._some_very_expensive_task)

    def _some_very_expensive_task(self):
        def _expensive_task():
            from time import sleep
            sleep(10)
        thread = threading.Thread(target=_expensive_task)
        thread.start()

if __name__ == "__main__":
    app = DummyApp()
    ret_val = app.run()
    print(f"Application exited with {ret_val=!r}"

In this second case, the call to app.run() actually finishes immediately after ^C. Whereas with the @work decorator, it waits for the end of the worker.

[willmcgugan]

Workers are tied to the app. They are shut down gracefully, rather than just dying with the process, so you have the opportunity to finish writing files etc. This is almost always what you want, but you may have to write additional code to handle that situation.

Your thread in this example is still running when you get to the last print statement. In that short window, it could attempt to do something with the app which is no longer running, and break in some way.

If you want your app to be reliable and robust, it is best practice to notify the thread you want to exit, and then join() until it is finished.

[imneedle]

I understand. That makes sense.

In that case, could you elaborate on

You could do this by polling get_current_worker().is_cancelled but there are other options.

How exactly would that be setup for this small DummyApp ?

Also, you said

Tasks can be cancelled, and will receive a CancelledError on the next await

Just to make sure, this means while the worker is stuck on an await some_task(), if a cancel happens, that task will only see it once it finishes the await to some_task() ?

[willmcgugan]

You would check is_cancelled periodically, and return from the function if it is True.

Your example uses sleep which you probably would avoid in a really app, specifically because it can't be cancelled. If you call sleep(10), it's going to sleep for 10 seconds no matter what. In a more realistic example, you would check for cancelation frequently enough that it would exit quickly when cancelled.

If you really did want to sleep, you could do something like this:

for n in range(10):
    sleep(1)
    if worker.is_cancelled:
        return

[imneedle]

Got it! Makes sense.

Other question I had was about when you said

Tasks can be cancelled, and will receive a CancelledError on the next await

Just to make sure, this means while the worker is stuck on an await some_task(), if a cancel happens, that task will only see it once it finishes the await to some_task() ?

Additionally, you said that

You could do this by polling get_current_worker().is_cancelled but there are other options

Out of curiosity, what could be some "other options" ?

[willmcgugan]

Just to make sure, this means while the worker is stuck on an await some_task(), if a cancel happens, that task will only see it once it finishes the await to some_task() ?

I think it depends on the operation being awaited. But only at or during an await.

Out of curiosity, what could be some "other options" ?

Any synchronization primitives, like Locks and Events.

An approach I like is to coordinate "jobs" with a Queue. You post a None to the queue, which tells the thread to exit.

It all depends on what you are implementing though.

[imneedle]

Thank you.

I like the idea of having the job queue for a worker. From what I understand, each worker has a Queue of jobs that it executes. If you want to kill the thread, you insert None as the next item, and when the worker reads the next item and sees None, it exits.

How exactly would that look like inside a Textual application ? Would I have to re do a lot of work from scratch ? Or is there a way to add this sort of mechanism onto the already existing WorkerManager.

My idea would be something like:

import threading
from functools import partial
from queue import Queue, Empty as QueueEmpty

class MyWorker(threading.Thread):
    def __init__(self, queue: Queue, *args, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self._queue = queue
        self._active = True

    def run(self) -> None:
        while self._active:
            try:
                task = self._queue.get(timeout=1)
                if task is None:
                    break  # stop running
                task()
                self._queue.task_done()
            except QueueEmpty:
                pass  # stop running

    def force_cancel(self) -> None:
        self._active = False
        with self._queue.mutex:
            self._queue.queue.clear()  # remove all pending items
        self._queue.put(None)
        self.join()

    def add_task(self, func, *args, **kwargs) -> None:
        self._queue.put(partial(func, *args, **kwargs))

And then my textual application would use it like this:

from textual.app import App

class DummyApp(App):
    def action_quit(self) -> None:
        self.worker.force_cancel()  # force quit all tasks
        self.app.exit()

    def on_mount(self) -> None:
        self.call_after_refresh(self._run_task)

    def _run_task(self) -> None:
        task_queue = Queue()
        self.worker = MyWorker(task_queue)
        self.worker.start()
        for _ in range(5):
            self.worker.add_task(self._my_task)

    def _my_task(self):
        from time import sleep
        sleep(5)


if __name__ == "__main__":
    app = DummyApp()
    ret_val = app.run()
    print(f"{ret_val=!r}")

This stops at the end of the first _my_task. Is this the way to go or do you suggest another approach ?