Introduktion
(enkelttrådad ) — är ett mönster för att skriva högpresterande programvara som används i många populära lösningar:
- …
I den här artikeln kommer vi att titta på grunderna för I/O reaktorn och dess funktion, skriva en implementation på mindre än 200 rader kod och få en enkel HTTP-server att hantera över 40 miljoner förfrågningar per minut.
Förord
- Artikeln är skriven för att hjälpa till att förstå hur I/O reaktorn fungerar, vilket också innebär att förstå riskerna med dess användning.
- För att förstå artikeln krävs grundläggande kunskaper i och en liten erfarenhet av utveckling av nätverksapplikationer.
- All kod är skriven på C-språket strikt enligt (obs: lång PDF) för Linux och är tillgänglig på .
Varför behövs detta?
Med den växande populariteten av internet har webbservrar behövt hantera ett stort antal anslutningar samtidigt, vilket har lett till två tillvägagångssätt: blockering av I/O med ett stort antal operativsystemstrådar och icke-blockerande I/O i kombination med ett händelseavisering system, även kallad "systemväljare" (///etc).
Det första tillvägagångssättet innebar att en ny OS-tråd skapades för varje inkommande anslutning. Dess nackdel är dålig skalbarhet: operativsystemet måste utföra många och Dessa är dyra operationer och kan leda till brist på ledigt RAM vid ett stort antal anslutningar.
Den modifierade versionen tilldelar (trådpool), vilket förhindrar att systemet oväntat avslutar körningen, men å sin sida inför en ny problematik: om trådpoolen just nu blockerar långvariga läsningsoperationer, kan andra sockets som redan är redo att ta emot data, inte göra det.
Det andra tillvägagångssättet använder (systemväljare) som tillhandahålls av OS. I denna artikel behandlas den vanligaste typen av systemväljare, baserad på aviseringar (event, notifieringar) om beredskap för I/O-operationer snarare än på Ett förenklat exempel på dess användning kan representeras av följande flödesdiagram:
Skillnaden mellan dessa tillvägagångssätt är som följer:
- Blockerande I/O-operationer stoppar användarflöde så länge, tills operativsystemet korrekt inkommande till en byteström (, datamottagning) eller tills det finns tillräckligt med utrymme i interna skrivbuffertar för efterföljande sändning genom (dataskickande).
- Systemväljare efter en tid meddelar programmet att operativsystemet redan har defragmenterat IP-paket (TCP, datamottagning) eller att det finns tillräckligt med utrymme i interna skrivbuffertar redan tillgängligt (dataskickande).
Sammanfattningsvis, att reservera flödet för varje I/O i operativsystemet är en ren slöseri av datorkraft, eftersom flödena egentligen inte utför något nyttigt arbete (härifrån kommer termen ). Systemväljaren löser detta problem genom att tillåta användarprogrammet att använda CPU-resurserna betydligt mer effektivt.
I/O reaktormodell
I/O-reaktorn fungerar som ett lager mellan systemväljaren och användarkoden. För principen bakom dess arbete, se följande flödesschema:
- Jag påminner om att en händelse är en signal om att en viss socket är redo att utföra en icke-blockerande I/O-operation.
- Händelsehanteraren är en funktion som kallas av I/O-reaktorn vid mottagandet av en händelse, vilken sedan utför en icke-blockerande I/O-operation.
Det är viktigt att notera att I/O-reaktorn per definition är entrådad, men det hindrar inte användningen av konceptet i en flertrådad miljö i förhållande 1 tråd: 1 reaktor, vilket därmed utnyttjar alla CPU-kärnor.
Implementering
Den offentliga gränssnittet placerar vi i filen , och implementeringen i . reactor.h kommer att bestå av följande deklarationer:
Visa deklarationer i reactor.h
typedef struct reactor Reactor;
/*
* Pekare till en funktion som kommer att anropas av I/O-reaktorn när en
* händelse inträffar från systemväljaren.
*/
typedef void (*Callback)(void *arg, int fd, uint32_t events);
/*
* Returnerar `NULL` vid fel, icke-`NULL` pekare till `Reactor` annars.
*/
Reactor *reactor_new(void);
/*
* Frigör systemväljaren, alla registrerade sockets vid tidpunkten
* och själva I/O-reaktorn.
*
* Följande funktioner returnerar -1 vid fel, 0 vid framgång.
*/
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);
/*
* Startar en händelseloop med timeout `timeout`.
*
* Denna funktion kommer att överföra kontrollen till den anropande koden om den angivna tiden har löpt ut
* eller/och vid avsaknad av registrerade sockets.
*/
int reactor_run(const Reactor *reactor, time_t timeout);Strukturen för I/O-reaktorn består av väljare och , som kopplar varje socket till CallbackData (struktur av händelsehanterare och användarargument för den).
Visa Reactor och CallbackData
struct reactor {
int epoll_fd;
GHashTable *table; // (int, CallbackData)
};
typedef struct {
Callback callback;
void *arg;
} CallbackData;Observera att vi har utnyttjat möjligheten att hantera via pekare. I reactor.h deklarerar vi strukturen reactor, och i reactor.c definierar vi den, vilket hindrar användaren från att ändra dess fält direkt. Detta är ett av mönstren för , vilket smidigt passar in i C-semantiken.
Funktioner reactor_register, reactor_deregister och reactor_reregister uppdaterar listan över intresserade sockets och motsvarande händelsehanterare i systemväljaren och hash-tabellen.
Visa registreringsfunktioner
#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;
}När I/O-reaktorn fångar en händelse med deskriptor fd, anropar den den motsvarande händelsehanteraren, som tar emot fd, av genererade händelser och användarpekaren till void.
Visa funktionen 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) {
// Fel
case -1:
perror("epoll_wait");
result = -1;
goto cleanup;
// Tiden har gått ut
case 0:
result = 0;
goto cleanup;
// Lyckad operation
default:
// Anropa händelsehanterarna
for (int i = 0; i table, &fd);
callback->callback(callback->arg, fd, events[i].events);
}
}
}
cleanup:
free(events);
return result;
}Sammanfattningsvis kommer anropet av funktioner i användarkoden att se ut på följande sätt:
En-trådad server
För att testa I/O-reaktorn under hög belastning, kommer vi att skriva en enkel HTTP-webbserver som svarar med en bild på alla förfrågningar.
Kort sammanfattning av HTTP-protokollet
— det är ett protokoll , mestadels använt för interaktion mellan servern och webbläsaren.
HTTP kan enkelt användas ovanpå protokollet , genom att skicka och ta emot meddelanden i ett format som definieras av .
Förfrågningsformat
CRLF
CRLF
CRLF
CRLF CRLF
CRLF— det är en sekvens av två tecken:är lika med noll.ochn, som separerar den första raden av förfrågningen, rubrikerna och data.<КОМАНДА>— en avCONNECT,TA BORT,GET,HEAD,OPTIONS,PATCH,POST,PUT,TRACE. Webbläsaren kommer att skicka kommandot till vår serverGET, vilket betyder "Skicka mig innehållet i filen."<URI>— . Till exempel, om URI =/index.html, begär klienten startsidan av webbplatsen.<ВЕРСИЯ HTTP>— versionen av HTTP-protokollet i formatetHTTP/X.Y. Den mest använda versionen idag ärHTTP/1.1.:, som skickas till servern för vidare bearbetning.— data som krävs av servern för att utföra operationen. Det är ofta bara eller något annat format.
Svarsformat
CRLF
CRLF
CRLF
CRLF CRLF
<КОД СТАТУСА>— är ett nummer som representerar resultatet av operationen. Vår server kommer alltid att returnera status 200 (lyckad operation).<ОПИСАНИЕ СТАТУСА>— en strängrepresentation av statuskoden. För statuskod 200 är detOK.Content-Length(filstorlek) ochContent-Type: text/html(typ av returnerade data).— de begärda användardata. I vårt fall är det sökvägen till bilden i .
Fil (enkärnserver) inkluderar filen , som innehåller följande funktionsprototyper:
Visa funktionsprototyper i 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);Även den funktionella makron SAFE_CALL() och funktionen fail(). Makron jämför värdet av uttrycket med ett fel, och om villkoret uppfylls, anropar funktionen fail():
#define SAFE_CALL(call, error)
do {
if ((call) == error) {
fail("%s", #call);
}
} while (false)Funktionen fail() skriver ut de angivna argumenten i terminalen (för som ) och avslutar programmet med koden 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);
}Funktionen new_server() returnerar en filbeskrivare för "server"-socketen som skapas av systemanrop , och och som kan ta emot inkommande anslutningar i icke-blockerande läge.
Visa funktionen 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;
}- Observera att socketen initialt skapas i icke-blockerande läge med flaggan
SOCK_NONBLOCK, så att i funktionenon_accept()(läs mer) kommer systemanropetaccept()inte stoppa trådens exekvering. - Om
reuse_portärtrue, så denna funktion kommer att konfigurera socketen med alternativet genom , för att använda samma port i en flertrådad miljö (se avsnittet "Flertrådad server").
Händelsehanteraren on_accept() anropas efter att OS har genererat ett händelse EPOLLIN, vilket i detta fall betyder att en ny anslutning kan accepteras. on_accept() accepterar en ny anslutning, växlar den till icke-blockerande läge och registrerar den med händelsehanteraren on_recv() i I/O-reaktorn.
Visa funktionen 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);
}Händelsehanteraren on_recv() anropas efter att OS har genererat ett händelse EPOLLIN, vilket i detta fall betyder att anslutningen som registrerats on_accept(), är redo att ta emot data.
on_recv() läser data från anslutningen tills hela HTTP-förfrågan har mottagits, sedan registrerar den hanteraren on_send() för att skicka HTTP-svaret. Om klienten avbryter anslutningen, avregistreras socketen och stängs via .
Visa funktionen on_recv()
static void on_recv(void *arg, int fd, uint32_t events) {
RequestBuffer *buffer = arg;
// Tar emot indata så länge recv returnerar 0 eller ett fel
ssize_t nread;
while ((nread = recv(fd, buffer->data + buffer->size,
REQUEST_BUFFER_CAPACITY - buffer->size, 0)) > 0)
buffer->size += nread;
// Klienten avbröt anslutningen
if (nread == 0) {
SAFE_CALL(reactor_deregister(reactor, fd), -1);
SAFE_CALL(close(fd), -1);
request_buffer_destroy(buffer);
return;
}
// read returnerade ett fel som inte är blockering
if (errno != EAGAIN && errno != EWOULDBLOCK) {
request_buffer_destroy(buffer);
fail("read");
}
// En komplett HTTP-förfrågan har mottagits från klienten. Nu registrerar vi händelsehanteraren
// för att skicka data
if (request_buffer_is_complete(buffer)) {
request_buffer_clear(buffer);
SAFE_CALL(reactor_reregister(reactor, fd, EPOLLOUT, on_send, buffer),
-1);
}
}Händelsehanteraren on_send() anropas efter att OS har genererat ett händelse EPOLLOUT, vilket betyder att anslutningen som registrerats on_recv(), är redo att skicka data. Denna funktion skickar HTTP-svaret som innehåller HTML med en bild till klienten, och ändrar sedan händelsehanteraren tillbaka till on_recv().
Visa funktionen on_send()
static 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);
}Och slutligen, i filen http_server.c, i funktionen main() skapar vi I/O-reaktorn via reactor_new(), skapar en server-socket och registrerar den, kör reaktorn med hjälp av reactor_run() exakt i en minut och sedan frigör resurserna och avslutar programmet.
Visa 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);
}Vi kontrollerar att allt fungerar som det ska. Vi kompilerar (chmod a+x compile.sh && .\/compile.sh i projektets rotkatalog) och kör den egna servern, öppnar i webbläsaren och ser vad vi förväntade oss:
Prestandamätning
Visa specifikationerna för min maskin
$ 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
.MMMMMMMMMMMMMMMMMMMMät prestandan för en entrådad server. Öppna två terminaler: i en kör vi ./http_server, i den andra — . Efter en minut kommer följande statistik att visas i den andra terminalen:
$ 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.19MBVår entrådade server lyckades behandla över 11 miljoner begärningar per minut, från 100 anslutningar. Ett bra resultat, men kan vi förbättra det?
Många trådar server
Som nämnts tidigare kan I/O-reaktorn skapas i separata trådar, vilket utnyttjar alla CPU-kärnor. Låt oss tillämpa denna metod praktiskt:
Visa 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);
}
}Nu äger varje tråd reaktor:
static Reactor *reactor;
#pragma omp threadprivate(reactor)Observera att argumentet till funktionen new_server() är true. Det betyder att vi tilldelar serverns socket alternativet , för att använda den i en flertrådad miljö. Mer kan läsas .
Andra försöket
Nu mäter vi prestandan för den flertrådade servern:
$ 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.14MBAntalet behandlade begärningar per minut ökade med ~3,28 gånger! Men det fattades bara ~två miljoner för att nå ett jämnt tal, låt oss försöka åtgärda det.
Låt oss först titta på statistiken som genererades :
$ 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, kompilering med -march=native, , ökning av träffsäkerhet i , ökning av MAX_EVENTS och användning av EPOLLET gav ingen signifikant prestandaökning. Men vad händer om vi ökar antalet samtidiga anslutningar?
Statistik vid 352 samtidiga anslutningar:
$ 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.34MBDet önskade resultatet har uppnåtts, och tillsammans med det en intressant grafik som visar beroendet mellan antalet behandlade förfrågningar per minut och antalet anslutningar:
Vi ser att efter ett par hundra anslutningar faller antalet behandlade förfrågningar plötsligt för båda servrarna (den multitrådade varianten märks mer). Är detta relaterat till implementeringen av TCP/IP-stacken i Linux? Dela gärna dina tankar och förslag angående detta beteende och optimeringar av multitrådiga och enkla varianter i kommentarerna.
Hur i kommentarerna, att detta prestandatest inte visar hur I/O-reaktorer beter sig under verkliga belastningar, eftersom servrar nästan alltid interagerar med databaser, skriver loggar, använder kryptografi med etc., vilket gör att belastningen blir oregelbunden (dynamisk). Tester tillsammans med externa komponenter kommer att genomföras i artikeln om I/O-reaktorer.
Nackdelar med I/O-reaktorer
Det måste förstås att I/O-reaktorer inte är utan sina nackdelar, nämligen:
- Att använda I/O-reaktorer i en multitrådad miljö är något mer komplicerat, eftersom man måste hantera trådarna manuellt.
- Praxis visar att belastningen oftast är ojämn, vilket kan leda till att en tråd arbetar medan en annan är överbelastad.
- Om en eventhanterare blockerar en tråd, blockeras även systemväljaren, vilket kan leda till svårfångade buggar.
Dessa problem löses av , ofta med en schemaläggare som jämnt fördelar belastningen i trådpoolen, och dessutom har ett mer intuitivt API. Mer om det kommer senare i min andra artikel.
Slutsats
Med detta har vår resa från teori direkt till profileringsutflödet kommit till slutet.
Vi ska inte stanna här, det finns många andra intressanta tillvägagångssätt för att skriva nätverksprogramvara med olika nivåer av bekvämlighet och hastighet. Intressanta länkar, enligt min mening, finns nedan.
Vi ses snart igen!
Intressanta projekt
- C
Vad mer att läsa?
Källa: habr.com
