Implementation:LMCache LMCache PQ Executor

Knowledge Sources	LMCache
Domains	Task Scheduling, Concurrency
Last Updated	2026-02-09 00:00 GMT

Overview

AsyncPQExecutor and AsyncPQThreadPoolExecutor provide priority-queue-based job scheduling for async and synchronous functions respectively.

Description

This module defines two executor classes extending BaseJobExecutor. AsyncPQExecutor runs async functions using a priority queue backed by asyncio.PriorityQueue, dispatching work to a configurable number of worker coroutines (default 4). Jobs are ordered by priority (lower number = higher priority) with FCFS tie-breaking via an atomic counter. AsyncPQThreadPoolExecutor extends AsyncPQExecutor to handle synchronous functions by wrapping them with asyncio.to_thread. Both executors use sentinel objects for graceful shutdown, support waiting for in-flight jobs to complete, and handle cancellation and generator exit exceptions cleanly. Each submitted job receives an asyncio.Future for result delivery.

Usage

Use AsyncPQExecutor when you need to schedule async I/O tasks (such as remote cache operations) with priority ordering. Use AsyncPQThreadPoolExecutor for CPU-bound or blocking synchronous functions that need priority-based scheduling. Both are used internally by the LMCache storage backend to manage concurrent remote operations.

Code Reference

Source Location

• Repository: LMCache
• File: lmcache/v1/storage_backend/job_executor/pq_executor.py
• Lines: 1-223

Signature

class AsyncPQExecutor(BaseJobExecutor):
    def __init__(self, loop: asyncio.AbstractEventLoop, max_workers: int = 4) -> None: ...
    async def submit_job(self, fn: Callable[..., Awaitable[Any]], *args: Any, **kwargs: Any) -> Any: ...
    def shutdown(self, wait: bool = True) -> None: ...

class AsyncPQThreadPoolExecutor(AsyncPQExecutor):
    def __init__(self, loop: asyncio.AbstractEventLoop, max_workers: int = 4) -> None: ...
    async def submit_job(self, fn: Callable[..., Awaitable[Any]], *args: Any, **kwargs: Any) -> Any: ...
    def shutdown(self, wait: bool = True) -> None: ...

Import

from lmcache.v1.storage_backend.job_executor.pq_executor import (
    AsyncPQExecutor,
    AsyncPQThreadPoolExecutor,
)

I/O Contract

Inputs

Name	Type	Required	Description
loop	asyncio.AbstractEventLoop	Yes	The event loop on which worker coroutines are scheduled
max_workers	int	No (default 4)	Maximum number of concurrent worker coroutines
fn	Callable[..., Awaitable[Any]]	Yes (for submit_job)	The async (or sync for ThreadPool variant) function to execute
*args	Any	No	Positional arguments passed to the function
priority	int	No (default 0, via kwargs)	Job priority (lower number = higher priority)

Outputs

Name	Type	Description
submit_job return	Any	The return value of the executed function, delivered via an asyncio.Future

Usage Examples

import asyncio
from lmcache.v1.storage_backend.job_executor.pq_executor import AsyncPQExecutor

loop = asyncio.get_event_loop()
executor = AsyncPQExecutor(loop=loop, max_workers=4)

# Submit an async job with default priority (0 = highest)
result = await executor.submit_job(some_async_function, arg1, arg2)

# Submit a lower-priority job (higher number = lower priority)
result = await executor.submit_job(
    another_async_function, arg1, priority=10
)

# Graceful shutdown waiting for jobs to complete
executor.shutdown(wait=True)