json_file_lock.py

"""同一数据目录内的稳定协调锁，不锁会被原子替换的JSON inode。"""
import hashlib
import os
import sys
import threading
import time


class DataFileLock:
    def __init__(self, directory, file_name, timeout=30):
        if (not isinstance(file_name, str) or not file_name or file_name in ('.', '..')
                or any(char in file_name for char in '/\\:') or any(ord(char) < 32 for char in file_name)):
            raise ValueError('invalid data file name')
        self.directory = os.path.abspath(directory)
        self.path = os.path.join(self.directory, '.locks', hashlib.sha256(os.path.normcase(file_name).encode()).hexdigest() + '.lock')
        self.timeout = timeout
        self.thread_lock = threading.Lock()
        self.handle = None

    def __enter__(self):
        self.thread_lock.acquire()
        handle = None
        try:
            os.makedirs(os.path.dirname(self.path), exist_ok=True)
            handle = open(self.path, 'a+b')
            # Windows允许锁定EOF之外的字节；不初始化写入，避免首次并发争用锁区。
            deadline = time.monotonic() + self.timeout
            while True:
                try:
                    if sys.platform == 'win32':
                        import msvcrt
                        handle.seek(0)
                        msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
                    else:
                        import fcntl
                        fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
                    break
                except OSError as error:
                    import errno
                    if error.errno not in (errno.EACCES, errno.EAGAIN, errno.EDEADLK):
                        raise
                    if time.monotonic() >= deadline:
                        raise TimeoutError('data coordination lock timed out') from None
                    time.sleep(.05)
            self.handle = handle
            return self
        except BaseException:
            if handle is not None:
                handle.close()
            self.thread_lock.release()
            raise

    def __exit__(self, exc_type, exc_value, traceback):
        handle, self.handle = self.handle, None
        try:
            if sys.platform == 'win32':
                import msvcrt
                handle.seek(0)
                msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
            else:
                import fcntl
                fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
        finally:
            try:
                handle.close()
            finally:
                self.thread_lock.release()
        return False