Async Route Matrix Automation

Computing a travel-time matrix for 500 delivery stops means issuing dozens of chunked requests against a routing engine, and doing it serially can take minutes per dispatch cycle — long enough to make same-day re-optimization impractical. This section of the Routing API Automation & Fleet Integration guide covers the async Python patterns that make large origin-destination (OD) matrix computation fast and reliable: bounded concurrency against OSRM and Valhalla table endpoints, chunk tiling to stay under engine limits, and NumPy-based assembly of the final matrix. The two pages beneath this one go deeper on the fan-out mechanics and the tiling math; this page ties them into a working pipeline.

Prerequisites

The pipeline in this section assumes a routing engine table endpoint is already reachable. If you have not stood one up, follow deploying OSRM with Docker for local routing first — the --max-table-size and --threads flags set there directly determine the chunk size and concurrency limits used below.

System requirements

Python environment

# Install client-side dependencies (Python 3.9+)
pip install aiohttp numpy tenacity

Resources referenced on this page

Conceptual Architecture

A naive implementation requests the full N×M matrix in a single HTTP call. That works for small stop lists, but two limits break it at fleet scale:

1. Engine-side caps. osrm-routed rejects any /table request whose coordinate count exceeds --max-table-size (default 100). Valhalla enforces an analogous max_distance and max_locations in its matrix costing config. Both exist to bound per-request memory and CPU inside the engine process.
2. Client-side serialization cost. Even under the cap, a single blocking request for a 900×900 matrix ties up one connection for the full computation. Multiple stop lists queued behind each other serialize what should be independent work.

The fix is to decompose the matrix into row/column blocks no larger than the engine limit, issue those blocks concurrently under a bounded semaphore, and write each returned block into a preallocated NumPy array at its correct index range. This is the same architecture used by distributed batch systems: a tiler produces work units, a worker pool with backpressure executes them, and a reducer assembles results. Chunking large origin-destination matrices covers the tiling math — how block size interacts with max_table_size and with reusing symmetric distance pairs — in more depth than fits here.

Concurrency itself needs a ceiling. Because /table computation is CPU-bound inside the engine (each additional source runs a full multi-target Dijkstra sweep from that source), issuing more concurrent requests than the engine has worker threads does not increase throughput — it just queues requests behind each other at the OS socket level, which shows up as client-side timeouts rather than engine errors. An asyncio.Semaphore sized to the engine’s --threads value keeps the client from generating load the engine cannot absorb. Parallel distance matrix requests with aiohttp covers connector-level tuning (connection pooling, per-host limits, timeout budgets) that complements the semaphore gate described here.

Both OSRM and Valhalla expose table-style endpoints, but with different request/response shapes:

• OSRM /table/v1/{profile}/{coordinates} takes semicolon-separated lon,lat pairs plus sources/destinations index lists, and returns flat durations and distances 2D arrays.
• Valhalla /sources_to_targets takes JSON sources and targets arrays of {lat, lon} objects, and returns a nested sources_to_targets array of per-pair {distance, time} objects, one list per source.

Normalizing both into the same NumPy shape at the assembly boundary — rather than branching on engine type throughout the pipeline — keeps downstream code (dispatch, isochrone scoring, VRP solvers) engine-agnostic.

Step-by-Step Implementation

1. Define the Origin-Destination Point Set

Preserve index order carefully — the matrix assembly step depends on origin index i and destination index j mapping consistently back to the same coordinate throughout the pipeline.

# requires: numpy (pip install numpy)
import numpy as np

def load_points(csv_path: str) -> np.ndarray:
    """Load lon/lat pairs from CSV; return an (N, 2) float64 array."""
    # Expects columns: lon, lat (matches OSRM coordinate order)
    return np.loadtxt(csv_path, delimiter=",", skiprows=1, usecols=(0, 1))

origins = load_points("depot_and_stops.csv")       # (N, 2)
destinations = origins                               # square matrix: same point set both sides
N, M = len(origins), len(destinations)
print(f"Matrix target shape: {N}x{M}")

For asymmetric problems — a fixed depot list routed against a rotating stop list — load origins and destinations from separate files. The chunk tiler in step 3 works identically either way.

2. Build the Async Table-Request Function

This coroutine issues one bounded OSRM /table request and parses the response into a NumPy sub-block. Keep it engine-specific and narrow; step 5 adds the Valhalla equivalent behind the same interface.

# requires: aiohttp, numpy (pip install aiohttp numpy)
import aiohttp
import numpy as np

OSRM_HOST = "http://localhost:5000"

def _coords_param(points: np.ndarray) -> str:
    return ";".join(f"{lon:.6f},{lat:.6f}" for lon, lat in points)

async def fetch_osrm_block(
    session: aiohttp.ClientSession,
    origin_block: np.ndarray,
    dest_block: np.ndarray,
) -> np.ndarray:
    """Fetch a durations sub-block for one origin/destination chunk pair."""
    all_points = np.vstack([origin_block, dest_block])
    n_src = len(origin_block)
    sources = ";".join(str(i) for i in range(n_src))
    destinations = ";".join(str(i) for i in range(n_src, len(all_points)))

    url = (
        f"{OSRM_HOST}/table/v1/driving/{_coords_param(all_points)}"
        f"?sources={sources}&destinations={destinations}&annotations=duration"
    )
    async with session.get(url, timeout=aiohttp.ClientTimeout(total=15)) as resp:
        resp.raise_for_status()
        data = await resp.json()

    if data["code"] != "Ok":
        raise RuntimeError(f"OSRM table error: {data.get('message')}")

    # None entries mean an unreachable pair; keep as NaN rather than 0
    durations = np.array(
        [[v if v is not None else np.nan for v in row] for row in data["durations"]],
        dtype=np.float64,
    )
    return durations

3. Fan Out Requests with a Bounded Semaphore

The semaphore caps in-flight requests; the aiohttp.TCPConnector limit should be set to the same value so the connection pool never queues more work than the semaphore already allows through.

# requires: aiohttp, numpy (pip install aiohttp numpy)
import asyncio
import aiohttp
import numpy as np

async def gated_fetch(
    sem: asyncio.Semaphore,
    session: aiohttp.ClientSession,
    origin_block: np.ndarray,
    dest_block: np.ndarray,
    row_range: tuple[int, int],
    col_range: tuple[int, int],
) -> tuple[tuple[int, int], tuple[int, int], np.ndarray]:
    async with sem:
        block = await fetch_osrm_block(session, origin_block, dest_block)
    return row_range, col_range, block

async def compute_matrix(
    origins: np.ndarray,
    destinations: np.ndarray,
    chunk_size: int = 100,
    concurrency: int = 4,
) -> np.ndarray:
    """Compute the full N×M duration matrix via chunked, bounded-concurrency requests."""
    N, M = len(origins), len(destinations)
    matrix = np.full((N, M), np.nan, dtype=np.float64)
    sem = asyncio.Semaphore(concurrency)
    connector = aiohttp.TCPConnector(limit=concurrency, limit_per_host=concurrency)

    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = []
        for r0 in range(0, N, chunk_size):
            r1 = min(r0 + chunk_size, N)
            for c0 in range(0, M, chunk_size):
                c1 = min(c0 + chunk_size, M)
                tasks.append(gated_fetch(
                    sem, session,
                    origins[r0:r1], destinations[c0:c1],
                    (r0, r1), (c0, c1),
                ))
        for coro in asyncio.as_completed(tasks):
            (r0, r1), (c0, c1), block = await coro
            matrix[r0:r1, c0:c1] = block

    return matrix

asyncio.as_completed writes each block into the matrix as soon as it lands, rather than waiting on the slowest chunk in the batch — useful when partial results feed a progress indicator or a streaming dispatch loop.

4. Assemble the Partial Responses into a NumPy Matrix

Assembly happens inline in step 3, but for pipelines that persist raw chunk responses (e.g., for replay or debugging) separate the reduce step so it can run independently of the fetch step:

# requires: numpy (pip install numpy)
import numpy as np

def assemble_matrix(
    n_rows: int,
    n_cols: int,
    chunks: list[tuple[tuple[int, int], tuple[int, int], np.ndarray]],
) -> np.ndarray:
    """Reassemble a matrix from a list of (row_range, col_range, block) tuples."""
    matrix = np.full((n_rows, n_cols), np.nan, dtype=np.float64)
    for (r0, r1), (c0, c1), block in chunks:
        expected_shape = (r1 - r0, c1 - c0)
        if block.shape != expected_shape:
            raise ValueError(f"Block shape {block.shape} != expected {expected_shape}")
        matrix[r0:r1, c0:c1] = block
    return matrix

missing_pairs = np.argwhere(np.isnan(matrix))
if len(missing_pairs):
    print(f"{len(missing_pairs)} unreachable or failed pairs remain — see step 6 for retry handling")

5. Add a Valhalla Fallback Backend

Valhalla’s payload shape differs from OSRM’s, so the coroutine parses sources_to_targets into the same flat NumPy layout before it reaches the assembly step:

# requires: aiohttp, numpy (pip install aiohttp numpy)
import aiohttp
import numpy as np

VALHALLA_HOST = "http://localhost:8002"

async def fetch_valhalla_block(
    session: aiohttp.ClientSession,
    origin_block: np.ndarray,
    dest_block: np.ndarray,
) -> np.ndarray:
    """Fetch a time sub-block from Valhalla's sources_to_targets endpoint."""
    payload = {
        "sources": [{"lat": lat, "lon": lon} for lon, lat in origin_block],
        "targets": [{"lat": lat, "lon": lon} for lon, lat in dest_block],
        "costing": "auto",
    }
    async with session.post(
        f"{VALHALLA_HOST}/sources_to_targets",
        json=payload,
        timeout=aiohttp.ClientTimeout(total=15),
    ) as resp:
        resp.raise_for_status()
        data = await resp.json()

    n_src, n_dst = len(origin_block), len(dest_block)
    block = np.full((n_src, n_dst), np.nan, dtype=np.float64)
    for i, row in enumerate(data["sources_to_targets"]):
        for j, pair in enumerate(row):
            if pair.get("time") is not None:
                block[i, j] = pair["time"]
    return block

Route matrix jobs that need cross-validation — confirming OSRM and Valhalla agree within a tolerance before trusting a dispatch decision — can call both coroutines for the same chunk and diff the results. See comparing routing engines for production for a broader latency and accuracy benchmarking harness.

6. Persist the Matrix and Handle Partial Failures

# requires: aiohttp, numpy, tenacity (pip install aiohttp numpy tenacity)
import numpy as np
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import aiohttp

@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=0.5, max=8),
    retry=retry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError)),
)
async def fetch_osrm_block_with_retry(session, origin_block, dest_block):
    return await fetch_osrm_block(session, origin_block, dest_block)

def save_matrix(matrix: np.ndarray, origins: np.ndarray, path: str) -> None:
    """Persist the matrix alongside the point index it was computed against."""
    np.savez_compressed(path, matrix=matrix, origins=origins)
    nan_count = int(np.isnan(matrix).sum())
    print(f"Saved {matrix.shape} matrix to {path} ({nan_count} unresolved pairs)")

Swap fetch_osrm_block for fetch_osrm_block_with_retry inside gated_fetch from step 3 to get exponential-backoff retry on transient failures without changing the fan-out logic. Persisting unresolved-pair counts alongside the matrix gives dispatch systems a signal for whether to trust the result or trigger a re-run.

Configuration Reference

Production Optimization and Scaling

Reuse symmetric pairs. For a square, symmetric-cost matrix (rare with one-way streets but common for straight-line pre-filtering), computing only the upper triangle and mirroring it halves request volume. Chunking large origin-destination matrices covers when this assumption holds for road-network routing and when it silently produces wrong results.

Scale out, not up. Because /table throughput is bound by engine CPU threads, raising client concurrency past that ceiling only adds queue depth. Run multiple osrm-routed replicas behind a round-robin load balancer (Nginx upstream block or a Kubernetes Service) and raise the semaphore to replicas × threads_per_replica. This is the same scaling model used for the containerized deployment in deploying OSRM with Docker for local routing, extended horizontally.

Session reuse across jobs. Opening a new aiohttp.ClientSession per matrix job pays a connection-establishment cost on every run. For a service that computes matrices repeatedly (e.g., a dispatch cycle every 15 minutes), keep one long-lived session with a TCPConnector and pass it into each job rather than recreating it.

Warm the connection pool before bursts. If matrix computation is triggered by a batch event (end-of-shift re-optimization for 200 vehicles), issue a handful of cheap /table warm-up requests before the real burst. TCP connection setup and OSRM’s internal query cache both benefit from not starting cold.

Adaptive chunk sizing. Fixed chunk_size values leave throughput on the table when point density varies — a chunk of tightly clustered urban stops routes faster than one spanning a rural highway network. Track per-chunk latency and shrink chunk_size dynamically if P95 chunk time exceeds a budget, rather than tuning a single static value for the worst case.

Validation and Testing

# requires: numpy (pip install numpy)
import numpy as np

def check_matrix_shape(matrix: np.ndarray, expected_n: int, expected_m: int) -> None:
    assert matrix.shape == (expected_n, expected_m), f"Shape mismatch: {matrix.shape}"
    print(f"[PASS] Matrix shape {matrix.shape}")

def check_diagonal_zero(matrix: np.ndarray) -> None:
    """Self-to-self duration should be exactly 0 for a square OD matrix."""
    if matrix.shape[0] == matrix.shape[1]:
        diag = np.diagonal(matrix)
        assert np.allclose(diag, 0.0, atol=1e-6), "Non-zero self-route duration found"
        print("[PASS] Diagonal is zero")

def check_nan_ratio(matrix: np.ndarray, max_ratio: float = 0.02) -> None:
    """Flag matrices with more than 2% unreachable or failed pairs."""
    ratio = np.isnan(matrix).sum() / matrix.size
    assert ratio <= max_ratio, f"NaN ratio {ratio:.2%} exceeds budget {max_ratio:.2%}"
    print(f"[PASS] NaN ratio {ratio:.2%} within budget")

def check_triangle_inequality_sample(
    matrix: np.ndarray, i: int, j: int, k: int, tolerance: float = 1.15
) -> None:
    """Spot-check that direct i->j isn't wildly worse than i->k->j (loose sanity bound)."""
    direct, via_k = matrix[i, j], matrix[i, k] + matrix[k, j]
    if not (np.isnan(direct) or np.isnan(via_k)):
        assert direct <= via_k * tolerance, "Direct route unexpectedly worse than detour"
        print(f"[PASS] Triangle spot-check {i}->{j} within tolerance")

check_matrix_shape(matrix, N, M)
check_diagonal_zero(matrix)
check_nan_ratio(matrix)
check_triangle_inequality_sample(matrix, 0, 5, 12)

Sanity metrics to monitor in production:

• Chunk-level P99 latency should scale roughly linearly with chunk_size; a sudden jump indicates engine-side contention, not client concurrency.
• NaN ratio trending upward over consecutive runs against a stable point set signals graph connectivity regressions worth checking against graph fragmentation prevention in OSM data.
• Total wall-clock time for a full matrix job should scale sub-linearly with concurrency up to the replica thread ceiling, then flatten — a useful signal for right-sizing the semaphore.

Troubleshooting

Related

• Parallel distance matrix requests with aiohttp — connection pooling, timeout budgets, and connector tuning for the fan-out layer
• Chunking large origin-destination matrices — block partitioning math, symmetric-pair reuse, and memory-bounded aggregation
• Deploying OSRM with Docker for local routing — standing up the /table endpoint this pipeline targets, including --max-table-size and --threads tuning
• Valhalla cost matrix generation for urban planners — Valhalla-specific matrix costing and /sources_to_targets configuration
• Comparing routing engines for production — cross-validating OSRM and Valhalla output before trusting a dispatch decision