Глава 108 Потоковые тела: загрузка, скачивание и асинхронные файлы

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

Тело HTTP — поле боя «содержимого» и «памяти»: загрузка 2 GB, потоковое скачивание, ответ, генерируемый по надобности, — размер тела не должен равняться занятой памяти — отправная точка всего дизайна. xhttpbody объединяет пять источников тела: фиксированное семейство (Empty/Copy/Borrow/Take/Reference — переигрываемо, известная длина), ограниченный поток-производитель (HttpBodyStreamCreate — производство в потоке задачи, бюджетный потолок, фундамент SSE), файловые тела (http_body_file — асинхронный файловый IO в форме тела), собственные источники (обратный вызов производителя) и трансформированные тела. На стороне потребления xhttpbodyreader — однопотребительское потоковое чтение, а xhttpbodychunk — независимый лизинг данных (ChunkRelease возвращает, можно освободить позже Reader — идеальный напарник отправки по ссылке TCP). Серверные обратные вызовы Body главы 104 и конец-производитель SSE главы 107 получают здесь полный фундаментальный рассказ.

108.1引入

Три сценария тела с требованиями к форме памяти. Загрузка 2 GB: сервер не может накопить всё и потом обрабатывать — потоковые обратные вызовы Body главы 104 уже закрыли принимающую сторону; но отправляющая сторона клиента? читать 2 GB в память и потом POST? Файловые тела дают «читаем и шлём по ходу». Потоковая генерация скачивания: длина ответа неизвестна, содержимое производится по надобности (курсор базы, живой транскодинг) — тело потока-производителя даёт «генерируем и шлём по ходу» под потолком бюджета. Потребность переигрывания: перенаправления 307/308 главы 100 требуют переигрываемого тела — фиксированное семейство переигрываемо по природе; а поток-производитель? Тело, которое даёт HttpBodyStreamCreate, потребляется один раз — сценариям повторной отправки при перенаправлении подходят фиксированное семейство или файловые тела.

Лизинг — важнейшее новое понятие главы: Chunk, выдаваемый Next, владеет независимым лизингом данных — данные не копируются, но вы обязаны вернуть ChunkRelease; Chunk может пережить Reader (стоять в очереди отправки по ссылке, когда объект тела уже уничтожен — управляет счётчиком ссылок). Это та же мысль, что отправка по ссылке WebSocket главы 96 (освобождение только по завершении записи), приземлившаяся на слой тела.

108.2概念

Пять источников и семантика переигрывания

Детали фиксированного семейства: Copy кладёт дескриптор и копию в одно компактное выделение по фактической длине (второго выделения под то же тело не строится); Borrow не продлевает жизненный цикл внешней памяти (исходный буфер обязан жить во время отправки); Reference обязан получить процедуру освобождения (если только заимствуете — берите Borrow). xrtHttpBodyView заимствует непрерывные байты фиксированного тела — источники, не гарантирующие непрерывность, возвращают false (файл/поток/трансформация могут быть разрывными).

Reader: один потребитель и лизинги

Дисциплина алиасинга (на уровне контракта): дескриптор/буфер/слоты длины выводимого Chunk не должны перекрывать нижележащие байты Reader, Body или фиксированного тела — ошибка алиасинга не двигает Reader вперёд и не очищает выходную длину, а фиксированное Body остаётся неизменным при переигрывании и конкурентных открытиях. Эта защита делает безопасным «объект тела, разделяемый несколькими Call» (ссылки тела с семантикой заморозки главы 99).

Ограниченный поток-производитель: бюджет и финалы

MaxBytes/MaxChunks из xhttpbodystreamconfig — бюджет конца-производителя (резервирования + очередь + активные лизинги покрыты целиком) — порог AGAIN для SSE приходит именно отсюда. Три финала: разрушение последней ссылки потока = очередь содержимого затем нормальный EOF; Close идемпотентно предзакрывает ввод; Fail отбрасывает недоставленное содержимое и передаёт стабильную Cause потребляющей стороне. Три формы записи Write/WriteRef/WriteTake соответствуют копия/ссылка/принятие — то же семейство, что пять ступеней отправки главы 68.

Файловые тела: асинхронный IO в форме тела

Семейство http_body_file оборачивает файловый IO портов завершения главы 72 в тело: xrtHttpBodyFileConfigInit + создание файлового тела — чтение идёт асинхронным файловым путём (двигают Worker среды выполнения), без блокировок и без чтения целиком; xrtHttpBodyFileFuture/FileRangeFuture дают форму Future «ждать готовности тела». Range-варианты естественно поддерживают докачку загрузки/скачивания. Файловые тела переигрываемы (переоткрыть файл, перечитать) — безопасная комбинация больших файлов + переотправка при перенаправлении.

Полные формы загрузки и скачивания

• Загрузка (клиент): файловое тело + SetBody построителя главы 99 — память клиента при загрузке файла 2 GB ≈ один блок чтения; сервер принимает потоковыми обратными вызовами Body главы 104.
• Скачивание (клиент): ResponseBodyLimit главы 99 ведает бюджетом, потоковое потребление (обратные вызовы Body или Reader) — память развязана с объёмом скачивания; декодеры главы 93 цепляются на путь потребления.
• Генерируемые ответы (сервер): тело потока-производителя + Respond — курсор базы выдаёт шаг за шагом; при упоре в бюджет AGAIN оставляет производителя ждать потребителя. При запуске всего этого поверх TLS-сервиса (глава 86) разбор заголовков запроса идёт через TLS-привязку парсера главы 90 (http1_tls) — без буфера уровня соединения между записями; дисциплина потоковости проходит насквозь.

108.3示例

Первая полная программа: выписывание потока-производителя и обратное чтение

Программа ниже из examples/http/body_stream — минимальный замкнутый цикл ограниченного потока-производителя:

#include <stdio.h>
#include <xhttp.h>



/* 读取完整异步正文；AGAIN 只通过公开 Future 契约继续。 */
static bool exampleReadBody(xhttpbody* pBody)
{
	xhttpbodyreader* pReader = xrtHttpBodyOpen(pBody);
	xhttpbodychunk Chunk;
	xhttpbodystatus Status;

	if ( pReader == NULL ) {
		return false;
	}
	for ( ;; ) {
		Status = xrtHttpBodyNext(pReader, 4, &Chunk);
		if ( Status == XHTTP_BODY_DATA ) {
			(void)fwrite(Chunk.Data, 1, Chunk.Size, stdout);
			xrtHttpBodyChunkRelease(&Chunk);
			continue;
		}
		if ( Status == XHTTP_BODY_AGAIN ) {
			xfuture* pReady = xrtHttpBodyReaderWait(pReader);

			if ( (pReady == NULL) ||
				(xrtFutureWait(pReady) != XWAIT_OK) ||
				(xrtFutureState(pReady) != XFUTURE_RESOLVED) ) {
				xrtFutureDestroy(pReady);
				xrtHttpBodyReaderDestroy(pReader);
				return false;
			}
			xrtFutureDestroy(pReady);
			continue;
		}
		break;
	}
	xrtHttpBodyReaderDestroy(pReader);
	return Status == XHTTP_BODY_EOF;
}



/* 创建有界生产流，写入两段数据并由最后一个生产端发布 EOF。 */
int main(void)
{
	xhttpbodystreamconfig Config;
	xhttpbodystream* pStream = NULL;
	xhttpbody* pBody;
	bool bResult;

	xrtHttpBodyStreamConfigInit(&Config);
	Config.MaxBytes = 1024;
	Config.MaxChunks = 16;
	pBody = xrtHttpBodyStreamCreate(&Config, &pStream);
	if ( (pBody == NULL) ||
		(xrtHttpBodyStreamWrite(
			pStream, XRT_BYTES_LITERAL("first chunk\n")
		) != XHTTP_BODY_STREAM_OK) ||
		(xrtHttpBodyStreamWrite(
			pStream, XRT_BYTES_LITERAL("second chunk\n")
		) != XHTTP_BODY_STREAM_OK) ) {
		xrtHttpBodyStreamDestroy(pStream);
		xrtHttpBodyDestroy(pBody);
		return 1;
	}
	xrtHttpBodyStreamDestroy(pStream);
	bResult = exampleReadBody(pBody);
	xrtHttpBodyDestroy(pBody);
	return bResult ? 0 : 1;
}

Что только что произошло. (1) Config.MaxBytes=1024/MaxChunks=16 ставит бюджет, затем HttpBodyStreamCreate одним вызовом даёт пару объектов: xhttpbody (отдаётся запросу/ответу) и xhttpbodystream (держит конец-производитель). (2) Два отрезка Write, затем поток уничтожен — разрушение последней ссылки = нормальный EOF: поставленное в очередь содержимое доставляется сперва, EOF следом. (3) Цикл обратного чтения (тот же каркас, что у клиента главы 107): Next(8, &Chunk) режет малым потолком, DATA печатает + ChunkRelease, AGAIN ждёт на Future xrtHttpBodyReaderWait, EOF завершает. (4) Две строки вывода означают «поток байтов, записанный производителем, добирается до потребителя целиком» — объект тела есть точка развязки производства и потребления: концы могут жить в разных потоках в разные моменты. В реальном развёртывании конец-производитель — поток задачи (события SSE/выдача курсора), потребитель — транспорт HTTP.

Вторая полная программа: загрузка файлового тела

Вторая программа из examples/http/body_file — форма загрузки больших файлов:

#include <stdio.h>

#include <xhttp.h>



/* 等待 Future 成功并返回借用值。 */
static ptr exampleFutureValue(xfuture* pFuture)
{
	if ( (pFuture == NULL) ||
		(xrtFutureWait(pFuture) != XWAIT_OK) ||
		(xrtFutureState(pFuture) != XFUTURE_RESOLVED) ) {
		return NULL;
	}
	return xrtFutureValue(pFuture);
}



/* 使用统一正文 Reader 异步读取文件内容。 */
int main(void)
{
	static const char sPath[] = "xrt-http-body-file-example.tmp";
	xtaskpoolconfig Config = { 1, 16, 0 };
	xtaskpool* pPool;
	xfile File;
	xfuture* pPrepare;
	xhttpbody* pBody;
	xhttpbodyreader* pReader;
	xhttpbodychunk Chunk;
	xhttpbodystatus Status;

	(void)xrtFileDelete(sPath);
	xrtClearError();
	File = xrtOpen(
		sPath,
		XFILE_WRITE | XFILE_CREATE | XFILE_EXCLUSIVE
	);
	if ( (File == NULL) ||
		!xrtWriteFull(File, "HTTP file body\n", 15, NULL) ||
		!xrtClose(File) ) {
		return 1;
	}
	pPool = xrtTaskPoolCreate(&Config);
	if ( pPool == NULL ) {
		(void)xrtFileDelete(sPath);
		return 1;
	}
	/* NULL 配置按需读取，默认单次申请不超过 64 KiB。 */
	pPrepare = xrtHttpBodyFileFuture(pPool, sPath, NULL);
	pBody = (xhttpbody*)exampleFutureValue(pPrepare);
	pReader = pBody != NULL ? xrtHttpBodyOpen(pBody) : NULL;
	while ( pReader != NULL ) {
		Status = xrtHttpBodyNext(pReader, 4096, &Chunk);
		if ( Status == XHTTP_BODY_AGAIN ) {
			xfuture* pReady = xrtHttpBodyReaderWait(pReader);

			if ( exampleFutureValue(pReady) == NULL ) {
				xrtFutureDestroy(pReady);
				break;
			}
			xrtFutureDestroy(pReady);
			continue;
		}
		if ( Status == XHTTP_BODY_EOF ) {
			break;
		}
		if ( Status == XHTTP_BODY_ERROR ) {
			break;
		}
		(void)fwrite(Chunk.Data, 1, Chunk.Size, stdout);
		xrtHttpBodyChunkRelease(&Chunk);
	}
	xrtHttpBodyReaderDestroy(pReader);
	xrtFutureDestroy(pPrepare);
	(void)xrtTaskPoolDestroy(pPool);
	(void)xrtFileDelete(sPath);
	return 0;
}

Что только что произошло. (1) xrtHttpBodyFileConfigInit настраивает путь и опции, затем создаётся файловое тело — длина из метаданных файла, содержимое читается асинхронно по надобности: память клиента не зависит от размера файла (сравните полное чтение Copy — 2 GB есть 2 GB). (2) Файловое тело переигрываемо (переоткрыть и перечитать) — требование переотправки перенаправлений 307/308 главы 100 выполняется здесь по природе; фиксированный поток-производитель — нет (одно потребление). (3) Форма Future xrtHttpBodyFileFuture вводит «ждать готовности файла/завершения чтения» в систему ожидания главы 58. (4) Сопроводители: examples/http/body (пять источников фиксированного семейства), body_compose (составные тела), body_inflate/deflate (трансформированные тела — сжатие отправляющей стороны, парное декодированию главы 93), server_body_async (серверные асинхронные телесные ответы) — полная инспекционная грань семейства тел.

108.4契约

• Пять источников: фиксированное семейство (переигрываемо, известная длина), поток-производитель (одно потребление, ограничен), файлы (переигрываемы, асинхронное чтение), собственные, трансформированные — единая форма xhttpbody.
• Дисциплина фиксированного семейства: Copy одно выделение (дескриптор + копия одним компактным блоком); Borrow не продлевает внешнюю жизнь; Reference обязан иметь процедуру освобождения; View успешен только для источников с гарантией непрерывности.
• Поток-производитель: бюджет MaxBytes/MaxChunks покрывает резервирования + очередь + лизинги; три финала EOF/Close/Fail; три формы Write (копия/ссылка/принятие).
• Reader: один потребитель; DATA (лизинг)/AGAIN (Wait Future)/EOF — три состояния; Chunk — независимый лизинг (ChunkRelease; освобождаем позже Reader/Body — дружественно отправке по ссылке).
• Защита от алиасинга: вывод не перекрывает нижележащие байты тела; ошибка алиасинга ничего не двигает и не ломает — фиксированные Body неизменны при переигрывании и конкурентных открытиях.
• Файловые тела: асинхронное чтение без блокировок и чтения целиком; переигрываемы; Range-варианты; форма готовности Future.
• Матрица переигрывания: переотправка 307/308 → фиксированное семейство/файлы; генерация один раз → поток-производитель.
• Принадлежность бюджета: клиентский ResponseBodyLimit (глава 99) ограничивает тело представления; бюджет потока-производителя ограничивает производителя — два слоя, каждый ограничен по-своему.
• Обрезка: HTTP_BODY/HTTP_BODY_STREAM/HTTP_BODY_FILE независимые макросы.

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

uint8 Buf[256];
fill(Buf);
xrtHttpRequestSetBody(pReq, xrtHttpBodyBorrow(view(Buf)));  /* стековый буфер */
send_async(pReq);   /* кадр стека умирает по возврату: отправка читает мусор */

/* либо Copy (малые тела), либо Reference + обратный вызов освобождения (свой кучный буфер) */
xrtHttpRequestSetBody(pReq, xrtHttpBodyCopy(view(Buf)));
/* или xrtHttpBodyReference(view(HeapBuf), release_cb, ctx) - освобождение автоматически по завершении отправки */

for ( each_item ) {
	if ( send(stream, item) == AGAIN ) {
		continue;   /* выброшено - у клиента пробел в этом отрезке */
	}
}

for ( each_item ) {
	while ( send(stream, item) == AGAIN ) {
		wait_writable(stream);   /* шаблон ожидания из главы 107, ловушка 2 */
	}
}

while ( next(&Reader, &Chunk) == DATA ) {
	queue_push(g_Q, Chunk);   /* лизинг не возвращён: бюджет исчерпан */
}

while ( next(&Reader, &Chunk) == DATA ) {
	if ( want_keep(Chunk) ) {
		queue_push(g_Q, copy_chunk(Chunk));  /* храним копию */
	}
	xrtHttpBodyChunkRelease(&Chunk);          /* вернуть лизинг */
}

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

Итоги