User interface for @batched

[cathyzbn]

Describe your changes

Enable batching in modal functions and class methods.

Backward/forward compatibility checks

---

Check these boxes or delete any item (or this section) if not relevant for this PR.

• Client+Server: this change is compatible with old servers
• Client forward compatibility: this change ensures client can accept data intended for later versions of itself

Note on protobuf: protobuf message changes in one place may have impact to
multiple entities (client, server, worker, database). See points above.

---

Changelog

Added support for dynamic batching. Functions or class methods decorated with @modal.batched will now automatically batch their invocations together, up to a specified max_batch_size. The batch will wait for a maximum of wait_ms for more invocations after the first invocation is made. See guide for more details.

@app.function()
@modal.batched(max_batch_size=4, wait_ms=1000)
async def batched_multiply(xs: list[int], ys: list[int]) -> list[int]:
    return [x * y for x, y in zip(xs, xs)]

@app.cls()
class BatchedClass():
    @modal.batched(max_batch_size=4, wait_ms=1000)
    async def batched_multiply(xs: list[int], ys: list[int]) -> list[int]:
        return [x * y for x, y in zip(xs, xs)]

The batched function is called with individual inputs:

await batched_multiply.remote.aio(2, 3)

[mwaskom]

Just a drive-by review with a few comments. There's a lot going on in this PR! Could potentially be easier to discuss with more atomic changes, although I appreciate that this is a complex feature and that it's helpful to test the live code on your branch!

[mwaskom]

Would it ever make sense to leave this unset? e.g. to say "I always want batches of size n?

[cathyzbn]

In that case the function might block forever. Now we have a upper limit of 10 minutes, would it be better to just let the user set it to 10 minutes?

[gongy]

Thanks for the thorough tests. This looks great!