Linux下如何正确清理Python废弃FileLock锁文件？含多进程场景

正如相关资料所述，在Linux系统中，Python的filelock库提供的FileLock在释放后不会自动删除锁文件（Windows下则会自动清理），比如以下代码的断言会执行失败：

import os
from filelock import FileLock

with FileLock("/tmp/test.lock"):
    print("Lock acquired")

assert not os.path.exists("/tmp/test.lock")  # 执行失败

下面针对不同场景给出具体的锁文件清理实现方案：

一、基础单进程场景：封装自定义上下文管理器 #

官方建议在应用层面将锁定逻辑封装到自定义上下文管理器中，在退出上下文时完成锁文件的清理。核心注意点是必须先释放锁，再删除锁文件，避免其他等待锁的进程出现异常。

实现代码 #

import os
from filelock import FileLock
from contextlib import contextmanager

@contextmanager
def cleanup_filelock(lock_path):
    lock = FileLock(lock_path)
    try:
        with lock:
            yield lock
    finally:
        # 锁释放后，检查文件是否存在并尝试删除
        if os.path.exists(lock_path):
            try:
                os.remove(lock_path)
            except OSError:
                # 处理竞态场景：比如其他进程刚获取了该锁
                pass

使用方式 #

with cleanup_filelock("/tmp/test.lock"):
    print("Lock acquired and will be cleaned up")

assert not os.path.exists("/tmp/test.lock")  # 此时断言会成功

二、多进程/PyTorch分布式场景 #

在多进程协作场景（比如PyTorch分布式训练）中，需要确保只有持有锁的进程负责删除锁文件，避免多进程同时删除引发的异常，同时要处理进程意外退出导致的锁文件残留问题。

方案1：适配分布式环境的上下文管理器 #

import os
import torch.distributed as dist
from filelock import FileLock
from contextlib import contextmanager

@contextmanager
def distributed_cleanup_filelock(lock_path):
    lock = FileLock(lock_path)
    is_acquired = False
    try:
        with lock:
            is_acquired = True
            yield lock
    finally:
        # 仅获取到锁的进程尝试删除文件
        if is_acquired and os.path.exists(lock_path):
            # 分布式场景下，优先让主进程（rank=0）执行删除
            if not dist.is_initialized() or dist.get_rank() == 0:
                try:
                    os.remove(lock_path)
                except OSError:
                    pass

使用示例（PyTorch分布式场景） #

# 先初始化分布式环境（根据实际场景调整后端）
dist.init_process_group(backend="nccl")

with distributed_cleanup_filelock("/tmp/distributed_test.lock"):
    if dist.get_rank() == 0:
        print("主进程获取到锁")
    # 执行分布式任务逻辑

dist.destroy_process_group()

方案2：兜底清理无效残留锁文件 #

如果进程意外崩溃，锁文件可能残留。可以在获取锁前先检查锁文件是否对应已退出的进程，清理无效的"死锁"文件：

import os
import psutil
from filelock import FileLock

def is_stale_lock(lock_path):
    """判断锁文件是否为无效的死锁文件"""
    try:
        with open(lock_path, 'r') as f:
            pid = int(f.read().strip())
        return not psutil.pid_exists(pid)
    except (FileNotFoundError, ValueError, psutil.NoSuchProcess):
        return False

def get_clean_lock(lock_path):
    """先清理死锁文件，再返回锁对象"""
    if is_stale_lock(lock_path):
        os.remove(lock_path)
    return FileLock(lock_path)

将这个函数和上下文管理器结合使用，就能避免因进程崩溃导致的锁文件残留问题。

内容的提问来源于stack exchange，提问作者Ricardo Decal