new_blocks = self.kv_cache_manager.allocate_slots()`

self.kv_cache_manager.get_computed_blocks(request)

meta = self.connector.build_connector_meta(scheduler_output)

            req_meta = ReqMeta.from_request_tracker(
                request_tracker,
                self._block_size,
                self._lmcache_chunk_size,
                load_spec=None,
                discard_partial_chunks=self._discard_partial_chunks,
                save_decode_cache=self._save_decode_cache,
            )

# test_engine_lmcache_preemption.py 
import time
from typing import Iterable

import pytest

from vllm.config import KVTransferConfig
from vllm.engine.arg_utils import EngineArgs
from vllm.platforms import current_platform
from vllm.sampling_params import SamplingParams
from vllm.utils.torch_utils import set_default_torch_num_threads
from vllm.v1.engine import EngineCoreEventType, EngineCoreRequest
from vllm.v1.engine.core import EngineCore
from vllm.v1.executor.abstract import Executor

MODEL_NAME = "Qwen/Qwen3-0.6B-FP8"
BLOCK_SIZE = 16
NUM_BLOCKS = 7
PROMPT_TOKEN_COUNT = BLOCK_SIZE * 2 + 2
MAX_NEW_TOKENS = BLOCK_SIZE * 2 + 3

pytestmark = pytest.mark.skipif(  # type: ignore[var-annotated]
    not current_platform.is_cuda(),
    reason="LMCache preemption test requires CUDA support.",
)

def _iter_events(outputs: Iterable):
    for output in outputs:
        if not output.events:
            continue
        for event in output.events:
            yield event


def test_scheduler_handles_preemption_with_lmcache(monkeypatch: pytest.MonkeyPatch):
    """Launch a minimal vLLM engine with LMCache and trigger preemption.

    The test configures the engine to use the LMCache connector via the
    ``--kv-transfer-config`` equivalent API and constrains the number of GPU
    KV blocks. When multiple requests are executed concurrently the scheduler
    must preempt one of them in order to allocate a fresh block. We assert that
    a PREEMPTED event is observed in the engine outputs, ensuring that the
    LMCache-backed scheduler path handles preemption correctly.
    """

    pytest.importorskip("lmcache")

    from lmcache.integration.vllm.utils import ENGINE_NAME
    from lmcache.v1.cache_engine import LMCacheEngineBuilder

    env_vars = {
        "LMCACHE_USE_EXPERIMENTAL": "True",
        "LMCACHE_CHUNK_SIZE": str(BLOCK_SIZE),
        "LMCACHE_LOCAL_CPU": "True",
        "LMCACHE_MAX_LOCAL_CPU_SIZE": "1.0",
    }
    for key, value in env_vars.items():
        monkeypatch.setenv(key, value)

    kv_transfer_config = KVTransferConfig(
        kv_connector="LMCacheConnectorV1",
        kv_role="kv_both",
    )

    engine_args = EngineArgs(
        model=MODEL_NAME,
        max_model_len=BLOCK_SIZE * NUM_BLOCKS,
        block_size=BLOCK_SIZE,
        num_gpu_blocks_override=NUM_BLOCKS,
        kv_transfer_config=kv_transfer_config,
        enforce_eager=True,
        gpu_memory_utilization=0.1,
    )

    vllm_config = engine_args.create_engine_config()
    executor_cls = Executor.get_class(vllm_config)

    engine_core: EngineCore | None = None
    with set_default_torch_num_threads(1):
        engine_core = EngineCore(
            vllm_config=vllm_config,
            executor_class=executor_cls,
            log_stats=True,
        )

    def make_request(idx: int) -> EngineCoreRequest:
        prompt_tokens = [idx + 1] * PROMPT_TOKEN_COUNT
        sampling_params = SamplingParams(
            temperature=0.0,
            max_tokens=MAX_NEW_TOKENS,
        )
        return EngineCoreRequest(
            request_id=str(idx),
            prompt_token_ids=prompt_tokens,
            mm_features=None,
            sampling_params=sampling_params,
            pooling_params=None,
            eos_token_id=None,
            arrival_time=time.time(),
            lora_request=None,
            cache_salt=None,
            data_parallel_rank=None,
        )

    requests = [make_request(0), make_request(1)]

    prompt_token_ids = [requests[0].prompt_token_ids, requests[1].prompt_token_ids]
    output_token_ids = [[], []]

    try:
        for req in requests:
            engine_core.add_request(*engine_core.preprocess_add_request(req))

        seen_preemption = False
        finished = set()

        for _ in range(128):
            outputs_by_engine, model_executed = engine_core.step()

            if not outputs_by_engine:
                print("No outputs yet.")
                break

            engine_outputs = outputs_by_engine[0]

            for out in engine_outputs.outputs:
                output_token_ids[int(out.request_id)].extend(out.new_token_ids)
                if out.finished:
                    print(f"FINISHED REQ {out.request_id}: {prompt_token_ids[int(out.request_id)]}")
                    print(f"FINISHED REQ {out.request_id}: {output_token_ids[int(out.request_id)]}")
                    finished.add(out.request_id)

            for event in _iter_events(engine_outputs.outputs):
                print(f"Event: {event}")
                if event.type is EngineCoreEventType.PREEMPTED:
                    seen_preemption = True

            engine_core.post_step(model_executed)

            if len(finished) == len(requests):
                print("All requests finished.")
                break

        assert finished == {req.request_id for req in requests}
        assert seen_preemption, "Expected at least one preemption event."\
            " Check KV block configuration if this fails."

    finally:
        if engine_core is not None:
            engine_core.shutdown()
        LMCacheEngineBuilder.destroy(ENGINE_NAME)

# Example.py

from __future__ import annotations
from typing import Any
from vllm import EngineArgs, LLM
from vllm.inputs import TokensPrompt
from vllm.sampling_params import SamplingParams
from vllm.utils.argparse_utils import FlexibleArgumentParser

MODEL_NAME = "Qwen/Qwen3-0.6B-FP8"
BLOCK_SIZE = 16
PROMPT_TOKEN_COUNT = BLOCK_SIZE * 2 + 2
MAX_NEW_TOKENS = BLOCK_SIZE * 2 + 3


def create_parser() -> FlexibleArgumentParser:
    parser = FlexibleArgumentParser()
    EngineArgs.add_cli_args(parser)
    parser.set_defaults(model=MODEL_NAME)
    return parser


def build_prompts() -> list[TokensPrompt]:
    return [
        TokensPrompt(prompt_token_ids=[idx + 1] * PROMPT_TOKEN_COUNT)
        for idx in range(2)
    ]


def main(args: dict[str, Any]) -> None:
    llm = LLM(**args)

    sampling_params = SamplingParams(
        temperature=0.0,
        max_tokens=MAX_NEW_TOKENS,
    )

    prompts = build_prompts()
    outputs = llm.generate(prompts, sampling_params)

    print("-" * 80)
    for request_idx, output in enumerate(outputs):
        print(f"Request {request_idx}")
        print(f"  Prompt token IDs: {output.prompt_token_ids}")

        for candidate_idx, candidate in enumerate(output.outputs):
            print(
                f"  Candidate {candidate_idx} token IDs: {candidate.token_ids}"
            )
            print(f"  Candidate {candidate_idx} text: {candidate.text!r}")
        print("-" * 80)


if __name__ == "__main__":
    parser = create_parser()
    parsed_args: dict[str, Any] = vars(parser.parse_args())
    main(parsed_args)