Implementation:InternLM Lmdeploy RequestQueue

Knowledge Sources	InternLM_Lmdeploy
Domains	Inference Engine, Concurrency
Last Updated	2026-02-07 15:00 GMT

Overview

Implements a thread-safe request queue that buffers incoming inference and kill requests for consumption by the engine, with support for blocking waits, cancellation, and monotonic ID assignment.

Description

The RequestQueue class provides a concurrent producer-consumer queue specialized for TurboMind inference requests. It maintains two separate internal queues: a polymorphic memory resource-backed list (std::pmr::list) for inference requests, and a standard vector for kill requests.

Key features:

• Thread safety: All operations are protected by a mutex with condition variable signaling for blocking waits.
• Push/Kill: push() enqueues an inference request; kill() enqueues a kill (session termination) request. Both throw std::runtime_error if the queue has been closed.
• Pop: pop() dequeues up to max_infer inference requests and all pending kill requests. It supports both blocking and non-blocking modes. During dequeue, it atomically checks each request's cancel_flag and skips requests that have been canceled (flag != 0).
• Close: close() sets the closed flag and wakes all waiters, causing pop operations to set the abort flag.
• Unique ID assignment: assign_unique_ids() assigns monotonically increasing IDs to dequeued requests using an atomic counter.
• Memory optimization: Uses std::pmr::unsynchronized_pool_resource as the backing allocator for the inference queue list nodes, reducing allocation overhead.

Usage

Created and managed by the Gateway, one per engine queue slot. The Gateway calls push() and kill() from the request submission path, while engine threads call pop() to retrieve work items. close() is called during shutdown.

Code Reference

Source Location

• Repository: InternLM_Lmdeploy
• File: src/turbomind/engine/request_queue.h
• Lines: 1-105

Signature

class RequestQueue {
public:
    explicit RequestQueue();

    void push(std::shared_ptr<Request> r);

    void kill(std::shared_ptr<Request> r);

    void pop(std::vector<std::shared_ptr<Request>>& infer_reqs,
             std::vector<std::shared_ptr<Request>>& kill_reqs,
             unsigned                               max_infer,
             bool                                   blocking,
             bool&                                  abort);

    void close();

    void notify();

    void assign_unique_ids(std::vector<std::shared_ptr<Request>>& rs);

private:
    std::atomic<uint64_t> unique_id_{};
    std::pmr::unsynchronized_pool_resource   pool_;
    std::pmr::list<std::shared_ptr<Request>> queue_;
    std::vector<std::shared_ptr<Request>> kill_;
    std::mutex              mutex_;
    std::condition_variable cv_;
    bool closed_{};
};

Import

#include "src/turbomind/engine/request_queue.h"

I/O Contract

Inputs

Name	Type	Required	Description
r (push)	std::shared_ptr<Request>	Yes	Inference request to enqueue
r (kill)	std::shared_ptr<Request>	Yes	Kill request to enqueue
max_infer (pop)	unsigned	Yes	Maximum number of inference requests to dequeue
blocking (pop)	bool	Yes	If true, block until requests are available or queue is closed

Outputs

Name	Type	Description
infer_reqs	std::vector<std::shared_ptr<Request>>&	Dequeued inference requests (non-canceled only)
kill_reqs	std::vector<std::shared_ptr<Request>>&	Dequeued kill requests
abort	bool&	Set to true if the queue has been closed

Usage Examples

RequestQueue queue;

// Producer: submit a request
auto req = std::make_shared<Request>();
queue.push(req);

// Consumer: retrieve requests
std::vector<std::shared_ptr<Request>> infer_reqs, kill_reqs;
bool abort = false;
queue.pop(infer_reqs, kill_reqs, /*max_infer=*/16, /*blocking=*/true, abort);

// Assign unique IDs to dequeued requests
queue.assign_unique_ids(infer_reqs);

// Shutdown
queue.close();

Related Pages

• Environment:InternLM_Lmdeploy_CUDA_GPU_Runtime