Именованный мьютекс в Windows через ctypes: как избежать параллельных запусков на Python

Почему именованный мьютекс в Windows из Python через ctypes не виден между процессами, и как исправить: Global\, CreateMutexW и корректный GetLastError.

Когда приложению на Python нужно согласованно управлять доступом к единственному аппаратному устройству из нескольких процессов, логичным выбором будет именованный мьютекс. Идея проста: каждый запущенный экземпляр пытается захватить мьютекс, привязанный к идентификатору устройства; если мьютекс уже существует, процесс понимает, что устройство занято, и завершает работу. Но в Windows именованный мьютекс виден другим процессам только при корректном именовании и корректном чтении ошибок. Иначе можно получить несколько параллельно работающих экземпляров, каждый уверенный, что владеет устройством.

Воспроизведение проблемы

Следующий минимальный пример пытается защитить устройство с помощью именованного мьютекса, но не останавливает параллельные экземпляры. Оба процесса спокойно работают и выводят сбивающий с толку код ошибки вместо того, чтобы обнаружить конфликт.

import ctypes as c
import sys

ERR_ALREADY_EXISTS = 183

sync_handle = c.windll.kernel32.CreateMutexW(None, True, "my-app\\1234")
last_err = c.GetLastError()
print(f"Error code received: {last_err}")

if last_err == ERR_ALREADY_EXISTS:
    print("Another instance is already using the device.")
    sys.exit(0)

input("Program running. Press Enter to exit...")

В чем на самом деле ошибка

Первый виновник — имя мьютекса. Windows использует специальные пространства имен для именованных объектов ядра. Нельзя придумывать произвольные префиксы вроде my-app\.... Чтобы мьютекс был виден между процессами, используйте одно из поддерживаемых пространств имен. В нашем случае правильный выбор — Global\; при необходимости подойдет и Local\. Без поддерживаемого пространства имен процессы не увидят один и тот же объект, и синхронизация сорвется.

Вторая проблема — способ чтения кода ошибки. С ctypes библиотеку DLL нужно загружать с параметром use_last_error=True и явно объявлять прототипы функций. Это гарантирует, что GetLastError() считывается сразу после вызова, а ошибочные значения превращаются в исключения вместо тихого возврата двусмысленных результатов.

Исправление

Исправленный подход использует пространство имен Global\, чтобы все процессы обращались к одному и тому же имени мьютекса, а также настраивает вызовы Win32 с корректными типами, обработкой ошибок и надежным доступом к GetLastError().

import ctypes as c
import ctypes.wintypes as wt

ERR_EXISTS = 183

# Преобразуем ошибочные результаты в исключения для предсказуемого управления потоком выполнения.
def _raise_on_null(res, func, args):
    if not res:
        raise c.WinError(c.get_last_error())
    return res

def _raise_on_zero(res, func, args):
    if res == 0:
        raise c.WinError(c.get_last_error())

# Фиксируем значение GetLastError() сразу после каждого вызова через ctypes.
k32 = c.WinDLL("kernel32", use_last_error=True)

# Объявляем сигнатуры вызываемых нами функций Win32 API.
_CreateMutexW = k32.CreateMutexW
_CreateMutexW.argtypes = (c.c_void_p, wt.BOOL, wt.LPCWSTR)
_CreateMutexW.restype = wt.HANDLE
_CreateMutexW.errcheck = _raise_on_null

_CloseHandle = k32.CloseHandle
_CloseHandle.argtypes = (wt.HANDLE,)
_CloseHandle.restype = wt.BOOL
_CloseHandle.errcheck = _raise_on_zero

try:
    hm = _CreateMutexW(None, True, "Global\\1234")
except OSError as exc:
    print(exc)
else:
    try:
        if c.get_last_error() == ERR_EXISTS:
            print("Another instance is already using the device.")
        else:
            input("Program running. Press Enter to exit...")
    finally:
        _CloseHandle(hm)

Почему это важно

Конфликты доступа к устройствам трудно отлаживать: они часто проявляются только под нагрузкой или на машинах пользователей. Использование правильного пространства имен гарантирует, что каждый процесс соревнуется за один и тот же примитив синхронизации. Корректные объявления типов и обработка ошибок вокруг вызовов Win32 предотвращают вводящие в заблуждение диагностические сообщения и делают поведение детерминированным, что снижает нагрузку на поддержку и количество отказов в поле.

Выводы

Используйте поддерживаемое пространство имен для именованных объектов ядра; для межпроцессной видимости в этом контексте подойдет Global\. Связывайте функции Win32 в ctypes с явными argtypes и restype, включайте use_last_error=True и преобразуйте результаты неудачных вызовов в исключения. При выполнении этих двух условий правило «один владелец» на основе мьютекса для аппаратных устройств будет работать как задумано, а приложение будет предсказуемо вести себя при запуске нескольких экземпляров.