如何在Python中实现无缓冲磁盘读取？解决参数错误及缓冲区问题

问题根源 #

Linux下使用O_DIRECT标志读取磁盘时，有三个严格的对齐要求，缺一不可：

• 文件读取的偏移量必须对齐到磁盘扇区大小
• 存储数据的缓冲区地址必须对齐到磁盘扇区大小
• 读取的字节数必须是扇区大小的整数倍

你原代码中os.read()会自动创建内部缓冲区，这个缓冲区的地址无法保证满足对齐要求，因此触发Invalid argument错误。

解决方法：用ctypes直接调用系统read函数 #

os.read()不支持传入自定义缓冲区，所以需要借助ctypes直接调用Linux系统级的read函数，这样就能使用你预分配的对齐缓冲区。

完整实现代码 #

import os
import ctypes
from ctypes import cdll, c_int, c_size_t, c_void_p

# 动态获取磁盘扇区大小（避免硬编码，适配不同设备）
disk_path = "/dev/sdc"
with open(disk_path, "rb") as f:
    sector_size = os.fstat(f.fileno()).st_blksize

# 定义系统级read函数的参数和返回值类型
libc = cdll.LoadLibrary("libc.so.6")
libc.read.argtypes = [c_int, c_void_p, c_size_t]
libc.read.restype = ctypes.c_ssize_t

buffer_size = 4096
# 确保缓冲区大小是扇区大小的整数倍
assert buffer_size % sector_size == 0, "缓冲区大小必须是扇区大小的整数倍"

# 预分配对齐的缓冲区
raw_buffer = ctypes.create_string_buffer(buffer_size + sector_size)
buffer_addr = (ctypes.addressof(raw_buffer) + sector_size - 1) & ~(sector_size - 1)
aligned_buffer = ctypes.string_at(buffer_addr, buffer_size)

disk_fd = None
try:
    # 打开磁盘设备，启用O_DIRECT
    disk_fd = os.open(disk_path, os.O_RDONLY | os.O_DIRECT)
    # 定位到对齐的偏移量（需确保是扇区大小的整数倍）
    target_offset = 12345 * 4096
    assert target_offset % sector_size == 0, "偏移量必须对齐到扇区大小"
    os.lseek(disk_fd, target_offset, os.SEEK_SET)
    
    # 调用系统read函数读取数据到对齐缓冲区
    bytes_read = libc.read(disk_fd, buffer_addr, buffer_size)
    if bytes_read == -1:
        raise OSError(f"读取失败: {ctypes.get_errno()}")
    
    # 处理读取到的数据，aligned_buffer即为包含数据的bytes对象
    print(f"成功读取{bytes_read}字节数据")
    # 示例：打印前16字节的十六进制值
    print(aligned_buffer[:16].hex())

finally:
    if disk_fd is not None:
        os.close(disk_fd)

关键注意事项 #

• 必须用sudo运行脚本，直接访问磁盘设备需要root权限
• 不要硬编码扇区大小，通过os.fstat().st_blksize动态获取更可靠
• 偏移量、缓冲区大小、缓冲区地址三者必须同时满足对齐要求，任一不满足都会触发错误
• 若需要写入操作，逻辑完全一致，替换为调用libc.write即可

内容的提问来源于stack exchange，提问作者Marus Gradinaru