Python 中如何在多线程读取时安全扩容 mmap？

2026-11-04数据小陈

应为读取、写入、扩容和关闭建立统一的同步规则。resize() 可能移动映射，但这并不意味着原代码中的每次并发切片都会读到失效地址。还要区分普通切片和零拷贝视图：当前 read_bytes() 返回的是复制得到的 bytes，并没有实现零拷贝。

如果需要长期保留零拷贝视图，同时让日志持续增长，可以采用固定容量的 segment，写满后创建下一个 segment，避免对正在被读取的映射执行 resize()。

原代码会不会在扩容时读到失效地址？

以启用 GIL 的 CPython 3.13.0 为例，普通整数索引的 mmap 切片复制和 resize() 的相关 C 实现，在操作期间都没有释放 GIL。因此，在没有外部文件修改、原生指针访问等额外因素时，这两项操作不会同时执行，不能直接推导出“一个线程复制数据时，另一个线程把映射移走”。这是特定实现的行为，不能当作所有 Python 版本和运行方式的通用保证。切片返回 bytes、扩容可能移动地址，也可以在 CPython 3.13.0 源码中确认。

原代码仍有几处需要同步：

在 assert(self.__mmap is not None) 与随后使用映射之间，release() 可能关闭资源。
扩容和写入包含多个步骤，读取方没有参与“数据何时写完并可见”的协议。
_refcount 虽然已经声明，但展示的代码没有用它管理读取期间的资源生命周期。

扩大文件为什么通常不会直接触发 SIGBUS？

容量不足时，这里的 truncate(new_capacity) 传入的是更大的值，执行的是扩展文件，并不会因为方法名叫 truncate 就缩短文件。

在 Linux 上，一种典型的 SIGBUS 场景是：文件被缩短后，程序继续访问映射中位于新文件末尾之外的页面。仅扩展文件并继续读取原映射范围，通常不属于这种情况。映射移动后继续使用旧原生地址是另一类无效内存访问问题，不能一概称为 SIGBUS。参见 Linux mmap(2) 手册。

文件变大也不会自动扩大已有映射的可访问范围。扩容完成前，读取边界仍应按当前映射长度判断。

如何用最少的修改建立明确的安全边界？

先让所有操作使用同一把锁。读取时，在锁内检查状态和范围并完成复制；写入时，在锁内检查状态、扩容并写入；关闭时，也必须持有这把锁。

以下方法可放回原来的 Entry 类。示例假定映射覆盖从文件偏移 0 开始的区域，运行平台支持 resize()，且没有其他进程修改文件大小，也没有向外借出 memoryview：

def read_bytes(self, offset: int, length: int) -> bytes:
    with self._lock:
        mm = self.__mmap
        if mm is None:
            raise RuntimeError("segment is not loaded")
        if offset < 0 or length < 0:
            raise ValueError("offset and length must be non-negative")
        end = offset + length
        if end > len(mm):
            raise ValueError("read exceeds mapped capacity")
        return mm[offset:end]


def write(self, offset: int, msg: bytes) -> None:
    with self._lock:
        if self.__mmap is None:
            raise RuntimeError("segment is not loaded")
        if not self.__mutable:
            raise RuntimeError("segment is immutable")
        if offset < 0:
            raise ValueError("offset must be non-negative")
        end = offset + len(msg)
        self._ensure_capacity_locked(end)
        self.__mmap[offset:end] = msg


def _ensure_capacity_locked(self, capacity: int) -> None:
    mm = self.__mmap
    if mm is None or self.__capacity is None:
        raise RuntimeError("segment is not loaded")
    if capacity <= self.__capacity:
        return
    new_capacity = max(
        capacity, self.__capacity + self.segment_size_inc
    )
    mm.resize(new_capacity)
    self.__capacity = new_capacity


def release(self) -> None:
    with self._lock:
        if self.__mmap is not None:
            if self.__mutable:
                self.__mmap.flush()
            self.__mmap.close()
            self.__mmap = None
        if self.__file_obj is not None:
            if self.__mutable:
                self.__file_obj.flush()
            self.__file_obj.close()
            self.__file_obj = None
        self.__capacity = None

resize() 本身会调整映射及底层文件，因此这里去掉了预先执行的 truncate()。它并非在所有平台都可用，ACCESS_READ 和 ACCESS_COPY 映射也不能这样扩容；Windows 还存在其他映射导致扩容失败的限制。具体条件见 Python mmap.resize() 文档。

这个方案会串行执行读取，可以先用它保证正确性，再根据实际性能需求改用读写锁。

要保留零拷贝和并发读取，应该怎么做？

零拷贝读取通常通过 memoryview(mm)[offset:end] 获得，但视图依赖映射持续有效。存在导出缓冲区时，CPython 会拒绝 resize() 和 close()，并抛出 BufferError，不会自动等待 reader 结束。这一限制见 CPython 的缓冲区检查实现。

如果继续使用可扩容的单个映射，需要实现完整的借用协议：

reader 获取共享读取权限，再创建视图。
在使用视图的整个期间保留读取权限。
释放视图及其派生视图后，归还读取权限。
writer 获取独占权限，等待所有 reader 退出后，再扩容或关闭映射。

只在创建 memoryview 时加锁、随后立即解锁，无法覆盖视图的整个使用期间。仅增加一个引用计数也不够：开始借用、阻止新 reader 进入、等待现有 reader 退出，都必须由同一套同步机制协调。

对于已有的 segment 设计，可以在创建时确定每个 segment 的容量，写满后切换到新 segment。旧 segment 不再扩容，reader 可以继续持有它的视图；回收时再等待所有借用结束。这更容易同时满足并发读取、零拷贝和存储持续增长的需求。

还需要注意哪些边界？

映射容量与有效日志长度应分开管理。上面的范围检查只能保证读取不超过映射，不能保证读到的区域已经写入完整记录。应在记录写完后，通过同步协议发布已提交长度，reader 只读取已提交范围。

扩容失败后，不能假定文件和映射都恢复到了原状。示例只在成功后更新容量；实际系统还需要在失败时暂停相关操作，检查资源状态并决定是否重新打开。

这些线程锁只协调本进程中遵守协议的调用方。如果其他进程可能缩短文件，或者原生扩展绕过接口访问内存，还需要另外协调文件生命周期。flush() 用于持久化相关操作，不能代替并发同步。