Function flush_heartbeats

gpustack/server/worker_status_buffer.py:25–55

Flush worker heartbeat updates to the database periodically. Uses a single UPDATE statement to update all workers with the same timestamp.

()

async def flush_heartbeats():
    """
    Flush worker heartbeat updates to the database periodically.
    Uses a single UPDATE statement to update all workers with the same timestamp.
    """
    if not heartbeat_flush_buffer:
        return

    # Copy buffer and clear it atomically
    async with heartbeat_flush_buffer_lock:
        local_buffer = set(heartbeat_flush_buffer)
        heartbeat_flush_buffer.clear()

    try:
        async with async_session() as session:
            # Single UPDATE for all workers with the same timestamp
            # UPDATE workers SET heartbeat_time = '2024-01-27 10:00:00' WHERE id IN (1, 2, 3, ...)
            heartbeat_time = datetime.datetime.now(datetime.timezone.utc).replace(
                microsecond=0
            )

            stmt = (
                update(Worker)
                .where(Worker.id.in_(local_buffer))
                .values(heartbeat_time=heartbeat_time)
            )

            await session.execute(stmt)
            await session.commit()
    except Exception as e:
        logger.error(f"Error flushing heartbeats to DB: {e}")


async def flush_worker_status():