Глава 48 Асинхронный файловый ввод-вывод

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

Модуль file-async поднимает файловый IO от блокирующей модели к модели «подача-завершение»: xrtAsyncFileReadAt/WriteAt подают операцию и немедленно возвращают Future — IO исполняется на привязанном потоке пула задач (параллельная редакция пулов главы 22), а вызывающий поток с Future идёт по своим делам; xrtFutureWait ждёт завершения, xrtFutureValue забирает результат (результат чтения — xfiledata: представление данных + размер + смещение + отметка EOF). Слой удобства для целых файлов xrtFileReadAllAsync/WriteAllAsync закрывает дело одним шагом. Глава разбирает семантику модели, дисциплины Future (железные законы асинхронной границы) и выбор между синхронностью и асинхронностью.

48.1引入

Службе архивации журналов нужно таскать большие файлы: синхронно Read кусок — обработать кусок — записать кусок — однопоточная сериализация, и всё время ожидания диска записывается на счёт обрабатывающего потока. Хотите параллельно таскать три файла? В синхронной модели это три потока, каждый со своей блокировкой (потоки дорогие — бухгалтерия главы 55). Асинхронная модель меняет рамку: подали запрос «прочитай этот кусок» — и поток свободен — система завершает IO в фоне и уведомляет вас; один поток может держать в полёте десятки незавершённых IO. Параллельный потенциал диска (NCQ/многоканальность SSD) и освобождение потоков случаются одновременно.

Цена — смена ментальной модели: от «вернулся из вызова — значит завершён» к «вернулся из вызова — значит принято»: момент завершения, поток обратного вызова, пути отказа — всё меняет форму. Задача главы — расписать дисциплины этой модели; она же — файловая репетиция сетевого движка главы 63 (событийного): две главы делят одно мышление «подача-завершение».

48.2概念

Модель «подача-завершение»

Ключевое понимание: получить Future не значит завершиться — между приёмом и завершением вызывающий поток свободен, но данные не готовы; всякое использование результата обязано сперва пройти Wait (или композицию продолжений тома 6). Пул задач — созданный вами (xrtTaskPoolCreate({线程数, 队列深度, 标志}) — число потоков, глубина очереди, флаги): противодавление ограниченной очереди срабатывает в точке «пул полон — подача отказала», а не в «разрослось до OOM» (файловая версия клапана противодавления главы 21).

Дисциплины Future: три закона асинхронной границы

Закон первый: значение заимствовано — после завершения FutureValue возвращает xfiledata как заимствованное представление, срок действия до следующей операции; чтобы держать долго — скопируйте сами (дисциплина заимствования главы 3).Закон второй: заморозка буфера — между подачей и завершением буфер операции неприкосновенен (поток IO читает и пишет его); нужно переиспользовать заранее — берите пул буферов, по буферу на IO.Закон третий: отказ — тоже завершение — состояние отказа и состояние успеха у Future проходят одни и те же интерфейсы ожидания/взятия; короткое чтение помечает EOF битом End (булево поле xfiledata), отказ несёт детали ошибки — код ожидания, написанный только под успех, при заполнении диска зависнет или солжёт.

Две позы ожидания (выбор есть архитектурное решение)

Блокирующее ожидание — не регресс: пакетный режим «подать много + ждать всех вместе» всё ещё быстрее чистой синхронности (IO исполнялись на пуле параллельно), сериализуется только обработка завершений — прагматичный выбор файловой пакетной обработки. Композиция Future — официальное рукопожатие асинхронности с системой параллелизма: глава 59 развернёт продолжения, глава 60 — параллельные слияния групп задач. Слой удобства целых файлов (образец async_whole): xrtFileReadAllAsync/WriteAllAsync — целый файл одним шагом; есть вариант с жёстким потолком (ReadAllLimitAsync — сверх лимита отказ; часовой границы ресурсов на асинхронном слое).

Выбор против синхронного слоя (не весь IO должен быть асинхронным)

Сперва холодный душ: не всякому файловому IO нужна асинхронность.Чтение-запись маленьких конфигураций — синхронные три строки (глава 45) проще, сложность асинхронности оплачена зря.Загрузка ресурсов на старте — последовательная семантика яснее, выигрыш асинхронности миллисекундный.Перенос больших файлов, параллельная пакетная обработка, IO-плотные службы — арена асинхронности. Мнемоника: IO много, каждый велик, параллелится — асинхронно; разрозненные мелочи — синхронно. Смешивание тоже обычно (старт синхронно, рантайм асинхронно).

48.3示例

Полная программа: асинхронный цикл запись-чтение

Из образца репозитория examples/file/async/main.c:

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



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





















 */


/* 使用有界任务池完成异步绝对偏移读写。 */
int main(void)
{
	xtaskpoolconfig Config = { 2, 32, 0 };
	xfileoptions Options;
	xtaskpool* pPool = xrtTaskPoolCreate(&Config);
	xasyncfile* pFile = NULL;
	xfuture* pWrite = NULL;
	xfuture* pRead = NULL;
	xfuture* pClose = NULL;
	xfiledata* pData;
	int iResult = 1;

	xrtFileOptionsInit(&Options);
	Options.Flags = XFILE_READ | XFILE_WRITE |
		XFILE_CREATE | XFILE_TRUNCATE;
	if ( pPool == NULL ) {
		goto Exit;
	}
	pFile = xrtAsyncFileOpen(
		pPool,
		"xrt-async-example.txt",
		&Options
	);
	if ( pFile == NULL ) {
		goto Exit;
	}
	pWrite = xrtAsyncFileWriteAt(
		pFile,
		0,
		XRT_BYTES_LITERAL("hello async file")
	);
	if ( waitValue(pWrite) == NULL ) {
		goto Exit;
	}
	pRead = xrtAsyncFileReadAt(pFile, 0, 64);
	pData = (xfiledata*)waitValue(pRead);
	if ( pData == NULL ) {
		goto Exit;
	}
	printf(
		"%.*s\n",
		(int)pData->Size,
		(const char*)pData->Data
	);
	pClose = xrtAsyncFileClose(pFile);
	pFile = NULL;
	if ( (pClose == NULL) ||
		(xrtFutureWait(pClose) != XWAIT_OK) ||
		(xrtFutureState(pClose) != XFUTURE_RESOLVED) ) {
		goto Exit;
	}
	iResult = 0;

Exit:
	if ( pFile != NULL ) {
		xfuture* pPending = xrtAsyncFileClose(pFile);

		if ( pPending != NULL ) {
			(void)xrtFutureWait(pPending);
			xrtFutureDestroy(pPending);
		}
	}
	xrtFutureDestroy(pWrite);
	xrtFutureDestroy(pRead);
	xrtFutureDestroy(pClose);
	if ( pPool != NULL ) {
		(void)xrtTaskPoolDestroy(pPool);
	}
	return iResult;
}

Что только что произошло. (1) xtaskpoolconfig Config = { 2, 32, 0 } — ограниченный пул задач: два потока, глубина очереди 32; IO исполняется на потоках пула, полный пул отказывает в подаче (точка приземления противодавления на файловом слое). (2) xrtAsyncFileOpen (пул, путь, опции) привязывает файл к пулу — дальнейшие чтение и запись идут через очередь пула. (3) xrtAsyncFileWriteAt и xrtAsyncFileReadAt возвращают по Future — помощник waitValue образца (Wait + проверка State + взятие Value) и есть полная форма позы блокирующего ожидания; результат чтения несёт представление данных и размер через xfiledata. (4) Порядок чтения и записи гарантируется цепочкой Future — сперва дождаться завершения записи, потом подавать чтение; зависимости цепочки строятся ожиданием явно (без ожидания — параллельно, порядка нет). (5) Close тоже асинхронен и возвращает Future — закрытие тоже нужно дождаться, только тогда ресурсы дескриптора действительно вернутся; вывод hello async file и есть проверка замкнутого цикла записи-чтения.

Полная программа: целый файл асинхронно

Из examples/file/async_whole/main.c — слой удобства целого чтения-записи:

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


















 */


/* 使用显式任务池完成原子写入和整文件读取。 */
int main(void)
{
	xtaskpoolconfig Config = { 2, 16, 0 };
	xtaskpool* pPool = xrtTaskPoolCreate(&Config);
	xfuture* pWrite = NULL;
	xfuture* pRead = NULL;
	xfiledata* pData;
	int iResult = 1;
	static const char sPath[] = "xrt-async-whole-example.txt";

	if ( pPool == NULL ) {
		goto Exit;
	}
	pWrite = xrtFileWriteAtomicAsync(
		pPool,
		sPath,
		XRT_BYTES_LITERAL("hello async whole file")
	);
	if ( (pWrite == NULL) ||
		(xrtFutureWait(pWrite) != XWAIT_OK) ||
		(xrtFutureState(pWrite) != XFUTURE_RESOLVED) ) {
		goto Exit;
	}
	pRead = xrtFileReadAllAsync(pPool, sPath);
	if ( (pRead == NULL) ||
		(xrtFutureWait(pRead) != XWAIT_OK) ||
		(xrtFutureState(pRead) != XFUTURE_RESOLVED) ) {
		goto Exit;
	}
	pData = (xfiledata*)xrtFutureValue(pRead);
	if ( pData == NULL ) {
		goto Exit;
	}
	printf("%.*s\n", (int)pData->Size, (const char*)pData->Data);
	iResult = 0;

Exit:
	xrtFutureDestroy(pWrite);
	xrtFutureDestroy(pRead);
	if ( pPool != NULL ) {
		(void)xrtTaskPoolDestroy(pPool);
	}
	return iResult;
}

Что только что произошло. (1) Слой удобства упаковывает «подать кусками — собрать завершения» в один вызов — асинхронное чтение-запись целого файла без раскрытия кусков. (2) Он закрывает сценарий «нужно всё содержимое целиком» (загрузка конфигурации, считывание ресурсов) — кусками и сборкой ведает библиотека. (3) Разделение с синхронным целым чтением главы 45: маленький файл — синхронно в три строки; большой или при потребности в параллелизме — сюда; «удобство» не меняет модель, только срезает заготовки.

48.4契约

• Приём и есть возврат: возвращение функции подачи означает только приём; состояние завершения доставляется обратным вызовом или объектом ожидания.
• Железный закон потока обратных вызовов: обратные вызовы исполняются в системных IO-потоках — лёгкая работа, передача в поток, полные ветви отказа.
• Пакетное ожидание: много подач + одно общее ожидание — легальная форма: выигрыш параллельного IO сохранён, обработка завершений сериализована.
• Вход через Future: асинхронные операции можно обернуть в Future и ввести в систему параллелизма (глава 59) — стандартный канал композиции и продолжений.
• Мнемоника выбора: много, велико, параллелится → асинхронно; разрозненные мелочи → синхронно; смешивание — по фазам.

От примеров к инженерии: три хозяина асинхронных файлов

Перенос больших файлов (арена конвейера): завершение Future чтения куска A запускает подачу записи A и чтения B — качание двойного буфера, на всём пути ноль простоя в ожидании; скелет этого конвейера и есть продвинутое упражнение.Пакетная архивация (арена параллельности): N файлов, каждый подаёт чтение целиком → Futures собраны — сжимаем по одному → асинхронная запись назад; глубина пула естественно ограничивает параллелизм (противодавление встроено) — сжатие главы 29 + пулы главы 22 сливаются на асинхронном слое.Предзагрузка на старте службы (арена слоя удобства): ReadAllAsync параллельно предзагружает несколько ресурсных файлов, главный поток продолжает инициализацию, при надобности Wait — перекрытие времени старта. Общая точка трёх хозяев: степень параллелизма управляется глубиной пула — асинхронность это не бесконечный параллелизм, а «контролируемая параллельность».

Стык с системой задач главы 60

Future и пул задач этой главы — файловый вход в систему задач главы 60: xrtTaskSubmit подаёт произвольную функцию в тот же пул — файловый IO и вычислительные задачи делят пул и единое противодавление; комбинаторы Future (глава 59) выражают «слить, когда завершатся и чтение A, и чтение B» декларативной оркестровкой; группы задач (глава 60) управляют «жизненным циклом и отменой десяти архивных задач» как единым целым. Здесь вы учите слой примитивов — пул, дескриптор, Future; том 6 соберёт их в полные параллельные рабочие процессы. С этим взглядом на будущее: у каждой ловушки этой главы в томе 6 есть соответствующее решение слоем-обёрткой.

Ментальный якорь: принято ≠ завершено

Самая важная фраза главы заслуживает собственного раздела: приём не равно завершение. Интуиция синхронного кода (вернулся из вызова = результат доступен) в асинхронном мире не работает — Future это «обещание результата», а не результат. Три конкретных напоминания: читать данные немедленно после получения Future — прямая причина ловушки 2; Close возвращает Future — значит «закрытие тоже ждать»: ресурсы дескриптора вернутся по-настоящему только после завершения Future у Close; «дождаться всех» после пачки подач — это ждать все, а не какую-нибудь одну. Приклейте эту фразу на монитор: она предупреждает восемьдесят процентов ям асинхронного кода.

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

static void onDone(ptr pValue)
{
	ProcessEntireFile(pValue);   /* тяжёлый счёт в обратном вызове Future - занимает поток-планировщик */
}

static void onDone(ptr pValue)
{
	xrtTaskSubmit(gPool, DoHeavyWork, pValue, NULL);   /* лёгкая работа: переподать в пул задач */
}
/* тяжёлое исполняется задачей пула - обратный вызов Future только пересылает (употребление системы задач, глава 60) */

xrtAsyncFileWriteAt(File, Offset, Buffer, Size, NULL);
memset(Buffer, 0, Size);   /* чистка буфера сразу после подачи - поток пула ещё читает его */

xrtAsyncFileWriteAt(File, Offset, Buffer, Size, NULL);
/* буфер заморожен от подачи до завершения Wait - трогать только после */
/* нужно переиспользовать раньше: пул буферов (глава 22), по буферу на IO */

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

Итоги