Function _get_file_hash_async

modelscope/utils/file_utils.py:302–378

Compute SHA256 with async I/O double-buffering for high-latency storage. Uses a producer-consumer pattern: a background thread reads file chunks into a bounded queue while the main thread computes the hash. This overlaps I/O latency with CPU computation.

(
    file_path: Union[str, Path],
    buffer_size_mb: int = 16,
    tqdm_desc: Optional[str] = '[Calculating]',
    disable_tqdm: Optional[bool] = True,
)

def _get_file_hash_async(
    file_path: Union[str, Path],
    buffer_size_mb: int = 16,
    tqdm_desc: Optional[str] = '[Calculating]',
    disable_tqdm: Optional[bool] = True,
) -> dict:
    """Compute SHA256 with async I/O double-buffering for high-latency storage.

    Uses a producer-consumer pattern: a background thread reads file chunks
    into a bounded queue while the main thread computes the hash. This
    overlaps I/O latency with CPU computation.
    """
    import queue
    import threading

    from tqdm.auto import tqdm

    file_path = str(file_path)
    declared_size = os.path.getsize(file_path)
    buffer_size = buffer_size_mb * 1024 * 1024

    chunk_queue = queue.Queue(maxsize=2)  # Bounded to limit memory usage
    read_error = [None]  # Mutable container for thread error propagation

    def _producer():
        try:
            with open(file_path, 'rb') as f:
                while True:
                    chunk = f.read(buffer_size)
                    if not chunk:
                        break
                    chunk_queue.put(chunk)
        except Exception as e:
            read_error[0] = e
        finally:
            chunk_queue.put(None)  # Sentinel to signal EOF

    reader_thread = threading.Thread(target=_producer, daemon=True)
    reader_thread.start()

    file_hash = hashlib.sha256()
    bytes_hashed = 0
    progress = tqdm(
        total=declared_size,
        desc=tqdm_desc,
        disable=disable_tqdm,
        dynamic_ncols=True,
        unit='B',
        unit_scale=True,
        unit_divisor=1024,
    )

    while True:
        chunk = chunk_queue.get()
        if chunk is None:
            break
        file_hash.update(chunk)
        bytes_hashed += len(chunk)