Python中SharedMemory在Debian上出现异常关闭问题求助

问题背景 #

尝试通过multiprocessing.shared_memory实现程序单实例运行，代码在Windows上正常，但在Debian 11（Python 3.9.2）下出现异常行为。

原始实现代码 #

import multiprocessing.shared_memory as share_mem
import sys

def isRunning(name: str) -> bool:
    try:
        shm = share_mem.SharedMemory(name, False)
        shm.close()
        return True
    except FileNotFoundError as e:
        return False

def runInstance(name: str) -> share_mem.SharedMemory:
    return share_mem.SharedMemory(name, True, 1)

NAME = 'foo.bar'

if isRunning(NAME):
    print('Already launched, exit')
    sys.exit(1)

print('Launching instance')
shm = runInstance(NAME)
print('Running')
input('Press key')
print('Quitting')
shm.close()
shm.unlink()

异常现象 #

• 终端1启动python3 si.py，程序正常启动并等待输入
• 终端2首次启动python3 si.py，检测到已有实例并退出
• 终端2再次启动python3 si.py，程序成功启动（单实例限制失效）
• 终端1按下回车后报错 FileNotFoundError: no such file /foo.bar
• 偶现两个实例同时正常运行的情况

问题根源 #

这是Python 3.9.x版本中multiprocessing.resource_tracker与shared_memory交互的已知bug，而非代码逻辑或系统问题：

• 当第二个实例调用isRunning时，会临时打开已存在的共享内存对象并调用shm.close()
• Python的资源追踪器会将这个临时打开的共享内存标记为待清理资源
• 当第二个实例退出时，资源追踪器会自动执行unlink操作，销毁第一个实例持有的共享内存
• 后续实例检测时，共享内存已不存在，因此允许新实例启动

尝试的修复及局限性 #

根据建议修改isRunning函数，跳过资源追踪：

import multiprocessing.shared_memory as share_mem
import sys
import multiprocessing.resource_tracker

def isRunning(name: str) -> bool:
    try:
        shm = share_mem.SharedMemory(name, False)
        # 避免资源追踪器在本进程退出时销毁共享内存
        multiprocessing.resource_tracker.unregister(shm._name, 'shared_memory')
        shm.close()
        return True
    except FileNotFoundError as e:
        return False

def runInstance(name: str) -> share_mem.SharedMemory:
    return share_mem.SharedMemory(name, True, 1)

NAME = 'foo.bar'

if isRunning(NAME):
    print('Already launched, exit')
    sys.exit(1)

print('Launching instance')
shm = runInstance(NAME)
print('Running')
input('Press key')
print('Quitting')
shm.close()
shm.unlink()

但该修复存在严重局限性：

• 当程序异常崩溃（如强制杀死进程）时，资源追踪器会输出错误信息并暂停执行，必须手动按键才能继续，完全无法满足后台运行的需求，因此不得不放弃共享内存方案。

可靠替代方案：文件锁实现单实例 #

Linux下推荐使用fcntl文件锁实现单实例运行，该方式无资源追踪bug问题，且进程崩溃时内核会自动释放锁，无需手动清理：

import fcntl
import sys
import os

LOCK_FILE = '/tmp/myapp.lock'

def is_running():
    try:
        # 打开锁文件，支持写入PID用于排查
        lock_file = open(LOCK_FILE, 'w')
        # 尝试获取排他锁，非阻塞模式
        fcntl.flock(lock_file, fcntl.LOCK_EX | fcntl.LOCK_NB)
        lock_file.write(str(os.getpid()))
        lock_file.flush()
        return False, lock_file
    except IOError:
        # 无法获取锁，说明已有实例运行
        return True, None

def main():
    running, lock_file = is_running()
    if running:
        print('Already launched, exit')
        sys.exit(1)
    
    print('Launching instance')
    print('Running')
    try:
        input('Press key')
    finally:
        # 释放锁并清理锁文件
        if lock_file:
            fcntl.flock(lock_file, fcntl.LOCK_UN)
            lock_file.close()
            os.unlink(LOCK_FILE)
    print('Quitting')

if __name__ == '__main__':
    main()

内容的提问来源于stack exchange，提问作者Antonius Hart