# Async Inference (/integrations/rayserve/how-to/async-inference)
## Overview [#overview]
Async inference is the core model of `kubemq-rayserve`. A producer enqueues a task onto a [KubeMQ Queue](/learn/queues) and gets back a `TaskResult` with `PENDING` status **immediately** — the call never blocks on the model. A consumer thread polls the queue, dispatches the task to the registered handler, and writes the outcome into the queue-peek result backend. Clients then poll `get_task_status_sync` until the status transitions to `SUCCESS` or `FAILURE`.
This decoupling is what makes the pattern suitable for ML inference: enqueue is cheap and non-blocking, work is processed out of band by the consumer, and results are durable for the configured TTL so clients can poll at their own pace.
Need a blocking request-response call instead? Use the [Sync Inference](/integrations/rayserve/how-to/sync-inference) model, which returns the handler result inline over a KubeMQ Query rather than via the queue and a poll loop.
## Enqueue and Poll [#enqueue-and-poll]
`enqueue_task_sync(task_name, args, kwargs)` serializes the call, stores a `PENDING` placeholder, sends the queue message, and returns a `TaskResult` straight away. The returned `id` is the handle you poll with.
`enqueue_task_sync` performs the following steps:
**Validate the request.** A non-empty `task_name` is required, and `initialize()` must have been called first — otherwise the adapter raises `ValueError` or `RuntimeError`.
**Generate a task id.** A UUID4 `task_id` is created to identify the task across enqueue, processing, and polling.
**Store `PENDING` before sending.** The placeholder result is written to the backend *before* the queue message is sent, so a fast consumer cannot overwrite a `SUCCESS` result with a late `PENDING` store.
**Send the queue message and count it.** A `QueueMessage` is sent to the configured queue channel and the `tasks_enqueued_total` metric is incremented.
The handler runs on the consumer side; the producer only sees status transitions through the backend. Both `args` and `kwargs` must be JSON-serializable — non-serializable values raise `TypeError` at enqueue time.
```python title="basic_enqueue.py"
"""Enqueue a single task and poll for its result."""
from __future__ import annotations
import os
import time
import uuid
from kubemq_rayserve import KubeMQAdapterConfig, KubeMQTaskProcessorAdapter
BROKER = os.environ.get("KUBEMQ_ADDRESS", "localhost:50000")
def classify(text: str) -> dict:
"""Simulated text classification."""
time.sleep(0.1) # Simulate model inference
return {"label": "positive", "confidence": 0.92}
def main():
channel = f"example-async-{uuid.uuid4().hex[:8]}"
config = KubeMQAdapterConfig(address=BROKER)
adapter = KubeMQTaskProcessorAdapter(config)
class _Cfg:
queue_name = channel
max_retries = 0
failed_task_queue_name = ""
unprocessable_task_queue_name = ""
adapter.initialize(consumer_concurrency=1, task_processor_config=_Cfg())
adapter.register_task_handler(classify, name="classify")
adapter.start_consumer()
try:
print("Enqueuing task...")
result = adapter.enqueue_task_sync("classify", args=["This is great!"])
print(f" task_id={result.id} status={result.status}")
print("Polling for result...")
for i in range(30):
status = adapter.get_task_status_sync(result.id)
print(f" poll {i + 1}: status={status.status}")
if status.status in ("SUCCESS", "FAILURE"):
print(f" result={status.result}")
break
time.sleep(0.5)
else:
print(" Timed out waiting for result.")
finally:
adapter.stop_consumer()
```
The first poll typically returns `PENDING`; once the consumer finishes the handler and stores the outcome, a subsequent poll returns `SUCCESS` with the handler's return value in `status.result`.
A running KubeMQ broker is required (default `localhost:50000`). Start one locally with Docker — port `50000` is the gRPC endpoint the adapter's SDK clients use, and port `9090` exposes the shared HTTP server for connector endpoints such as KEDA scaling:
## Passing Arguments [#passing-arguments]
A task call is `(task_name, args, kwargs)`. Positional arguments go in `args`, keyword arguments in `kwargs`, and you can pass both together. The consumer invokes the handler as `handler(*args, **kwargs)`, so the values map onto the handler signature exactly as a normal Python call would.
`args` and `kwargs` are serialized to JSON before they cross the queue. Use only JSON-serializable values (strings, numbers, booleans, lists, dicts, `None`). Passing a non-serializable object raises `TypeError` at enqueue time.
### Positional args [#positional-args]
```python title="task_with_args.py"
def add(a: int, b: int) -> dict:
return {"sum": a + b}
def concat(first: str, second: str) -> dict:
return {"result": f"{first} {second}"}
# Enqueue with positional arguments
result1 = adapter.enqueue_task_sync("add", args=[3, 7])
result2 = adapter.enqueue_task_sync("concat", args=["hello", "world"])
```
### Keyword kwargs [#keyword-kwargs]
```python title="task_with_kwargs.py"
def greet(name: str, greeting: str = "Hello") -> dict:
return {"message": f"{greeting}, {name}!"}
# All keyword arguments
result1 = adapter.enqueue_task_sync("greet", kwargs={"name": "Alice", "greeting": "Hi"})
# Partial kwargs — the handler default for `greeting` is used
result2 = adapter.enqueue_task_sync("greet", kwargs={"name": "Bob"})
```
### Mixed args and kwargs [#mixed-args-and-kwargs]
```python title="task_with_mixed_args.py"
def search(query: str, max_results: int = 10, language: str = "en") -> dict:
return {
"query": query,
"max_results": max_results,
"language": language,
"results": [f"Result {i} for '{query}'" for i in range(min(3, max_results))],
}
# Positional `query` plus keyword `max_results` and `language`
result = adapter.enqueue_task_sync(
"search",
args=["machine learning"],
kwargs={"max_results": 5, "language": "en"},
)
```
## Polling Strategies [#polling-strategies]
Polling is just a loop over `get_task_status_sync(task_id)` that stops when the status reaches a terminal state (`SUCCESS` or `FAILURE`). How you space those polls is a trade-off between latency and load on the broker. Three patterns cover most needs.
Poll at a fixed short interval. Lowest latency for fast tasks; highest polling overhead.
```python title="result_polling_patterns.py"
def poll_tight_loop(adapter, task_id: str) -> None:
"""Poll every 0.1s, up to 50 attempts."""
for i in range(50):
status = adapter.get_task_status_sync(task_id)
if status.status in ("SUCCESS", "FAILURE"):
print(f" Completed in {i + 1} polls: result={status.result}")
return
time.sleep(0.1)
print(" Timed out.")
```
Start fast, then double the delay up to a cap. Balances quick pickup of short tasks against low overhead for longer ones.
```python title="result_polling_patterns.py"
def poll_exponential_backoff(adapter, task_id: str) -> None:
"""Start at 0.1s, double each time, cap at 2s, give up after 30s."""
start = time.time()
delay = 0.1
max_delay = 2.0
attempt = 0
while True:
attempt += 1
status = adapter.get_task_status_sync(task_id)
if status.status in ("SUCCESS", "FAILURE"):
print(f" Completed in {attempt} polls: result={status.result}")
return
if time.time() - start > 30:
print(" Timed out after 30s.")
return
time.sleep(delay)
delay = min(delay * 2, max_delay)
```
Poll at a fixed interval until a hard deadline. Predictable upper bound on wait time.
```python title="result_polling_patterns.py"
def poll_with_timeout(adapter, task_id: str, timeout: float = 10.0) -> None:
"""Poll every 0.5s until the deadline."""
start = time.time()
attempt = 0
while time.time() - start < timeout:
attempt += 1
status = adapter.get_task_status_sync(task_id)
if status.status in ("SUCCESS", "FAILURE"):
print(f" Completed in {attempt} polls: result={status.result}")
return
time.sleep(0.5)
print(f" Timed out after {timeout}s ({attempt} polls).")
```
For long-running handlers, give the poll loop enough headroom. A handler that sleeps for several seconds will return `PENDING` on every poll until it completes, so size your interval and attempt count accordingly.
```python title="long_running_task.py"
def heavy_computation(data: str) -> dict:
time.sleep(5) # Simulate expensive processing
return {"input": data, "result": "processed", "duration_seconds": 5}
result = adapter.enqueue_task_sync("heavy_computation", args=["large dataset"])
# Poll for ~5 seconds of PENDING before SUCCESS appears
for i in range(30):
status = adapter.get_task_status_sync(result.id)
if status.status in ("SUCCESS", "FAILURE"):
print(f" result={status.result}")
break
time.sleep(1.0)
```
## Batch Enqueue [#batch-enqueue]
To process a batch, enqueue every task in a loop, keep the returned ids, then poll until all of them reach a terminal state. Each task is independent and may complete in any order.
```python title="batch_enqueue.py"
BATCH_SIZE = 5
def score(text: str) -> dict:
time.sleep(0.1)
return {"text": text, "score": 0.85}
# Enqueue N tasks, collecting their ids
task_ids = []
for text in [f"Review {i}" for i in range(BATCH_SIZE)]:
result = adapter.enqueue_task_sync("score", args=[text])
task_ids.append(result.id)
# Poll all ids until every task is done
results = {}
for attempt in range(60):
all_done = True
for task_id in task_ids:
if task_id in results:
continue
status = adapter.get_task_status_sync(task_id)
if status.status in ("SUCCESS", "FAILURE"):
results[task_id] = status
else:
all_done = False
if all_done:
break
time.sleep(0.5)
print(f"Completed {len(results)}/{BATCH_SIZE} tasks.")
```
A single adapter can register multiple handlers and dispatch each task to the right one by `task_name`. This lets one consumer serve several inference functions off the same queue.
```python title="multiple_handlers.py"
adapter.register_task_handler(classify, name="classify")
adapter.register_task_handler(summarize, name="summarize")
adapter.register_task_handler(translate, name="translate")
# Each enqueue routes to its named handler
r1 = adapter.enqueue_task_sync("classify", args=[text])
r2 = adapter.enqueue_task_sync("summarize", args=[text])
r3 = adapter.enqueue_task_sync("translate", args=[text], kwargs={"target_lang": "fr"})
```
If a task names a handler that was never registered, the consumer acks the message and stores no result, so polling continues to read `PENDING`. Register every handler before calling `start_consumer()`.
## Consumer Concurrency [#consumer-concurrency]
`consumer_concurrency` controls how many tasks the consumer processes in parallel. It maps directly to the `max_messages` argument of `receive_queue_messages`, so a higher value pulls a larger batch off the queue per poll and processes it concurrently. In-flight tasks are tracked in a lock-guarded set and surfaced as the `in_flight` metric.
```python title="consumer_concurrency.py"
NUM_TASKS = 10
CONCURRENCY = 5
def slow_handler(task_num: int) -> dict:
time.sleep(1) # Each task takes ~1 second
return {"task_num": task_num, "status": "done"}
# Process up to 5 tasks in parallel
adapter.initialize(consumer_concurrency=CONCURRENCY, task_processor_config=_Cfg())
adapter.register_task_handler(slow_handler, name="slow_handler")
adapter.start_consumer()
# Enqueue 10 one-second tasks; with concurrency=5 they finish in ~2s, not ~10s
task_ids = [
adapter.enqueue_task_sync("slow_handler", args=[i]).id
for i in range(NUM_TASKS)
]
```
With ten one-second tasks and `consumer_concurrency=5`, total wall-clock time is roughly two seconds instead of the ten it would take serially.
## Handler Return Types [#handler-return-types]
Whatever a handler returns is stored verbatim as the task result. `dict`, `list`, `str`, `int`, and `None` are all valid return types, and each round-trips back through `status.result` as the same JSON type.
```python title="handler_return_types.py"
def return_dict() -> dict:
return {"key": "value", "nested": {"a": 1}}
def return_list() -> list:
return [1, 2, 3, "four", 5.0]
def return_str() -> str:
return "hello from handler"
def return_int() -> int:
return 42
def return_none() -> None:
return None
```
A handler returning `None` still produces a `SUCCESS` result with `status.result` set to `None` — that is a successful completion, not a failure.
## Result Expiry [#result-expiry]
Stored results carry an `expiration_in_seconds` equal to `result_expiry_seconds`. The default is `3600` seconds (one hour) and the accepted range is `0` to `86400` seconds (24 hours). Once a result's TTL passes, its channel is empty and `get_task_status_sync` reads `PENDING` again — there is no separate "expired" status.
```python title="result_expiry.py"
# Short-lived results for this demo
config = KubeMQAdapterConfig(
address=BROKER,
result_expiry_seconds=5,
)
result = adapter.enqueue_task_sync("quick_task", args=["test-data"])
# Immediately after completion the result is available
status = adapter.get_task_status_sync(result.id)
print(f" status={status.status} result={status.result}") # SUCCESS
# After the TTL elapses the channel is empty again
time.sleep(6)
status = adapter.get_task_status_sync(result.id)
print(f" status={status.status}") # PENDING — result expired
```
Because an expired result is indistinguishable from a never-stored one, set `result_expiry_seconds` longer than the maximum time a client could wait before its first successful poll. Keep it well within the `0`–`86400` range.
## Cancellation [#cancellation]
An enqueued task can be soft-cancelled with `cancel_task_sync(task_id)`, which overwrites its stored
result with a `CANCELLED` status. The handler still runs to completion on the consumer — only the
producer-visible result changes. See [Cancellation](/integrations/rayserve/how-to/cancellation) for the full walkthrough and the
race-condition caveats.
## Next Steps [#next-steps]