Skip to content

vllm.distributed.kv_transfer.kv_connector.v1.hf3fs.utils.common

Classes:

AtomicCounter

Thread-safe atomic counter for round-robin operations.

Methods:

  • next –

    Get next value in round-robin fashion.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/hf3fs/utils/common.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
class AtomicCounter:
    """Thread-safe atomic counter for round-robin operations."""

    def __init__(self, n: int):
        assert n > 0, "Counter size must be positive"
        self._n = n
        self._value = 0
        self._lock = threading.Lock()

    def next(self) -> int:
        """Get next value in round-robin fashion."""
        with self._lock:
            current = self._value
            self._value = (current + 1) % self._n
            return current

next()

Get next value in round-robin fashion.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/hf3fs/utils/common.py 
20
21
22
23
24
25
def next(self) -> int:
    """Get next value in round-robin fashion."""
    with self._lock:
        current = self._value
        self._value = (current + 1) % self._n
        return current

HF3FSConnectorMetadata

Bases: KVConnectorMetadata

Container for HF3FS connector metadata.

Methods:

Source code in vllm/distributed/kv_transfer/kv_connector/v1/hf3fs/utils/common.py 
131
132
133
134
135
136
137
138
139
class HF3FSConnectorMetadata(KVConnectorMetadata):
    """Container for HF3FS connector metadata."""

    def __init__(self):
        self.requests: list[HF3FSRequestMetadata] = []

    def add_request(self, request_metadata: HF3FSRequestMetadata) -> None:
        """Add request to metadata."""
        self.requests.append(request_metadata)

add_request(request_metadata)

Add request to metadata.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/hf3fs/utils/common.py 
137
138
139
def add_request(self, request_metadata: HF3FSRequestMetadata) -> None:
    """Add request to metadata."""
    self.requests.append(request_metadata)

HF3FSRequestMetadata dataclass

Metadata for a single request in HF3FS connector.

Methods:

Source code in vllm/distributed/kv_transfer/kv_connector/v1/hf3fs/utils/common.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@dataclass
class HF3FSRequestMetadata:
    """Metadata for a single request in HF3FS connector."""

    request_id: str
    token_ids: list[int]
    block_ids: list[int]
    load_block_op: LoadBlockInfo | None = None
    save_block_op: SaveBlockInfo | None = None

    @staticmethod
    def from_scheduling_state(
        state: "RequestSchedulingState",
        block_size: int,
        load_op: LoadBlockInfo | None = None,
        skip_leading_blocks: int | None = None,
    ) -> Optional["HF3FSRequestMetadata"]:
        """Create request metadata from scheduling state."""
        token_count = len(state.token_ids)
        total_blocks = token_count // block_size

        skip_blocks = (
            state.num_saved_blocks
            if skip_leading_blocks is None
            else skip_leading_blocks
        )

        new_blocks_to_save = total_blocks - state.num_saved_blocks
        if new_blocks_to_save <= 0 and load_op is None:
            return None

        state.num_saved_blocks = total_blocks
        return HF3FSRequestMetadata(
            request_id=state.request_id,
            token_ids=state.token_ids,
            block_ids=state.allocated_block_ids,
            load_block_op=load_op,
            save_block_op=SaveBlockInfo(skip_leading_blocks=skip_blocks),
        )

from_scheduling_state(state, block_size, load_op=None, skip_leading_blocks=None) staticmethod

Create request metadata from scheduling state.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/hf3fs/utils/common.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
@staticmethod
def from_scheduling_state(
    state: "RequestSchedulingState",
    block_size: int,
    load_op: LoadBlockInfo | None = None,
    skip_leading_blocks: int | None = None,
) -> Optional["HF3FSRequestMetadata"]:
    """Create request metadata from scheduling state."""
    token_count = len(state.token_ids)
    total_blocks = token_count // block_size

    skip_blocks = (
        state.num_saved_blocks
        if skip_leading_blocks is None
        else skip_leading_blocks
    )

    new_blocks_to_save = total_blocks - state.num_saved_blocks
    if new_blocks_to_save <= 0 and load_op is None:
        return None

    state.num_saved_blocks = total_blocks
    return HF3FSRequestMetadata(
        request_id=state.request_id,
        token_ids=state.token_ids,
        block_ids=state.allocated_block_ids,
        load_block_op=load_op,
        save_block_op=SaveBlockInfo(skip_leading_blocks=skip_blocks),
    )

LoadBlockInfo dataclass

Operation for loading blocks from external storage.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/hf3fs/utils/common.py 
28
29
30
31
32
33
34
@dataclass
class LoadBlockInfo:
    """Operation for loading blocks from external storage."""

    num_computed_blocks: int
    num_blocks_to_load: int
    need_fetch_block_ids: list[int]

RequestSchedulingState dataclass

Unified request scheduling state management.

Methods:

Source code in vllm/distributed/kv_transfer/kv_connector/v1/hf3fs/utils/common.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@dataclass
class RequestSchedulingState:
    """Unified request scheduling state management."""

    request_id: str
    request: Request | None = None

    # Token and block tracking
    token_ids: list[int] = field(default_factory=list)
    allocated_block_ids: list[int] = field(default_factory=list)
    num_saved_blocks: int = 0

    # Load operation info
    load_op: LoadBlockInfo | None = None

    # Scheduling phase
    phase: str = "NEW"  # NEW -> WAITING_TO_LOAD -> ACTIVE -> FINISHED

    def needs_loading(self) -> bool:
        """Check if request needs loading."""
        return self.load_op is not None and self.load_op.num_blocks_to_load > 0

    def is_ready_to_load(self) -> bool:
        """Check if request is ready for loading."""
        return self.phase == "WAITING_TO_LOAD" and self.needs_loading()

    def update_tokens_and_blocks(self, new_token_ids: list[int], new_block_ids) -> None:
        """Update with new tokens and blocks."""
        if new_token_ids:
            self.token_ids.extend(new_token_ids)

        if new_block_ids is not None:
            normalized_block_ids = self._normalize_block_ids(new_block_ids)
            self.allocated_block_ids.extend(normalized_block_ids)

    def _normalize_block_ids(self, block_ids) -> list[int]:
        """Normalize block_ids to list format."""
        if not block_ids:
            return []
        if isinstance(block_ids, tuple):
            return block_ids[0] if block_ids else []
        if isinstance(block_ids, list):
            return block_ids
        return []

_normalize_block_ids(block_ids)

Normalize block_ids to list format.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/hf3fs/utils/common.py 
79
80
81
82
83
84
85
86
87
def _normalize_block_ids(self, block_ids) -> list[int]:
    """Normalize block_ids to list format."""
    if not block_ids:
        return []
    if isinstance(block_ids, tuple):
        return block_ids[0] if block_ids else []
    if isinstance(block_ids, list):
        return block_ids
    return []

is_ready_to_load()

Check if request is ready for loading.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/hf3fs/utils/common.py 
66
67
68
def is_ready_to_load(self) -> bool:
    """Check if request is ready for loading."""
    return self.phase == "WAITING_TO_LOAD" and self.needs_loading()

needs_loading()

Check if request needs loading.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/hf3fs/utils/common.py 
62
63
64
def needs_loading(self) -> bool:
    """Check if request needs loading."""
    return self.load_op is not None and self.load_op.num_blocks_to_load > 0

update_tokens_and_blocks(new_token_ids, new_block_ids)

Update with new tokens and blocks.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/hf3fs/utils/common.py 
70
71
72
73
74
75
76
77
def update_tokens_and_blocks(self, new_token_ids: list[int], new_block_ids) -> None:
    """Update with new tokens and blocks."""
    if new_token_ids:
        self.token_ids.extend(new_token_ids)

    if new_block_ids is not None:
        normalized_block_ids = self._normalize_block_ids(new_block_ids)
        self.allocated_block_ids.extend(normalized_block_ids)

SaveBlockInfo dataclass

Operation for saving blocks to external storage.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/hf3fs/utils/common.py 
37
38
39
40
41
@dataclass
class SaveBlockInfo:
    """Operation for saving blocks to external storage."""

    skip_leading_blocks: int