Глава 21 Очереди без блокировок: SPSC / MPSC / MPMC

Ориентация главы

Очередь — единственный потокобезопасный контейнер тома 3 — она и родилась для передачи данных сквозь потоки, поэтому она же мост между томом 3 и томом 6 (конкурентность): глава учит употреблению и дисциплине контейнера, а том 6 встроит его в полную систему потоков, корутин и Channel — сперва освоите одиночное оружие здесь, строй наступит позже. XRT даёт три топологии: SPSC (один производитель, один потребитель — самая быстрая), MPSC (много производителей, один потребитель — стандартная форма агрегации событий) и MPMC (любые многим-к-многим — самая общая и самая дорогая). Три делят один язык употребления: привязка внешних слотов (ноль внутренних выделений), пакетная передача (PushBatch/PopBatch), протокол закрытия-остановки (после Close производители останавливаются, потребители выбирают остаток). Главба вдавливает эти три вещи и даёт дисциплину выбора «топология прежде всего» — за универсальность есть цена; можете определить топологию — не платите за неё.

21.1引入

Поток журналирования принимает события журналов от всех рабочих потоков — это MPSC: несколько производителей, один потребитель. Поток рендера и поток декодирования передают кадры один-к-одному — это SPSC. Диспетчер задач позволяет любому потоку подавать и любому забирать — это MPMC. Если все три сцены решить «одним мьютексом поверх обычной очереди», заработает — но на пути миллионов передач в секунду конкуренция за замок стирает преимущество многоядерности — восемь ядер из десяти, ждущих замок, — заурядное побоище; а разница запертого и свободного от блокировок решений выдаст конкретные цифры вашей машины в сравнительном упражнении главы.

Очереди без блокировок заменяют замки атомарными операциями: производитель и потребитель каждый продвигает собственный указатель и публикует его атомарно для другой стороны — SPSC обходится даже двумя атомарными указателями головы и хвоста и делает миллионы одиночных передач в секунду по медиане; MPMC обязан синхронизировать публикацию каждого слота — самая дорогая поштучно, но пакетный интерфейс продавливает амортизированную стоимость вниз (репозиторный базис: пакет 32 — 134 миллиона элементов в секунду). Чем определённее топология, тем меньше синхронизации — это и есть «топология прежде всего»: агрегация журналов определённо MPSC — берите MPSC; не тяните MPMC «на всякий случай».

21.2概念

Три топологии в одной таблице

Три общих идиомы

Привязка слотов значит: хранилище очереди предоставляете вы (стековый массив, статическая память, пул — что угодно) — очередь даёт только синхронизацию и индексную логику, и это одна из форм, покрываемых контрактным тестом _noalloc. Возвращаемое значение пакетного интерфейса — «сколько реально завершено» (Count у xqueuebatchresult) — очередь может быть ни полна ни пуста и всё же завершить запрос частично; возврат сообщает, докуда вы дошли. Закрытие и остановка — сердце протокола завершения конкурентной программы: Close вызывается только после «подтверждения, что все производители закончили»; уже поставленные в очередь данные не теряются, потребители продолжают выбирать остаток.

FIFO и голодание

Все три очереди — FIFO: первым пришёл — первым ушёл, порядок сохраняется сквозь партии. Когда несколько потребителей MPMC конкурентно выполняют PopBatch, очередь внутренне гарантирует ровно одно получение каждого сообщения — ни повторной доставки, ни координации потребителей. «Справедливость» обещана не быть: высокочастотный потребитель может хронически забирать больше — нужна взвешенная справедливость, делайте её на бизнесовом уровне (счётчики квот); от очереди не ждите.

Отношение к тому 6

Глава учит только «передаче указателей между потоками одной машины» — межпроцессная передача принадлежит территории сети и сериализации тома 7. Том 6 поставит очередь в полную систему конкурентности: кооперацию с планировщиком корутин (блокирующее ожидание превращается в ожидаемый Pop), Channel как обёртку «очередь с семантикой закрытия», и когда вместо очереди нужен Future. Сейчас запомните одно: очереди передают ptr — жизненным циклом объекта-сообщения распоряжается отправитель на выделение и получатель на обработку; за границей очереди отправитель больше не освобождает.

21.3示例

Первая полная программа: пакетная передача и остановка MPMC

Из репозиторного образца examples/containers/queue_mpmc/main.c:

 */

#include <stdio.h>
#include <xrt.h>



int main(void)
{
	xmpmcqueue Queue;
	xqueueslot Storage[8];
	int pValues[] = { 10, 20, 30, 40 };
	ptr pFirstBatch[] = { &pValues[0], &pValues[1] };
	ptr pSecondBatch[] = { &pValues[2], &pValues[3] };
	ptr pOutput[4];
	xqueuebatchresult Batch;

	/* 绑定外部槽位；MPMC 槽位与 MPSC 同类型（xqueueslot）。 */
	if ( !xrtMPMCQueueInitBuffer(&Queue, Storage, 8u) ) {
		return 1;
	}

	/* 两批各 2 条：并发场景下任意线程都可这样批量发布。 */
	if ( xrtMPMCQueuePushBatch(&Queue, pFirstBatch, 2u).Count != 2u ) {
		xrtMPMCQueueUnit(&Queue);
		return 2;
	}
	if ( xrtMPMCQueuePushBatch(&Queue, pSecondBatch, 2u).Count != 2u ) {
		xrtMPMCQueueUnit(&Queue);
		return 3;
	}

	


	 */
	xrtMPMCQueueClose(&Queue);

	


	 */
	Batch = xrtMPMCQueuePopBatch(&Queue, pOutput, 4u);
	for ( size_t i = 0; i < Batch.Count; i++ ) {
		printf("%d\n", *(int*)pOutput[i]);
	}
	xrtMPMCQueueUnit(&Queue);
	return Batch.Count == 4u ? 0 : 4;
}

Что только что произошло. (1) xqueueslot Storage[8] — массив слотов, предоставленный вызывающим, привязан InitBuffer(&Queue, Storage, 8) — ёмкость 8 решается длиной массива, очередь сама без выделений. (2) Два PushBatch по 2 элемента: возврат .Count обязан сверяться — «попросил 2, сколько реально вошло» при загрузке — настоящий вопрос, программа движется по фактическому числу. (3) Close вызывается после окончания всех производителей — в однопоточном образце он следует сразу; в многопоточной программе он принадлежит протоколу остановки (дождаться join потоков-производителей, потом Close). (4) PopBatch забирает 4 разом, вывод строго FIFO: 10, 20, 30, 40 — порядок сохраняется сквозь партии. При конкурентном исполнении одного PopBatch несколькими потребителями внутренняя часть гарантирует ровно-одно на элемент. (5) Сообщения — ptr: &pValues[i] в очереди указывает в стековый массив — получатель только заимствует, не освобождает; дисциплина принадлежности та же, что в главе 5.

Вторая полная программа: минимальная форма SPSC

Из examples/containers/queue_spsc/main.c, самый быстрый путь топологии один-к-одному:

 */

#include <stdio.h>
#include <xrt.h>



int main(void)
{
	xspscqueue Queue;
	ptr Storage[8];          /* 槽位由调用方提供：地址稳定、零分配 */
	int pValues[] = { 10, 20, 30 };
	ptr pValue;

	/* 绑定外部槽位：容量 8，Push/Pop 全在这块内存上进行。 */
	if ( !xrtSPSCQueueInitBuffer(&Queue, Storage, 8u) ) {
		return 1;
	}

	/* 依次入队三个 int 指针（队列搬运的是 ptr，不拥有对象）。 */
	for ( size_t i = 0; i < 3u; i++ ) {
		if ( xrtSPSCQueueTryPush(&Queue, &pValues[i]) != XQUEUE_OK ) {
			xrtSPSCQueueUnit(&Queue);
			return 2;
		}
	}

	



	 */
	xrtSPSCQueueClose(&Queue);
	while ( xrtSPSCQueueTryPop(&Queue, &pValue) == XQUEUE_OK ) {
		printf("%d\n", *(int*)pValue);
	}
	xrtSPSCQueueUnit(&Queue);
	return 0;
}

Что только что произошло. Форма интерфейса полностью изоморфна MPMC — привязанные слоты, пакетная передача, закрытие и остановка, все три на месте — различие целиком внутреннее: у SPSC только один атомарный указатель головы и один хвоста; производитель пишет голову, потребитель пишет хвост, и они не конкурируют. На тот же объём сообщений её поштучная цена наименьшая — вот исполнение «специализированная очередь при определённой топологии». Прочтите интерфейсные вызовы двух образцов бок о бок: выучите одно употребление — им управляются все три очереди; это замысел API семейства очередей.

Ёмкость впервые встречается с противодавлением

Ёмкость очереди (длина массива слотов) — не «чем больше, тем лучше»: это первый клапан противодавления — при полной очереди PushBatch завершается частично, и производитель, почувствовав, что не двигается, должен сбавить темп или буферизовать — точная миниатюра сетевого противодавления главы 68. Обе крайности планирования ёмкости платят: слишком мала — пиковой трафик забивает мгновенно, производители простаивают в повторах; слишком велика — резидентная память растёт, а при отказе потребителя накоплений больше и восстановление дольше. Практическая стартовая точка: ёмкость в 2–4 раза больше того, что потребитель обрабатывает за один цикл планирования, затем калибруйте статистикой главы 6 и фактическим уровнем очереди. Count очереди — публичное поле — проверка «почти полна» у производителя ничего не стоит.

Дисциплинарная граница свободного от блокировок

Свободное от блокировок значит «внутри без мьютексов», но не «конкурентно всё позволено» — две дисциплины остаются. Первая: оба конца SPSC каждый монополен — один поток-производитель и один поток-потребитель; многопоточный Push в одну SPSC — неопределённое поведение, это сцена MPSC. Вторая: порядок публикации данных внутри слотов очередь гарантирует, но ваши собственные объекты-сообщения обязаны быть дописаны до постановки и перезаписаны только после выборки — конкурентная запись через границу очереди не ответственность очереди. Держите эти две дисциплины крепко — и скорость свободных от блокировок очередей по-настоящему ваша.

21.4契约

• Привязанное хранилище: массив слотов даёт вызывающий; ёмкость = длина массива слотов; очередь без внутренних выделений.
• Пакетная сверка: возврат Push/Pop Count — фактически завершённое; частичный успех возможен и при полной, и при пустой.
• Протокол остановки: Close один раз после окончания всех производителей; поставленные данные можно доизбрать; Push после Close не удаётся.
• Ровно одно: при конкурентных Pop нескольких потребителей каждое сообщение получается ровно раз — внешняя дедупликация не нужна.
• Принадлежность сообщений: очереди передают ptr; объекты создаёт отправитель и обрабатывает получатель, дисциплина принадлежности как в главе 5; дописать до постановки, перезаписывать после выборки.
• Топология прежде всего: можете определить один-к-одному или многие-к-одному — берите SPSC/MPSC; MPMC берёт наивысшую поштучную цену за универсальность.

Быстрый выбор: три вопроса решают очередь

На практике задайте себе три вопроса по порядку. Первый: сколько конвейеров? Два потока один-к-одному — SPSC, конец. Второй: кто агрегирует? Много источников сходятся в один обработчик (журналы, мониторинг, цикл событий) — MPSC. Третий: действительно любые-к-любым? Много сцен, записанных как MPMC, при внимании — «много производителей + фиксированный пул потребителей»: число потребителей фиксировано и задачи равноценны — MPSC в диспетчер и SPSC от диспетчера каждому потребителю, два яруса специализированных очередей нередко быстрее и яснее одного яруса MPMC. Что осталось после трёх ответов — настоящие сцены MPMC: динамическое вытягивание работы, безцентровые графы задач. Запишите эти три вопроса-ответа в раздел выбора проектной документации — ревью будет с чем сверять.

Типичные ошибки

xrtMPMCQueuePushBatch(&Queue, pMessages, 32u);   /* возвращаемое значение отброшено */
/* очередь могла взять только первые N - остальные 32-N молча застревают в буфере отправки */

xqueuebatchresult Sent = xrtMPMCQueuePushBatch(&Queue, pMessages, 32u);
if ( Sent.Count < 32u ) {
	RetryLater(pMessages + Sent.Count, 32u - Sent.Count);   /* продолжить с точки разрыва */
}

/* главный поток: не ждёт рабочих */
xrtMPMCQueueClose(&Queue);
/* какой-то рабочий в этот момент ещё Push-ит - протокол остановки сломан */

for ( i = 0; i < WorkerCount; ++i ) {
	xrtThreadWait(Workers[i], NULL);   /* сперва дождаться всех производителей (API потоков главы 56) */
}
xrtMPMCQueueClose(&Queue);            /* потом закрыть */
DrainAndShutdown(&Queue);             /* потребители доизбирают и завершаются естественно */

Практические задания

Итоги