Async Inference
Enqueue inference tasks onto a KubeMQ Queue and poll for results — the core asynchronous model.
Overview
Async inference is the core model of kubemq-rayserve. A producer enqueues a task onto a KubeMQ Queue 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 model, which returns the handler result inline over a KubeMQ Query rather than via the queue and a poll loop.
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.
"""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:
docker run -d \ --name kubemq \ -p 50000:50000 \ -p 9090:9090 \ -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \ europe-docker.pkg.dev/kubemq/images/kubemq:nextPassing 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
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
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
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 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.
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.
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.
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.
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
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.
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.
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 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.
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
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.
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 NoneA handler returning None still produces a SUCCESS result with status.result set to None — that is a successful completion, not a failure.
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.
# 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 expiredBecause 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
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 for the full walkthrough and the
race-condition caveats.
Next Steps
Was this page helpful?