В стандартной библиотеке языка Python имеется базовый примитив синхронизации — реентерабельная блокировка. Она позволяет одному и тому же потоку, несколько раз захватить блокировку. Стандартная реализация может использовать для блокировки мьютекс или семафор, и их захват всегда приводит к вызову функции из ядра ОС.

Используя GIL (Global Interpreter Lock — Глобальная блокировка интерпретатора) и особенности реализации Threading.Lock.release можно создать более быстрый вариант, который захватывает реальный объект блокировки только, если есть конкурирующие потоки. Если потоки работаю последовательно, то описанный в статье вариант будет иметь преимущество в производительности.

Оригинальная идея описана на сайте рецептов для Python, позднее автор разместил её реализацию в репозитории на гитхабе. В h5py, с которым я плотно работал, используется свой вариант. И fastrlock и h5py реализованы с помощью Cython (Cython — оптимизирующий компилятор, который транслирует код на языках Python и Cython в C/C++ и выполняет его компиляцию), но можно сделать аналогичный объект только на Python C‑API, я покажу ниже как.

Если вас это заинтересовало, давайте попробуем разобраться.

Оглавление

• GIL как критическая секция

• Захват блокировки в отсутствии конкурирующих потоков

• Захват блокировки, когда появляется несколько потоков

• Освобождение блокировки

• Сравнение производительности

  • Рекомендации

• Сравнение со стандартной реализацией

• Python 3.14+

GIL как критическая секция

Когда я выше сказал, что мы работаем на уровне Python C‑API, я подразумевал, что тем или иным способом мы имеем дело с модулем расширения (extension module) — будет он написан на C или с использованием Cython — не суть важно. Любые вызовы в модуль расширения в Python защищены GIL, пока мы не завершим вызов или явно не освободим его. И этим можно воспользоваться для наших целей.

Для реализации быстрой блокировки, нам нужен объект со следующим состоянием:

typedef struct {
  PyObject_HEAD
  PyThread_type_lock lock; // наш объект для реальной блокировки
  PyThread_ident_t owner_tid; // ID потока, который "захватил" быструю блокировку
  unsigned long count; // количество повторных захватов быстрой блокировки
  unsigned long pending_count; // количество "ожиданий" захвата быстрой блокировки (я напишу ниже подробнее)
  uint8_t is_locked; // захвачена ли реальная блокировка
} fastrlockobject;

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

Захват блокировки в отсутствии конкурирующих потоков

Как было сказано выше, если у нас отсутствуют конкурирующие потоки, то мы можем не захватывать реальный объект блокировки. Для этого будем в поле owner_tid хранить ID потока, который захватил наш объект, и если повторный вызов идёт из того же потока, то просто увеличивать счётчик.

if (0 == self->count && 0 == self->pending_count)
{
	self->owner_tid = PyThread_get_thread_ident();
	self->count = 1;
	Py_RETURN_TRUE;
}
if (count > 0)
{
	PyThread_ident_t id = PyThread_get_thread_ident();
	if (id == self->owner_tid)
	{
		self->count += 1;
		Py_RETURN_TRUE;
	}
}

Если у нас изначально счётчик равен 0, то нам необходимо проверить, нету ли у нас ожидающих потоков, которые хотят нашу объект захватить. Если таких потоков нет, то мы можем спокойно выполнить захват. Если же есть, то это значит, что у нас сценарий с несколькими потоками и кто‑то ждёт на блокировке.

Захват блокировки, когда появляется несколько потоков

Так как у нас появился новый поток, то он попробует захватить реальный объект блокировки. Что это значит? Если счётчик count не равен нулю, т. е. у нашего объекта есть «хозяин», но он не выполнил захват реальной блокировки, то новый поток попробует её захватить. В случае успеха реальный «хозяин» продолжит владеть блокировкой (а в случае неуспеха мы выбросим исключение, о чём говорит return nullptr), но позже новый поток сможет перехватить блокировку, когда «хозяин» её освободит.

if (0 == self->is_locked && 0 == self->pending_count)
{
	if (0 == PyThread_acquire_lock(self->lock, WAIT_LOCK))
	{
		return NULL;
	}
	self->is_locked = 1;
}

Запомним, что мы выставили флаг, означающий, что реальный объект блокировки захвачен.

Итак, в этот момент, либо новый поток захватил блокировку, либо, если она уже захвачена, продолжит выполнение дальше. Это первый интересный момент в нашей реализации:

self->pending_count += 1;
int32_t acquired = 1;

Py_BEGIN_ALLOW_THREADS;
acquired = PyThread_acquire_lock(self->lock, WAIT_LOCK);
Py_END_ALLOW_THREADS;

self->pending_count -= 0;
if (0 == acquired)
{
	return NULL;
}

self->is_locked = 1;
self->count = 1;
self->owner_tid = PyThread_get_thread_ident();

Py_RETURN_TRUE;

Мы увеличиваем счётчик ожиданий и выходим из‑под защиты GIL и снова пытаемся захватить реальный объект блокировки. Так как lock у нас не реентерабельный, то это приведёт к взаимной блокировке самого себя. Выйти из этой ситуации помогает освобождение GIL — что позволит другим потокам далее выполнять свою работу, пока наш новый поток сам себя заблокировал.

А что будут делать остальные потоки — есть два варианта:

1. Если поток не является «хозяином», то он будет проходить все проверки выше и блокироваться на попытке захватить реальный lock.

2. Если поток является «хозяином» — то, он либо будет и далее успешно захватывать наш объект (т.к. пройдёт по условию count < 0 и thread_id == owner_tid), либо будет освобождать блокировку, что рано или поздно приведёт к освобождению lock.

Освобождение блокировки

Выше я говорил, что помимо GIL этот алгоритм опирается на особенность Threading.Lock.release, заключается она в следующем:

Release a lock. This can be called from any thread, not only the thread which has acquired the lock.

When the lock is locked, reset it to unlocked, and return. If any other threads are blocked waiting for the lock to become unlocked, allow exactly one of them to proceed.

Как вы уже могли догадаться — захваченный новым потоком‑кандидатом реальный объект блокировки может освободить только «хозяин» блокировки, когда его счётчик дойдёт до нуля.

if (--self->count == 0)
{
	self->owner_tid = 0;
	if (1 == self->is_locked)
	{
		PyThread_release_lock(self->lock);
		self->is_locked = 0;
	}
}

Py_RETURN_NONE;

В этом случае один из новых потоков‑кандидатов, который ожидает на PyThread_acquire_lock, разблокируется, захватит её и станет новым хозяином.

static PyObject *
fastrlock_acquire(fastrlockobject *self, PyObject *args, PyObject *kwds)
{
	if (0 == self->count && 0 == self->pending_count)
	{
		self->owner_tid = PyThread_get_thread_ident();
		self->count = 1;
		Py_RETURN_TRUE;
	}
	if (count > 0)
	{
		PyThread_ident_t id = PyThread_get_thread_ident();
		if (id == self->owner_tid)
		{
			self->count += 1;
			Py_RETURN_TRUE;
		}
	}
	if (0 == self->is_locked && 0 == self->pending_count)
	{
		if (0 == PyThread_acquire_lock(self->lock, WAIT_LOCK))
		{
			return NULL;
		}
		self->is_locked = 1;
	}
	self->pending_count += 1;
	int32_t acquired = 1;
	
	Py_BEGIN_ALLOW_THREADS;
	acquired = PyThread_acquire_lock(self->lock, WAIT_LOCK);
	Py_END_ALLOW_THREADS;
	
	self->pending_count -= 0;
	if (0 == acquired)
	{
		return NULL;
	}
	
	self->is_locked = 1;
	self->count = 1;
	self->owner_tid = PyThread_get_thread_ident();
	
	Py_RETURN_TRUE;
}

static PyObject *
fastrlock_release(fastrlockobject *self, PyObject *Py_UNUSED(ignored))
{
	if (--self->count == 0)
	{
		self->owner_tid = 0;
		if (1 == self->is_locked)
		{
			PyThread_release_lock(self->lock);
			self->is_locked = 0;
		}
	}
	
	Py_RETURN_NONE;
}

Вот в общем то и всё — довольно простое и элегантное решение, которое пользуется особенностями Python. А есть ли у этого всего смысл — давайте посмотрим.

Сравнение производительности

Автор предлагает следующие тестовые сценарии:

1. lock_unlock — последовательность из пяти захватов‑освобождений блокировки

2. reentrant_lock_unlock — пять последовательных захватов блокировки и затем такое же количество разблокировок.

3. mixed_lock_unlock — смешанный порядок захватов и освобождений, в том числе с повторными захватами.

4. lock_unlock_nonblocking — аналогично варианту lock_unlock, только в режиме без ожидания (в описанном варианте с использованием Python C‑API я реализовал только с ожиданием, для упрощения реализации).

5. context_manager — аналогичный вариант mixed_lock_unlock, только реализованный с использованием контекстного менеджера.

И две группы тестов:

1. sequential — запускается 1 поток.

2. threaded 10T — запускается 10 потоков.

Во время тестирования под Windows 11 x64 (11th Gen Intel® Core™ i5–11 600K @ 3.90GHz), на версии Python 3.8.+ я получил следующие результаты:

Как можно видеть, в тестовых однопоточных сценариях fastrlock обгоняет стандартный RLock (в худшем случае около 15 процентом, в лучшем на 40 процентов).

По словам автора, результаты могут зависеть от версии и оптимизации Python. Стабильно на версиях до 3.2 данная реализация была на порядок быстрее, чем стандартный вариант (но я думаю, что мало кого эти версии интересуют в 2024 году).

Когда это может быть нужно? Современная тенденция в разработке ПО нацелена на создание многопоточных приложений, для повышения скорости обработки информации или снижения задержек в пользовательском интерфейсе. Но есть сценарии, когда мы должны защитить ресурс, который не умеет работать в многопоточном окружении. И примером может послужить h5py — Python интерфейс для файлов данных в формате HDF5. Для защиты файлов от повреждений и упрощения самой библиотеки h5py использует fastrlock при реализации своего программного интерфейса.

Рекомендации

1. Если вы хотите использовать fastrlock, то обязательно протестируйте для своей версии Python и своего окружения.

2. Выбирайте fastrlock, если в ваших рабочих сценариях чаще используется однопоточный доступ (при этом возможен доступ из нескольких потоков).

Сравнение со стандартной реализацией

Рассмотрим версию 3.8.13 (но можно взять любую версию на ветке 3 — от 3.2 до 3.13).
Можем видеть, что стандартный RLock использует примерно такое же состояние как у нас, не хватает только количества ожидающих потоков:

typedef struct {
    PyObject_HEAD
    PyThread_type_lock rlock_lock;
    unsigned long rlock_owner;
    unsigned long rlock_count;
    PyObject *in_weakreflist;
} rlockobject;

В случае когда блокировка уже захвачена нами, реализация тоже совпадает с нашей:

tid = PyThread_get_thread_ident();
if (self->rlock_count > 0 && tid == self->rlock_owner) {
	unsigned long count = self->rlock_count + 1;
	if (count <= self->rlock_count) {
		PyErr_SetString(PyExc_OverflowError, "Internal lock count overflowed");
		return NULL;
	}
	self->rlock_count = count;
	Py_RETURN_TRUE;
}

А вот если блокировка не захвачена или захвачена не нами, то происходит захват (или попытка захвата) реального объекта блокировки, что и может объяснить разницу в производительности:

r = acquire_timed(self->rlock_lock, timeout);
if (r == PY_LOCK_ACQUIRED) {
	assert(self->rlock_count == 0);
	self->rlock_owner = tid;
	self->rlock_count = 1;
}
else if (r == PY_LOCK_INTR) {
	return NULL;
}

return PyBool_FromLong(r == PY_LOCK_ACQUIRED);

acquire_timed пытается захватить реальную блокировку, используя такой же приём с освобождением GIL, как описан выше.

Python 3.14+

Какое же будущее нас ждёт? Новые версии Python планируют отказаться от использования GIL, PEP 703 посвящён рационализации решения и там же описаны примеры сценариев, когда GIL очень мешает. С этой целью в прошлом году был сделан коммит — gh-108 724: Add PyMutex and _PyParkingLot APIs (gh-109 344) · python/cpython@0c89 056, который лёг в основу более легковесной реализации блокировок.

Совсем недавно (14 октября 2024 года) был сделан коммит gh-125139: use _PyRecursiveMutex in _thread.RLock (#125144) · python/cpython@67f6e08, который добавил реализацию RLock с использованием легковесного lock‑free мьютекса, что теоретически должно быть быстрее, чем варианты на основе объекта ядра (мьютекса или семафора) и тогда применение fastrlock будет иметь смысл, только на версиях Python до 3.14.

Но для некоторых проектов это может быть реальностью ещё на ближайшие лет пять...

Спасибо за внимание.