Python でファイルロックを実装する（fcntl, msvcrt, filelock）

複数のプロセスやスレッドが同じファイルにアクセスする場合、ファイルロックを使ってデータの破損を防ぐ必要がある。

ファイルロックが必要な場面

複数のプロセスが同じログファイルに書き込む

設定ファイルを読み書きするアプリケーション

データベース的にファイルを使う場合

fcntl を使う（Unix 系）

Unix 系 OS では fcntl モジュールでファイルロックを取得できる。

import fcntl

with open('data.txt', 'r+') as f:
    # 排他ロックを取得
    fcntl.flock(f.fileno(), fcntl.LOCK_EX)
    
    try:
        content = f.read()
        f.seek(0)
        f.write(content + '\nnew line')
        f.truncate()
    finally:
        # ロックを解放
        fcntl.flock(f.fileno(), fcntl.LOCK_UN)

ロックの種類は以下の通りだ。

定数	意味
LOCK_SH	共有ロック（読み取り用、複数プロセスが同時に取得可）
LOCK_EX	排他ロック（書き込み用、1プロセスのみ取得可）
LOCK_NB	ノンブロッキング（ロック取得に失敗したら即座にエラー）
LOCK_UN	ロック解放

msvcrt を使う（Windows）

Windows では msvcrt モジュールを使う。

import msvcrt
import os

with open('data.txt', 'r+') as f:
    # ファイルの一部をロック
    msvcrt.locking(f.fileno(), msvcrt.LK_LOCK, os.path.getsize('data.txt'))
    
    try:
        content = f.read()
        f.seek(0)
        f.write(content + '\nnew line')
        f.truncate()
    finally:
        f.seek(0)
        msvcrt.locking(f.fileno(), msvcrt.LK_UNLCK, os.path.getsize('data.txt'))

filelock ライブラリを使う（クロスプラットフォーム）

filelock はクロスプラットフォームで動作するサードパーティライブラリだ。

pip install filelock

from filelock import FileLock

lock = FileLock('data.txt.lock')

with lock:
    with open('data.txt', 'r+') as f:
        content = f.read()
        f.seek(0)
        f.write(content + '\nnew line')
        f.truncate()

ロックファイル（.lock）を別に作成する方式で、元のファイルを直接ロックするわけではない。

タイムアウト付きロック

ロック取得を待ち続けるとデッドロックになる可能性がある。タイムアウトを設定すべきだ。

from filelock import FileLock, Timeout

lock = FileLock('data.txt.lock')

try:
    with lock.acquire(timeout=10):
        # 10秒以内にロックを取得できれば実行
        with open('data.txt', 'a') as f:
            f.write('new line\n')
except Timeout:
    print('ロック取得がタイムアウトしました')

アドバイザリロックの注意点

Python のファイルロックはアドバイザリロック（勧告的ロック）であり、他のプロセスがロックを無視してアクセスすることを強制的に防ぐわけではない。すべてのプロセスが同じロック機構を使う必要がある。

# プロセス A がロックを取得
with FileLock('data.lock'):
    with open('data.txt', 'w') as f:
        f.write('from A')

# プロセス B がロックを無視（これは防げない）
with open('data.txt', 'w') as f:
    f.write('from B')

コンテキストマネージャでラップする

自前でファイルロッククラスを作る場合の例を示す。

import fcntl
from contextlib import contextmanager

@contextmanager
def locked_file(path, mode='r'):
    with open(path, mode) as f:
        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
        try:
            yield f
        finally:
            fcntl.flock(f.fileno(), fcntl.LOCK_UN)

# 使用例
with locked_file('data.txt', 'a') as f:
    f.write('safe write\n')