Apps typically process requests until completion. However, users may cancel long-running tasks. Implementing graceful cancellation ensures resources are released immediately and partial results are handled correctly.

The on_cancel Hook

Define an async on_cancel method in your App class. This method is called when the supervisor receives a cancellation signal.

1class App(BaseApp):2    async def setup(self, config):3        self.cancel_flag = False4 5    async def on_cancel(self):6        """Called when user cancels the task"""7        print("Cancellation requested...")8        self.cancel_flag = True9        10        # Return True to confirm you received the signal11        return True12 13    async def run(self, input):14        self.cancel_flag = False15        16        # Long running loop17        for i in range(100):18            if self.cancel_flag:19                print("Stopping work...")20                # Clean up resources if needed21                break22                23            await self.heavy_computation(i)

Best Practices

1. Check Frequently: In loops, check your cancellation flag at the start of every iteration.
2. Clean Up: Close database connections, delete temporary files, or free GPU memory before exiting.
3. Return Quickly: The on_cancel handler should be fast. Do note block execution there; just set a flag or signal an event.
4. Force Kill: If an app does not respond to on_cancel within a timeout period (default 30s), it will be forcefully terminated (SIGKILL).