Skip to content
Open
Show file tree
Hide file tree
Changes from 27 commits
Commits
Show all changes
33 commits
Select commit Hold shift + click to select a range
829279b
Add GPUPool in the core rust code
YaxinCheng Jul 12, 2026
5365787
Switch python to use rust GPUPool
YaxinCheng Jul 12, 2026
a05e77e
Updated unit tests for GPUPool
YaxinCheng Jul 13, 2026
bba983b
Add function to acquire integer number of gpus
YaxinCheng Jul 14, 2026
953cf0d
Add back comments
YaxinCheng Jul 14, 2026
89c4646
Add clarification on the acquire full gpus
YaxinCheng Jul 14, 2026
527bc4f
release.notify_one() instead of notify_all() to avoid the thundering …
YaxinCheng Jul 16, 2026
4d0dbab
Expose acquire_full to python
YaxinCheng Jul 16, 2026
2f85c53
acquire_full returns error instead of crash
YaxinCheng Jul 17, 2026
b231cf7
Updated the comment
YaxinCheng Jul 17, 2026
95422da
Explicit waiting queue for GPUPool
YaxinCheng Jul 17, 2026
73e4e51
Add a Waiter struct for readability
YaxinCheng Jul 17, 2026
81af87f
Introduce GPU reservation for GPUs that are not currently available. …
YaxinCheng Jul 21, 2026
ed40a2d
Use try_join_all to better handle error
YaxinCheng Jul 22, 2026
d8a0c04
Update the functions to return errors when unexpected values are pass…
YaxinCheng Jul 22, 2026
e6eb8d4
find_available no longer finds the max, but finds the amount that is …
YaxinCheng Jul 24, 2026
567c2ad
Introduce GPUFraction type which handles f32 to usize, and handles th…
YaxinCheng Jul 24, 2026
70cd5d2
find_available no longer finds the max, but finds the amount that is …
YaxinCheng Jul 24, 2026
9043f87
Add a few traits to GPUFraction, and also make SCALE private
YaxinCheng Jul 24, 2026
4379ee0
Revamped queuing and handling strategy
YaxinCheng Jul 25, 2026
1c2c325
Code cleanup to improve readability
YaxinCheng Jul 29, 2026
808f869
Updated scheduling and reserving strategy
YaxinCheng Jul 31, 2026
dd3ba9f
Add comments and improve code readability
YaxinCheng Jul 31, 2026
6e7f512
Fix code based on comments
YaxinCheng Aug 1, 2026
a7acfd0
releas is now sync
YaxinCheng Aug 1, 2026
f7b088d
Fix code based on comments
YaxinCheng Aug 2, 2026
95c229e
Clean up
YaxinCheng Aug 3, 2026
eeba17d
Merge find excluding
YaxinCheng Aug 4, 2026
b574039
Updated tests
YaxinCheng Aug 4, 2026
2251496
Update function names
YaxinCheng Aug 6, 2026
53365a1
The queue is no longer strict FIFO
YaxinCheng Aug 6, 2026
54031e6
Use the right length
YaxinCheng Aug 7, 2026
ee61291
Release acquired GPUs when reservation is cancelled
YaxinCheng Aug 22, 2026
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -44,3 +44,4 @@ examples/**/uv.lock

# Claude Code local state
.claude/*.lock
.idea/
10 changes: 10 additions & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

36 changes: 36 additions & 0 deletions python/cocoindex/_internal/core.pyi
Original file line number Diff line number Diff line change
Expand Up @@ -704,3 +704,39 @@ class RateLimiter:
cls, max_rows_per_second: float, burst_window_secs: float = 1.0
) -> "RateLimiter": ...
def acquire(self, n: int = 1) -> Coroutine[Any, Any, None]: ...

########################################################
# GPU pool
########################################################

class GPUPool:
def __init__(self, num_gpus: int = 1) -> None: ...
Comment thread
YaxinCheng marked this conversation as resolved.
Outdated

@property
def num_gpus(self) -> int: ...

async def acquire(self, fraction: float) -> int:
"""
Acquires a fraction of a GPU and returns the GPU ID.
"""
...

async def acquire_full(self, gpu_count: int) -> list[int]:
"""
Acquires a given integer number of fully available GPUs (capacity == 1.0) from the GPU pool.

The gpu_count should be greater or equal to 1 and less than or equal to the number of GPUs in the pool.
"""
...

async def release(self, gpu_id: int, fraction: float) -> None:
Comment thread
YaxinCheng marked this conversation as resolved.
Outdated
"""
Releases a fraction of capacity back to the specified GPU ID.
"""
...

@staticmethod
def default() -> "GPUPool": ...

@staticmethod
def from_config(config_str: str) -> "GPUPool": ...
112 changes: 4 additions & 108 deletions python/cocoindex/_internal/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -213,110 +213,6 @@ def in_subprocess() -> bool:
# ============================================================================


def _detect_num_gpus() -> int:
"""Detect the number of GPUs available for the default pool.

Detection order:

1. ``COCOINDEX_NUM_GPUS`` environment variable (explicit override).
2. ``CUDA_VISIBLE_DEVICES`` environment variable (count of entries).
3. ``nvidia-smi`` command output (if available).
4. Default to ``1``.
"""
env_num = os.environ.get("COCOINDEX_NUM_GPUS")
if env_num is not None:
return max(1, int(env_num))

cuda_visible = os.environ.get("CUDA_VISIBLE_DEVICES")
if cuda_visible is not None:
devices = [d.strip() for d in cuda_visible.split(",") if d.strip()]
if devices:
return len(devices)
# Empty CUDA_VISIBLE_DEVICES disables CUDA devices; keep a logical
# pool size of 1 so GPU runners still behave predictably.
return 1

try:
result = subprocess.run(
["nvidia-smi", "--query-gpu=count", "--format=csv,noheader"],
capture_output=True,
text=True,
timeout=5,
check=False,
)
if result.returncode == 0:
output = result.stdout.strip().splitlines()
if output:
count = int(output[0].strip())
if count > 0:
return count
except Exception:
pass

return 1


class GPUPool:
Comment thread
YaxinCheng marked this conversation as resolved.
"""Tracks fractional GPU capacity across multiple GPUs.

Each GPU starts with capacity 1.0. ``acquire(fraction)`` blocks until a
GPU with enough remaining capacity is available, then returns its id.
``release(gpu_id, fraction)`` restores capacity and wakes waiters.

The default pool size is auto-detected from ``COCOINDEX_NUM_GPUS``,
``CUDA_VISIBLE_DEVICES``, or ``nvidia-smi`` (falling back to 1).
Call ``configure_gpu_pool(N)`` to override programmatically.
"""

_num_gpus: int
_capacity: list[float]
_cond: asyncio.Condition | None
_bound_loop: asyncio.AbstractEventLoop | None

def __init__(self, num_gpus: int) -> None:
if num_gpus < 1:
raise ValueError(f"num_gpus must be >= 1, got {num_gpus}")
self._num_gpus = num_gpus
self._capacity = [1.0] * num_gpus
self._cond = None
self._bound_loop = None

@property
def num_gpus(self) -> int:
return self._num_gpus

def _get_cond(self) -> asyncio.Condition:
loop = asyncio.get_running_loop()
if self._cond is None or self._bound_loop is not loop:
self._cond = asyncio.Condition()
self._bound_loop = loop
return self._cond

def _find_available(self, fraction: float) -> int | None:
best_gpu = None
best_cap = -1.0
for i, cap in enumerate(self._capacity):
if cap >= fraction and cap > best_cap:
best_gpu = i
best_cap = cap
return best_gpu

async def acquire(self, fraction: float) -> int:
async with self._get_cond():
while True:
gpu_id = self._find_available(fraction)
if gpu_id is not None:
self._capacity[gpu_id] -= fraction
return gpu_id
await self._get_cond().wait()

async def release(self, gpu_id: int, fraction: float) -> None:
cond = self._get_cond()
async with cond:
self._capacity[gpu_id] += fraction
cond.notify_all()


# ============================================================================
# GPU identity propagation
# ============================================================================
Expand Down Expand Up @@ -359,23 +255,23 @@ def _run_with_gpu_context(
# Default GPU pool
# ============================================================================

_default_gpu_pool: GPUPool | None = None
_default_gpu_pool: core.GPUPool | None = None
_default_gpu_pool_lock = threading.Lock()


def _get_default_gpu_pool() -> GPUPool:
def _get_default_gpu_pool() -> core.GPUPool:
global _default_gpu_pool
with _default_gpu_pool_lock:
if _default_gpu_pool is None:
_default_gpu_pool = GPUPool(num_gpus=_detect_num_gpus())
_default_gpu_pool = core.GPUPool.default()
return _default_gpu_pool


def configure_gpu_pool(num_gpus: int) -> None:
"""Override the default GPU pool. Must be called before any GPU function runs."""
global _default_gpu_pool
with _default_gpu_pool_lock:
_default_gpu_pool = GPUPool(num_gpus=num_gpus)
_default_gpu_pool = core.GPUPool(num_gpus=num_gpus)


# ============================================================================
Expand Down
Loading