How to Handle Concurrent Requests in Local Model Serving

How to Handle Concurrent Requests in Local Model Serving

When you deploy a machine learning model locally for inference, the first request usually works flawlessly. The trouble begins when multiple requests arrive at the same time. Without a deliberate concurrency strategy, your model server becomes a bottleneck: requests queue up, latency spikes, and under heavy load the process may even crash from memory exhaustion. This tutorial walks through what concurrency means in the context of local model serving, why it matters, and how to implement it correctly with practical code examples.

What Is Concurrent Request Handling?

Concurrency in model serving refers to the ability of a server process to handle multiple inference requests overlapping in time, rather than processing them strictly one after another. There are several flavors of concurrency you should distinguish:

• Sequential processing: Requests are handled one at a time. Simple but slow under load.
• Asynchronous I/O: The server accepts many connections and switches between them while waiting on I/O, but inference itself still runs on a single thread.
• Parallel inference: Multiple worker threads or processes execute the model simultaneously, leveraging multiple CPU cores or GPU streams.
• Batched inference: Incoming requests are grouped into a single batch and processed together by the model, dramatically improving throughput on GPUs.

The right approach depends on your hardware, model size, and latency requirements. A small scikit-learn model on CPU benefits from thread pools, while a large transformer on GPU benefits from dynamic batching.

Why Concurrency Matters for Local Serving

Local model serving has unique constraints compared to cloud deployments. You typically have a single machine with finite CPU cores, limited RAM, and possibly one GPU. Every concurrent request consumes memory for input tensors, intermediate activations, and output buffers. Without controls, a burst of traffic can exhaust resources and bring the whole service down.

Concurrency also directly affects user experience. If your chatbot endpoint takes 200ms per request and processes sequentially, ten simultaneous users will see latencies ranging from 200ms to 2 seconds. With proper concurrency, all ten can receive responses near the 200ms mark. Finally, well-managed concurrency improves hardware utilization — a GPU running one inference at a time may only use 30% of its compute capacity, while batching can push that above 90%.

Setting Up a Basic Sequential Server

Before improving concurrency, let's establish a baseline. Here is a minimal FastAPI server that loads a model and handles one request at a time:

from fastapi import FastAPI
from pydantic import BaseModel
import numpy as np
import time

app = FastAPI()

# Simulate a loaded model
class DummyModel:
    def predict(self, features: np.ndarray) -> np.ndarray:
        time.sleep(0.2)  # simulate inference latency
        return (features.sum(axis=1, keepdims=True) > 0.5).astype(int)

model = DummyModel()

class InputData(BaseModel):
    features: list[list[float]]

class OutputData(BaseModel):
    prediction: list[int]

@app.post("/predict", response_model=OutputData)
def predict(data: InputData):
    features = np.array(data.features, dtype=np.float32)
    result = model.predict(features)
    return OutputData(prediction=result.tolist())

This works, but if you fire ten concurrent requests with a tool like locust or ab, you will see them complete one after another. FastAPI runs sync endpoints in a threadpool by default, which gives some concurrency, but our model.predict call blocks each thread. Let's fix that properly.

Approach 1: Thread Pool for CPU-Bound Models

For CPU-bound models (most scikit-learn, XGBoost, and small PyTorch models), a thread pool or process pool is the simplest concurrency strategy. Python's Global Interpreter Lock (GIL) is released during many numerical operations in NumPy, scikit-learn, and native extensions, so threads can achieve real parallelism for inference workloads.

import concurrent.futures
from fastapi import FastAPI
from pydantic import BaseModel
import numpy as np
import time

app = FastAPI()

class DummyModel:
    def predict(self, features: np.ndarray) -> np.ndarray:
        time.sleep(0.2)
        return (features.sum(axis=1, keepdims=True) > 0.5).astype(int)

model = DummyModel()

# Limit workers to avoid overwhelming CPU and memory
MAX_WORKERS = 4
executor = concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS)

class InputData(BaseModel):
    features: list[list[float]]

class OutputData(BaseModel):
    prediction: list[int]

@app.post("/predict", response_model=OutputData)
async def predict(data: InputData):
    features = np.array(data.features, dtype=np.float32)
    # Offload blocking inference to the thread pool
    result = await asyncio_loop().run_in_executor(
        executor, model.predict, features
    )
    return OutputData(prediction=result.tolist())

def asyncio_loop():
    import asyncio
    return asyncio.get_event_loop()

By making the endpoint async and using run_in_executor, the main event loop stays free to accept new connections while inference runs in a worker thread. The MAX_WORKERS cap prevents resource exhaustion. Tune this value based on your CPU core count and model memory footprint.

Approach 2: Process Pool for GIL-Bound Models

If your model code is pure Python and does not release the GIL (for example, a custom inference loop with heavy Python logic), threads will not give you true parallelism. In that case, use a process pool instead:

import concurrent.futures
from fastapi import FastAPI
from pydantic import BaseModel
import numpy as np

app = FastAPI()

# The model must be loadable in each worker process.
# Use a module-level loader function so each process initializes its own copy.
def load_model():
    import time
    class DummyModel:
        def predict(self, features):
            time.sleep(0.2)
            return (features.sum(axis=1, keepdims=True) > 0.5).astype(int)
    return DummyModel()

# ProcessPoolExecutor requires picklable callables.
# Wrap inference in a top-level function.
def run_inference(features: np.ndarray) -> np.ndarray:
    global _model
    if _model is None:
        _model = load_model()
    return _model.predict(features)

_model = None

executor = concurrent.futures.ProcessPoolExecutor(max_workers=4)

class InputData(BaseModel):
    features: list[list[float]]

class OutputData(BaseModel):
    prediction: list[int]

@app.post("/predict", response_model=OutputData)
async def predict(data: InputData):
    import asyncio
    features = np.array(data.features, dtype=np.float32)
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(executor, run_inference, features)
    return OutputData(prediction=result.tolist())

Process pools have higher overhead because data must be serialized and sent to worker processes. They are best when inference time is large relative to serialization cost. For small, fast models, stick with threads.

Approach 3: GPU Inference with Batching

GPUs are most efficient when processing batches of inputs together. Handling concurrent requests on a GPU therefore means collecting incoming requests, grouping them into a batch, running a single forward pass, and splitting the results back to each caller. This is called dynamic batching.

Here is a simplified implementation using an asyncio queue and a background batching loop:

import asyncio
import time
from fastapi import FastAPI
from pydantic import BaseModel
import numpy as np

app = FastAPI()

# Configuration
MAX_BATCH_SIZE = 16
MAX_BATCH_WAIT = 0.01  # 10ms

class BatchInferenceServer:
    def __init__(self):
        self.queue: asyncio.Queue = asyncio.Queue()
        self.model = None  # load your model here
        self._task = None

    def start(self):
        self._task = asyncio.create_task(self._batch_loop())

    async def stop(self):
        if self._task:
            self._task.cancel()

    async def submit(self, features: np.ndarray) -> np.ndarray:
        future = asyncio.get_event_loop().create_future()
        await self.queue.put((features, future))
        return await future

    async def _batch_loop(self):
        while True:
            batch = []
            futures = []
            deadline = time.monotonic() + MAX_BATCH_WAIT

            # Collect requests up to MAX_BATCH_SIZE or until deadline
            while len(batch) < MAX_BATCH_SIZE:
                timeout = deadline - time.monotonic()
                if timeout <= 0:
                    break
                try:
                    features, future = await asyncio.wait_for(
                        self.queue.get(), timeout=timeout
                    )
                    batch.append(features)
                    futures.append(future)
                except asyncio.TimeoutError:
                    break

            if not batch:
                continue

            # Stack into a single batch tensor
            batched = np.stack(batch, axis=0)

            # Run inference (replace with your model forward pass)
            results = await self._infer(batched)

            # Distribute results back to each waiting request
            for fut, result in zip(futures, results):
                if not fut.done():
                    fut.set_result(result)

    async def _infer(self, batch: np.ndarray) -> np.ndarray:
        # Simulate GPU inference latency
        await asyncio.sleep(0.05)
        return (batch.sum(axis=1, keepdims=True) > 0.5).astype(int)

server = BatchInferenceServer()

class InputData(BaseModel):
    features: list[float]

class OutputData(BaseModel):
    prediction: int

@app.on_event("startup")
async def startup():
    server.start()

@app.on_event("shutdown")
async def shutdown():
    await server.stop()

@app.post("/predict", response_model=OutputData)
async def predict(data: InputData):
    features = np.array(data.features, dtype=np.float32)
    result = await server.submit(features)
    return OutputData(prediction=int(result[0]))

This pattern is the foundation of production servers like NVIDIA Triton and TorchServe. The key parameters are MAX_BATCH_SIZE (how many requests to group) and MAX_BATCH_WAIT (how long to wait for a full batch). Larger batches improve throughput but add latency for the first request in the batch. Tune these based on your latency SLA and throughput goals.

Approach 4: Semaphore-Based Rate Limiting

Sometimes you cannot batch or parallelize easily — for example, a large language model that fills GPU memory with a single request. In that case, you still need concurrency control to prevent overload. An asyncio.Semaphore limits how many requests can be in-flight simultaneously:

import asyncio
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import time

app = FastAPI()

MAX_CONCURRENT = 2  # only 2 inferences at once
semaphore = asyncio.Semaphore(MAX_CONCURRENT)

class InputData(BaseModel):
    text: str

class OutputData(BaseModel):
    result: str

async def heavy_inference(text: str) -> str:
    await asyncio.sleep(1.0)  # simulate LLM generation
    return f"processed: {text}"

@app.post("/generate", response_model=OutputData)
async def generate(data: InputData):
    # Acquire semaphore with a timeout to avoid indefinite waiting
    try:
        async with asyncio.timeout(30):
            acquired = await semaphore.acquire()
    except asyncio.TimeoutError:
        raise HTTPException(status_code=503, detail="Server busy, try again later")

    try:
        result = await heavy_inference(data.text)
        return OutputData(result=result)
    finally:
        semaphore.release()

When the semaphore is exhausted, additional requests wait. The timeout ensures they fail fast with a 503 instead of hanging forever. Clients can retry with exponential backoff.

Best Practices

Right-Size Your Worker Count

More workers is not always better. Each worker holds memory for model weights (in the case of threads, shared; in the case of processes, duplicated) and intermediate tensors. For CPU inference, a good starting point is min(cpu_count, 4 * num_cpu_cores) for I/O-bound workloads and num_cpu_cores for pure compute. Monitor CPU utilization and memory, then adjust.

Use Bounded Queues

Unbounded queues let latency grow without limit under sustained load. Set a maximum queue depth and reject or shed load when full. This keeps latency predictable even if throughput drops:

MAX_QUEUE_DEPTH = 100

@app.post("/predict")
async def predict(data: InputData):
    if server.queue.qsize() >= MAX_QUEUE_DEPTH:
        raise HTTPException(status_code=429, detail="Too many requests")
    result = await server.submit(np.array(data.features, dtype=np.float32))
    return OutputData(prediction=int(result[0]))

Measure Before You Optimize

Concurrency strategies have trade-offs. Measure latency percentiles (p50, p95, p99) and throughput under realistic load before and after each change. Tools like locust, wrk, or vegeta can generate concurrent traffic. A common mistake is optimizing for maximum throughput while ignoring tail latency, which is what users actually feel.

Handle Model Thread Safety

Not all models are thread-safe. Some PyTorch models mutate internal state during inference, and sharing a single model instance across threads can produce wrong results or crashes. If you are unsure, either use a process pool (each process has its own model copy) or maintain a pool of model instances, one per worker thread:

import threading

class ModelPool:
    def __init__(self, factory, size=4):
        self._models = [factory() for _ in range(size)]
        self._lock = threading.Lock()
        self._available = threading.Semaphore(size)

    def acquire(self):
        self._available.acquire()
        with self._lock:
            return self._models.pop()

    def release(self, model):
        with self._lock:
            self._models.append(model)
        self._available.release()

Consider Dedicated Serving Frameworks

For production workloads, consider purpose-built serving frameworks that handle concurrency for you:

• NVIDIA Triton Inference Server: Supports dynamic batching, multi-model serving, and GPU concurrency out of the box.
• TorchServe: PyTorch's official serving library with built-in batching and worker management.
• vLLM: Optimized for LLM serving with continuous batching and PagedAttention.
• Ray Serve: Scales from local to distributed deployments with autoscaling and request routing.

Building your own concurrency layer is valuable for learning and for simple use cases, but dedicated frameworks handle edge cases like backpressure, graceful shutdown, and model warm-up that are tedious to implement correctly.

Implement Graceful Shutdown

When stopping your server, in-flight requests should be allowed to complete rather than being killed abruptly. Cancel background tasks, drain queues, and wait for workers to finish:

@app.on_event("shutdown")
async def shutdown():
    # Stop accepting new batch items
    await server.stop()
    # Wait for in-flight requests with a deadline
    deadline = time.monotonic() + 10
    while server.queue.qsize() > 0 and time.monotonic() < deadline:
        await asyncio.sleep(0.1)
    executor.shutdown(wait=True, cancel_futures=False)

Conclusion

Handling concurrent requests in local model serving is a balancing act between throughput, latency, and resource usage. The right strategy depends on your model and hardware: thread pools for CPU-bound numerical models, process pools for GIL-bound logic, dynamic batching for GPU inference, and semaphores for memory-heavy models that cannot run in parallel. Start simple, measure under realistic load, and add complexity only when profiling justifies it. Above all, enforce bounds — on workers, queue depth, and in-flight requests — so that bursts of traffic degrade gracefully instead of crashing your server. With the patterns in this tutorial, you can build a local model server that stays responsive and reliable even when requests pile up.