[Bugfix] Fix request cancellation without polling

[joerunde]

See discussion and alternate solution in #11096

I came to agree with @jakkdl in encode/starlette#2094 and the linked issues that polling for disconnects with request.is_disconnected() introduces more problems than it's worth. Instead, we can use the pattern that's already in StreamingResponse and have a separate async task wait for a disconnect message, cancelling our work if one is received. The key here is that a StreamingResponse is able to safely consume all new messages because the request body has already been read. Our request handlers have the same guarantee, since fastapi first reads and parses the request and builds a pdydantic object for us before invoking our handler.

This PR implements a decorator for our fastapi handlers that will cancel them if a disconnect message is received while they are running. This is implemented with asynco directly instead of with anyio, because the rest of the code base assumes asyncio.

The advantages here are:

• No extra polling overhead and need for configuring timeouts and poll intervals
• Work can be canceled as soon as a disconnect happens, instead of at the next disconnect check
• This is completely transparent to our code, and we can remove the existing cancelation handling that's coupled inside the server logic

Disadvantages

• This doesn't work with entrypoints/api_sever.py because that handler reads the request body itself, so our one cancellation test in tests/async_engine/test_api_server.py fails. I'll need to write a new one, but it's 6pm on a Friday 🙃

I manually verified that this works to cancel both streaming and non-streaming requests, let me know what y'all think of doing this instead

FIX #10087

[github-actions]

👋 Hi! Thank you for contributing to the vLLM project.
Just a reminder: PRs would not trigger full CI run by default. Instead, it would only run fastcheck CI which starts running only a small and essential subset of CI tests to quickly catch errors. You can run other CI tests on top of those by going to your fastcheck build on Buildkite UI (linked in the PR checks section) and unblock them. If you do not have permission to unblock, ping simon-mo or khluu to add you in our Buildkite org.

Once the PR is approved and ready to go, your PR reviewer(s) can run CI to test the changes comprehensively before merging.

To run CI, PR reviewers can do one of these:

• Add ready label to the PR
• Enable auto-merge.

🚀

[joerunde]

Turns out the old async llm engine needed to check for cancellation errors for this new method to work, so I shuffled around the base api_server.py to hook up the new decorator so our test with it + the async llm engine works again.

Working now on a test for the openai api server, that doesn't necessarily need to block merging if we're on a tight schedule but should be in soon 🤞

@mgoin @simon-mo Can I get a ready on this to get the full CI suite running?

[joerunde]

Okay, the test for request cancellation with the openai server is in as well. It overloads the server with a ton of requests and then cancels them, ensuring the server can still respond after. I would have rather been able to do something like check the /metrics output for aborted requests, but those aren't currently tracked by our metrics.

[mgoin]

Even though it does have the potential to be a footgun (as you documented), I like the usage of @with_cancellation so LGTM pending green!

[joerunde]

Ah shoot, the test was too much for the smaller cards we run on in CI and it failed even though the logs do show plenty of requests being aborted :(

I'll dial it back some- I don't want this to be flaky so I think fewer requests with more tokens each would be better at piling up load without burdening the server with handling so many aborts