Въведение
(единичен поток ) — това е шаблон за писане на приложения с висока натовареност, използван в много популярни решения:
- …
В тази статия ще разгледаме основите на I/O реактора и принципа на неговата работа, ще напишем реализация с по-малко от 200 реда код и ще накараме прост HTTP сървър да обработва над 40 милиона заявки/мин.
Предговор
- Статията е написана с цел да помогне да се разбере функционирането на I/O реактора, а следователно и да се осъзнаят рисковете при неговото използване.
- За усвояването на статията е необходимо да имате основи и опит в разработването на мрежови приложения.
- Целият код е написан на езика C стриктно според (внимание: дълъг PDF) за Linux и е достъпен на .
Защо е необходимо?
С нарастващата популярност на интернет, уеб сървърите започнаха да се нуждаят от обработка на голям брой съединения едновременно, в резултат на което бяха изпробвани два подхода: блокиращо I/O с голям брой потоци на ОС и неблокиращо I/O в комбинация със система за уведомяване за събития, известна още като «системен селектор» (///etc).
Първият подход предвиждаше създаване на нов поток на ОС за всяко входящо съединение. Неговият недостатък е лошата мащабируемост: операционната система трябва да извърши множество и . Те са скъпи операции и могат да доведат до недостиг на свободна RAM при значителен брой съединения.
Модифицираната версия определя (пул от потоци), като по този начин не позволява на системата да прекрати изпълнението аварийно, но в същото време внася нова проблематика: ако в момента пулът от потоци блокира продължителни операции на четене, то другите сокети, които вече са в състояние да приемат данни, не могат да го направят.
Вторият подход използва (системен селектор), предоставена от ОС. В тази статия ще разгледаме най-често срещаната форма на системен селектор, основана на уведомления (събития, уведомления) за готовност за I/O операции, вместо на . Опростен пример за неговото използване може да бъде представен в следната блок-схема:
Разликата между тези подходи е следната:
- Блокиращите I/O операции спират потребителския поток докато AWS не направи същото и не изгради успешен бизнес около него, и когато клиентите буквално поискат същото. Все пак трябва да се положат известни усилия, за да накарате Google да поддържа нещо., докато ОС не постъпващите в поток от байтове (, получаване на данни) или не остава достатъчно място във вътрешните буфери за запис за последваща изпращане чрез (изпращане на данни).
- Системният селектор след известно време уведомява програмата, че ОС вече е дефрагментирала IP пакетите (TCP, получаване на данни) или достатъчно място във вътрешните буфери за запис вече е налично (изпращане на данни).
В заключение, резервацията на потока на ОС за всяко I/O е напразно изразходване на изчислителна мощ, тъй като на практика потоците не са заети с полезна работа (тук произлиза терминът ). Системният селектор решава този проблем, позволявайки на потребителската програма да използва ресурсите на CPU много по-икономично.
Моделът на I/O реактора
I/O реакторът действа като слой между системния селектор и потребителския код. Принципът на работа е описан в следната блок-схема:
- Да напомня, че събитието е уведомление, че определен сокет е в състояние да изпълни неблокираща I/O операция.
- Обработчикът на събития е функция, извиквана от I/O реактора при получаване на събитие, която след това извършва неблокираща I/O операция.
Важно е да се отбележи, че I/O реакторът по дефиниция е еднопотоков, но нищо не пречи концепцията да се използва в многопоточна среда в съотношение 1 поток: 1 реактор, като по този начин се използват всички ядра на CPU.
Реализация
Публичният интерфейс ще бъде поставен в файл , а реализацията — в . reactor.h ще се състои от следните декларации:
Показване на декларации в reactor.h
typedef struct reactor Reactor;
/*
* Показател на функция, която ще бъде извикана от I/O реактора при получаване
* на събитие от системния селектор.
*/
typedef void (*Callback)(void *arg, int fd, uint32_t events);
/*
* Връща `NULL` в случай на грешка, не-`NULL` указател на `Reactor` в
* противен случай.
*/
Reactor *reactor_new(void);
/*
* Освобождава системния селектор, всички регистрирани сокети в момента
* и самия I/O реактор.
*
* Следните функции връщат -1 в случай на грешка, 0 в случай на успех.
*/
int reactor_destroy(Reactor *reactor);
int reactor_register(const Reactor *reactor, int fd, uint32_t interest,
Callback callback, void *callback_arg);
int reactor_deregister(const Reactor *reactor, int fd);
int reactor_reregister(const Reactor *reactor, int fd, uint32_t interest,
Callback callback, void *callback_arg);
/*
* Стартира цикъл от събития с таймаут `timeout`.
*
* Тази функция ще предаде управлението на извикващия код, ако отреденото време е изтекло
* или/и при липса на регистрирани сокети.
*/
int reactor_run(const Reactor *reactor, time_t timeout);Структурата на I/O реактора се състои от селектор и , която съпоставя всеки сокет с CallbackData (структура от обработчик на събитие и аргумент за потребителя към него).
Покажи Reactor и CallbackData
struct reactor {
int epoll_fd;
GHashTable *table; // (int, CallbackData)
};
typedef struct {
Callback callback;
void *arg;
} CallbackData;Обърнете внимание, че сме използвали възможността за работа с по указател. В reactor.h обявяваме структурата reactor, а в reactor.c я дефинираме, така че да не позволим на потребителя явно да променя полетата ѝ. Това е един от моделите , лаконично вписващ се в семантиката на Си.
Функции reactor_register, reactor_deregister и reactor_reregister обновяват списъка на интересуващите сокети и съответстващите обработчици на събития в системния селектор и в хеш-таблицата.
Покажи функции за регистрация
#define REACTOR_CTL(reactor, op, fd, interest)
if (epoll_ctl(reactor->epoll_fd, op, fd,
&(struct epoll_event){.events = interest,
.data = {.fd = fd}}) == -1) {
perror("epoll_ctl");
return -1;
}
int reactor_register(const Reactor *reactor, int fd, uint32_t interest,
Callback callback, void *callback_arg) {
REACTOR_CTL(reactor, EPOLL_CTL_ADD, fd, interest)
g_hash_table_insert(reactor->table, int_in_heap(fd),
callback_data_new(callback, callback_arg));
return 0;
}
int reactor_deregister(const Reactor *reactor, int fd) {
REACTOR_CTL(reactor, EPOLL_CTL_DEL, fd, 0)
g_hash_table_remove(reactor->table, &fd);
return 0;
}
int reactor_reregister(const Reactor *reactor, int fd, uint32_t interest,
Callback callback, void *callback_arg) {
REACTOR_CTL(reactor, EPOLL_CTL_MOD, fd, interest)
g_hash_table_insert(reactor->table, int_in_heap(fd),
callback_data_new(callback, callback_arg));
return 0;
}След като I/O реакторът е уловил събитие с дескриптор fd, той извиква съответния обработчик на събития, в който предава fd, на генерираните събития и потребителски указател на празен.
Покажи функцията reactor_run()
int reactor_run(const Reactor *reactor, time_t timeout) {
int result;
struct epoll_event *events;
if ((events = calloc(MAX_EVENTS, sizeof(*events))) == NULL)
abort();
time_t start = time(NULL);
while (true) {
time_t passed = time(NULL) - start;
int nfds =
epoll_wait(reactor->epoll_fd, events, MAX_EVENTS, timeout - passed);
switch (nfds) {
// Грешка
case -1:
perror("epoll_wait");
result = -1;
goto cleanup;
// Времето изтече
case 0:
result = 0;
goto cleanup;
// Успешна операция
default:
// Извикване на обработчиците на събития
for (int i = 0; i table, &fd);
callback->callback(callback->arg, fd, events[i].events);
}
}
}
cleanup:
free(events);
return result;
}В обобщение, веригата от извиквания на функции в потребителския код ще изглежда по следния начин:
Однопоточен сървър
За да тестваме I/O реактора под високо натоварване, ще напишем прост HTTP уеб сървър, който отговаря с изображение на всяка заявка.
Кратка информация за протоколa HTTP
— това е протокол , който се използва основно за взаимодействие между сървъра и браузъра.
HTTP може лесно да се използва над протокол , изпращайки и получавайки съобщения във формата, определен от .
Формат на заявка
CRLF
CRLF
CRLF
CRLF CRLFCRLF— това е последователност от два символа:rиn, разделяща първия ред на заявката, заглавията и данните.<КОМАНДА>— една отCONNECT,DELETE,GET,HEAD,OPTIONS,PATCH,POST,PUT,TRACE. Браузерът ще изпрати на нашия сървър команда,GET, която означава "Изпрати ми съдържанието на файла".<URI>— . Например, ако URI =/index.html, клиентът иска началната страница на сайта.<ВЕРСИЯ HTTP>— версия на протокола HTTP в форматаHTTP/X.Y. Най-често използваната версия в днешно време еHTTP/1.1.<ЗАГОЛОВОК N>— това е двойка ключ-стойност в формата:, изпращана на сървъра за по-нататъшен анализ.<ДАННЫЕ>— данни, необходими на сървъра за изпълнение на операцията. Често това е просто или друг формат.
Формат на отговор
CRLF
CRLF
CRLF
CRLF CRLF<КОД СТАТУСА>— това число представлява резултата от операцията. Нашият сървър винаги ще връща статус 200 (успешна операция).<ОПИСАНИЕ СТАТУСА>— низово представяне на статус кода. За статус кода 200 — това еOK.<ЗАГОЛОВОК N>— заглавие в същия формат като в заявката. Ще връщаме заглавияContent-Length(размер на файла) иContent-Type: text/html(тип връщани данни).<ДАННЫЕ>— данните, изисквани от потребителя. В нашия случай това е пътят към изображението в .
Файл (еднопоточен сървър) включва файла , който съдържа следните прототипи на функции:
Покажи прототипи на функции в common.h
/*
* Обработчик событий, который вызовется после того, как сокет будет
* готов принять новое соединение.
*/
static void on_accept(void *arg, int fd, uint32_t events);
/*
* Обработчик событий, который вызовется после того, как сокет будет
* готов отправить HTTP ответ.
*/
static void on_send(void *arg, int fd, uint32_t events);
/*
* Обработчик событий, который вызовется после того, как сокет будет
* готов принять часть HTTP запроса.
*/
static void on_recv(void *arg, int fd, uint32_t events);
/*
* Переводит входящее соединение в неблокирующий режим.
*/
static void set_nonblocking(int fd);
/*
* Печатает переданные аргументы в stderr и выходит из процесса с
* кодом `EXIT_FAILURE`.
*/
static noreturn void fail(const char *format, ...);
/*
* Возвращает файловый дескриптор сокета, способного принимать новые
* TCP соединения.
*/
static int new_server(bool reuse_port);Също така е описан функционален макрос SAFE_CALL() и е определена функция fail(). Макросът сравнява стойността на израза с грешка, и ако условието е изпълнено, извиква функцията fail():
#define SAFE_CALL(call, error)
do {
if ((call) == error) {
fail("%s", #call);
}
} while (false)Функция fail() извежда предадените аргументи в терминала (като ) и завършва изпълнението на програмата с код EXIT_FAILURE:
static noreturn void fail(const char *format, ...) {
va_list args;
va_start(args, format);
vfprintf(stderr, format, args);
va_end(args);
fprintf(stderr, ": %sn", strerror(errno));
exit(EXIT_FAILURE);
}Функция new_server() върща файлов дескриптор на "сървърния" сокет, създаден от системните извиквания , и и способен да приема входящи връзки в неблокиращ режим.
Покажи функция new_server()
static int new_server(bool reuse_port) {
int fd;
SAFE_CALL((fd = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, IPPROTO_TCP)),
-1);
if (reuse_port) {
SAFE_CALL(
setsockopt(fd, SOL_SOCKET, SO_REUSEPORT, &(int){1}, sizeof(int)),
-1);
}
struct sockaddr_in addr = {.sin_family = AF_INET,
.sin_port = htons(SERVER_PORT),
.sin_addr = {.s_addr = inet_addr(SERVER_IPV4)},
.sin_zero = {0}};
SAFE_CALL(bind(fd, (struct sockaddr *)&addr, sizeof(addr)), -1);
SAFE_CALL(listen(fd, SERVER_BACKLOG), -1);
return fd;
}- Обърнете внимание, че сокетът първоначално се създава в неблокиращ режим с помощта на флага
SOCK_NONBLOCK, за да функциятаon_accept()не прекъсва изпълнението на потока.accept()reuse_port - Ако
равен на, то тази функция ще конфигурира сокета с опциятаtrueSO_REUSEPORT setsockopt() Обработчикът на събития
се извиква след като ОС генерира събитие on_accept() EPOLLIN , в този случай означаващо, че нова връзка може да бъде приета.приема нова връзка, премества я в неблокиращ режим и я регистрира с обработчика на събития on_accept() on_recv() в I/O реактора. Покажи функция on_accept()
Показване на функцията on_accept()
static void on_accept(void *arg, int fd, uint32_t events) {
int incoming_conn;
SAFE_CALL((incoming_conn = accept(fd, NULL, NULL)), -1);
set_nonblocking(incoming_conn);
SAFE_CALL(reactor_register(reactor, incoming_conn, EPOLLIN, on_recv,
request_buffer_new()),
-1);
}се извиква след като ОС генерира събитие в I/O реактора. EPOLLIN , в този случай означаващо, че нова връзка може да бъде приета., което означава, че регистрираната връзка on_accept(), е готова да приема данни.
в I/O реактора. чете данни от връзката, докато HTTP заявката е напълно получена, след което регистрира обработчик on_send() за изпращане на HTTP отговор. Ако клиентът прекъсне връзката, сокетът се дерегистрира и затваря чрез .
Покажи функцията on_recv()
static void on_recv(void *arg, int fd, uint32_t events) {
RequestBuffer *buffer = arg;
// Прийом на входящите данни, докато recv не върне 0 или грешка
ssize_t nread;
while ((nread = recv(fd, buffer->data + buffer->size,
REQUEST_BUFFER_CAPACITY - buffer->size, 0)) > 0)
buffer->size += nread;
// Клиентът прекъсна връзката
if (nread == 0) {
SAFE_CALL(reactor_deregister(reactor, fd), -1);
SAFE_CALL(close(fd), -1);
request_buffer_destroy(buffer);
return;
}
// read върна грешка, различна от грешка, при която извикването блокира
// потока
if (errno != EAGAIN && errno != EWOULDBLOCK) {
request_buffer_destroy(buffer);
fail("read");
}
// Получена е пълна HTTP заявка от клиента. Сега регистрираме обработчика
// на събитията за изпращане на данни
if (request_buffer_is_complete(buffer)) {
request_buffer_clear(buffer);
SAFE_CALL(reactor_reregister(reactor, fd, EPOLLOUT, on_send, buffer),
-1);
}
}се извиква след като ОС генерира събитие on_send() EPOLLIN EPOLLOUT, което означава, че регистрираната връзка в I/O реактора., е готова да изпраща данни. Тази функция изпраща HTTP отговор, съдържащ HTML с изображение, на клиента, а след това отново променя обработчика на събитията на в I/O реактора..
Покажи функцията on_send()
статичен void on_send(void *arg, int fd, uint32_t events) {
const char *content = "<img "
"src="https://habrastorage.org/webt/oh/wl/23/"
"ohwl23va3b-dioerobq_mbx4xaw.jpeg">";
char response[1024];
sprintf(response,
"HTTP/1.1 200 OK" CRLF "Content-Length: %zd" CRLF "Content-Type: "
"text/html" DOUBLE_CRLF "%s",
strlen(content), content);
SAFE_CALL(send(fd, response, strlen(response), 0), -1);
SAFE_CALL(reactor_reregister(reactor, fd, EPOLLIN, on_recv, arg), -1);
}И накрая, в файла http_server.c, във функцията main() създаваме I/O реактор с помощта на reactor_new(), създаваме сървърен сокет и го регистрираме, стартираме реактора с помощта на reactor_run() точно за една минута, след което освобождаваме ресурсите и излизаме от програмата.
Покажи http_server.c
#include "reactor.h"
static Reactor *reactor;
#include "common.h"
int main(void) {
SAFE_CALL((reactor = reactor_new()), NULL);
SAFE_CALL(
reactor_register(reactor, new_server(false), EPOLLIN, on_accept, NULL),
-1);
SAFE_CALL(reactor_run(reactor, SERVER_TIMEOUT_MILLIS), -1);
SAFE_CALL(reactor_destroy(reactor), -1);
}Нека проверим, че всичко работи както трябва. Компилираме (chmod a+x compile.sh && ./compile.sh в корена на проекта) и стартираме самостоятелния сървър, отваряме в браузъра и наблюдаваме това, което очаквахме:
Тест за производителност
Покажи характеристиките на моята машина
$ screenfetch
MMMMMMMMMMMMMMMMMMMMMMMMMmds+. OS: Mint 19.1 tessa
MMm----::-://///////////oymNMd+` Kernel: x86_64 Linux 4.15.0-20-generic
MMd /++ -sNMd: Uptime: 2h 34m
MMNso/` dMM `.::-. .-::.` .hMN: Packages: 2217
ddddMMh dMM :hNMNMNhNMNMNh: `NMm Shell: bash 4.4.20
NMm dMM .NMN/-+MMM+-/NMN` dMM Resolution: 1920x1080
NMm dMM -MMm `MMM dMM. dMM DE: Cinnamon 4.0.10
NMm dMM -MMm `MMM dMM. dMM WM: Muffin
NMm dMM .mmd `mmm yMM. dMM WM Theme: Mint-Y-Dark (Mint-Y)
NMm dMM` ..` ... ydm. dMM GTK Theme: Mint-Y [GTK2/3]
hMM- +MMd/-------...-:sdds dMM Icon Theme: Mint-Y
-NMm- :hNMNNNmdddddddddy/` dMM Font: Noto Sans 9
-dMNs-``-::::-------.`` dMM CPU: Intel Core i7-6700 @ 8x 4GHz [52.0°C]
`/dMNmy+/:-------------:/yMMM GPU: NV136
./ydNMMMMMMMMMMMMMMMMMMMMM RAM: 2544MiB / 7926MiB
.MMMMMMMMMMMMMMMMMMMИзмерваме производителността на еднопоточен сървър. Отваряме два терминала: в единия стартираме ./http_server, а в другия — . След минута във втория терминал ще се появи следната статистика:
$ wrk -c100 -d1m -t8 http://127.0.0.1:18470 -H "Host: 127.0.0.1:18470" -H "Accept-Language: en-US,en;q=0.5" -H "Connection: keep-alive"
Running 1m test @ http://127.0.0.1:18470
8 threads and 100 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 493.52us 76.70us 17.31ms 89.57%
Req/Sec 24.37k 1.81k 29.34k 68.13%
11657769 requests in 1.00m, 1.60GB read
Requests/sec: 193974.70
Transfer/sec: 27.19MBНашият еднопоточен сървър успя да обработи над 11 милиона заявки в минута, идващи от 100 връзки. Добр резултат, но можем ли да го подобрим?
Многопоточен сървър
Както беше споменато по-горе, I/O реакторът може да се създаде в отделни потоци, като по този начин се оползотворяват всичките ядра на ЦПУ. Нека приложим този подход на практика:
Показване на http_server_multithreaded.c
#include "reactor.h"
static Reactor *reactor;
#pragma omp threadprivate(reactor)
#include "common.h"
int main(void) {
#pragma omp parallel
{
SAFE_CALL((reactor = reactor_new()), NULL);
SAFE_CALL(reactor_register(reactor, new_server(true), EPOLLIN,
on_accept, NULL),
-1);
SAFE_CALL(reactor_run(reactor, SERVER_TIMEOUT_MILLIS), -1);
SAFE_CALL(reactor_destroy(reactor), -1);
}
}Сега всеки поток реактор:
static Reactor *reactor;
#pragma omp threadprivate(reactor)Обърнете внимание, че аргументът на функцията new_server() се застъпва true. Това означава, че задаваме опцията на сървърния сокет , за да го използваме в многопоточна среда. Можете да прочетете повече .
Второ опитване
Сега ще измерим производителността на многопоточния сървър:
$ wrk -c100 -d1m -t8 http://127.0.0.1:18470 -H "Host: 127.0.0.1:18470" -H "Accept-Language: en-US,en;q=0.5" -H "Connection: keep-alive"
Running 1m test @ http://127.0.0.1:18470
8 threads and 100 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 1.14ms 2.53ms 40.73ms 89.98%
Req/Sec 79.98k 18.07k 154.64k 78.65%
38208400 requests in 1.00m, 5.23GB read
Requests/sec: 635876.41
Transfer/sec: 89.14MBБроят на обработените заявки за 1 минута се увеличи с ~3.28 пъти! Но до кръгло число не достигнаха едва ~два милиона, нека опитаме да го поправим.
Първо, нека да разгледаме статистиката, генерирана :
$ sudo perf stat -B -e task-clock,context-switches,cpu-migrations,page-faults,cycles,instructions,branches,branch-misses,cache-misses ./http_server_multithreaded
Performance counter stats for './http_server_multithreaded':
242446,314933 task-clock (msec) # 4,000 CPUs utilized
1 813 074 context-switches # 0,007 M/sec
4 689 cpu-migrations # 0,019 K/sec
254 page-faults # 0,001 K/sec
895 324 830 170 cycles # 3,693 GHz
621 378 066 808 instructions # 0,69 insn per cycle
119 926 709 370 branches # 494,653 M/sec
3 227 095 669 branch-misses # 2,69% of all branches
808 664 cache-misses
60,604330670 seconds time elapsed, компилиране с -march=native, , увеличаване на броя на попаденията в , увеличаване на MAX_EVENTS и използване на EPOLLET не доведе до значително повишение на производителността. Но какво ще стане, ако увеличим броя на едновременните връзки?
Статистика при 352 едновременни връзки:
$ wrk -c352 -d1m -t8 http://127.0.0.1:18470 -H "Host: 127.0.0.1:18470" -H "Accept-Language: en-US,en;q=0.5" -H "Connection: keep-alive"
Running 1m test @ http://127.0.0.1:18470
8 threads and 352 connections
Thread Stats Avg Stdev Max +/-- Stdev
Latency 2.12ms 3.79ms 68.23ms 87.49%
Req/Sec 83.78k 12.69k 169.81k 83.59%
40006142 requests in 1.00m, 5.48GB read
Requests/sec: 665789.26
Transfer/sec: 93.34MBЖеланият резултат е постигнат, а заедно с него и интересна графика, демонстрираща зависимостта на броя на обработените заявки за 1 минута от броя на връзките:
Виждаме, че след няколко стотин връзки, броят на обработените заявки рязко намалява и при двата сървъра (при многопоточния вариант е по-забележимо). Свързано ли е това с реализирания TCP/IP стек на Linux? Споделете вашите предположения относно такова поведение на графиката и оптимизации на многопоточните и еднопоточни варианти в коментарите.
Как в коментарите, този тест за производителност не показва поведението на I/O реактора при реални натоварвания, тъй като почти винаги сървърът взаимодейства с БД, извежда логове, използва криптография с и т.н., в резултат на което натоварването става нехомогенно (динамично). Тестовете заедно с външни компоненти ще бъдат проведени в статия за I/O проактора.
Недостатъци на I/O реактора
Трябва да разберем, че I/O реакторът не е без недостатъци, а именно:
- Използването на I/O реактора в многопоточна среда е малко по-сложно, тъй като ще се наложи ръчно управление на потоковете.
- Практиката показва, че в повечето случаи натоварването е нееднородно, което може да доведе до ситуация, в която един поток ще обработва данни, докато друг ще бъде зает с работа.
- Ако един обработчик на събития блокира потока, то също ще блокира и самия системен селектор, което може да доведе до трудно откриваеми бъгове.
Тези проблеми решава , който обикновено има планировчик, който равномерно разпределя натоварването в пул от потоци, и също така предлага по-удобен API. За него ще говорим по-късно в моята друга статия.
Заключение
С това нашето пътуване от теория направо към изпускането на профайлера приключи.
Не бива да спираме дотук, тъй като съществуват множество други не по-малко интересни подходи за създаване на мрежов софтуер с различно ниво на удобство и скорост. Интересни, според мен, връзки са посочени по-долу.
До нови срещи!
Интересни проекти
- Си
Какво още да прочетете?
Източник: habr.com
