Pendahuluan

Reaktor I / O ( loop peristiwa tunggal) adalah pola untuk menulis perangkat lunak yang sangat banyak digunakan dalam banyak solusi populer:

• Node.js
• Tor
• Transmisi
• Chromium
• Memcached
• ...

Pada artikel ini, kita akan mempertimbangkan seluk beluk reaktor I / O dan prinsip operasinya, menulis implementasi untuk kurang dari 200 baris kode dan memaksa server HTTP sederhana untuk memproses lebih dari 40 juta permintaan / menit.

Kata Pengantar

• Artikel ini ditulis dengan tujuan membantu memahami fungsi reaktor I / O, dan karenanya menyadari risiko ketika menggunakannya.
• Untuk menguasai artikel, Anda perlu pengetahuan tentang dasar-dasar bahasa C dan sedikit pengalaman dalam mengembangkan aplikasi jaringan.
• Semua kode ditulis dalam C secara ketat dengan ( hati-hati: PDF panjang ) standar C11 untuk Linux dan tersedia di GitHub .

Mengapa ini dibutuhkan?

Dengan semakin populernya Internet, server web diperlukan untuk memproses sejumlah besar koneksi pada saat yang sama, dan karenanya dua pendekatan dicoba: memblokir I / O pada sejumlah besar thread OS dan I-O non-blocking dalam kombinasi dengan sistem notifikasi acara, juga disebut "sistem" pemilih "( epoll / kqueue / IOCP / etc).

Pendekatan pertama melibatkan pembuatan utas OS baru untuk setiap koneksi yang masuk. Kerugiannya adalah skalabilitas yang buruk: sistem operasi harus membuat banyak transisi konteks dan panggilan sistem . Mereka adalah operasi yang mahal dan dapat menyebabkan kurangnya RAM gratis dengan jumlah koneksi yang mengesankan.

Versi yang dimodifikasi mengalokasikan sejumlah thread yang tetap (pool thread), dengan demikian mencegah sistem dari eksekusi yang tidak normal, tetapi pada saat yang sama memperkenalkan masalah baru: jika pada saat tertentu pool thread diblokir oleh operasi baca yang lama, maka soket lain yang sudah dapat menerima data tidak akan bisa melakukan ini.

Pendekatan kedua menggunakan sistem notifikasi acara (pemilih sistem) yang disediakan oleh OS. Artikel ini membahas jenis pemilih sistem yang paling umum berdasarkan peringatan (acara, pemberitahuan) tentang kesiapan operasi I / O, bukan peringatan tentang penyelesaiannya . Contoh penggunaan yang disederhanakan dapat diwakili oleh diagram alur berikut:

Perbedaan antara pendekatan ini adalah sebagai berikut:

• Memblokir operasi I / O menunda aliran pengguna sampai OS dengan benar mendefragmen paket IP yang masuk ke aliran byte ( TCP , menerima data) atau membebaskan ruang yang cukup dalam buffer tulis internal untuk pengiriman selanjutnya melalui NIC (mengirim data).
• Setelah beberapa saat, pemilih sistem memberi tahu program bahwa OS telah mendefragmentasi paket IP (TCP, menerima data) atau cukup ruang dalam buffer rekaman internal sudah tersedia (mengirim data).

Untuk meringkas, pemesanan utas OS untuk setiap I / O adalah pemborosan daya komputasi, karena pada kenyataannya, utas tidak sibuk dengan pekerjaan yang bermanfaat (istilah "gangguan perangkat lunak" berakar di dalamnya ). Pemilih sistem memecahkan masalah ini dengan memungkinkan program pengguna untuk mengkonsumsi sumber daya CPU jauh lebih ekonomis.

Model I / O Reaktor

Reaktor I / O bertindak sebagai lapisan antara pemilih sistem dan kode pengguna. Prinsip operasinya dijelaskan oleh diagram alur berikut:

• Biarkan saya mengingatkan Anda bahwa suatu peristiwa adalah pemberitahuan bahwa soket tertentu dapat melakukan operasi I / O yang tidak menghalangi.
• Pengatur kejadian adalah fungsi yang dipanggil oleh reaktor I / O ketika suatu peristiwa diterima, yang kemudian melakukan operasi I / O yang tidak menghalangi.

Penting untuk dicatat bahwa reaktor I / O menurut definisi adalah single-threaded, tetapi tidak ada yang mencegah menggunakan konsep dalam lingkungan multi-threaded sehubungan dengan reaktor 1 stream: 1, dengan demikian memanfaatkan semua inti CPU.

Implementasi

Kami meletakkan antarmuka publik di file reactor.h , dan implementasinya di reactor.c . reactor.h akan terdiri dari deklarasi berikut:

 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); 

Struktur I / O reaktor terdiri dari deskriptor file pemilih epoll dan tabel hash GHashTable , yang masing-masing peta peta ke CallbackData (struktur dari pengendali event dan argumen pengguna untuk itu).

 struct reactor { int epoll_fd; GHashTable *table; // (int, CallbackData) }; typedef struct { Callback callback; void *arg; } CallbackData; 

Harap perhatikan bahwa kami telah menggunakan kemampuan untuk menangani tipe yang tidak lengkap dengan pointer. Dalam reactor.h kami mendeklarasikan struktur reactor , dan dalam reactor.c mendefinisikannya, sehingga mencegah pengguna mengubah ladangnya secara eksplisit. Ini adalah salah satu pola penyembunyian data yang cocok secara organik ke dalam semantik C.

Fungsi reactor_register , reactor_deregister dan reactor_reregister memperbarui daftar soket yang menarik dan penangan peristiwa yang sesuai di pemilih sistem dan di tabel hash.

 #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; } 

Setelah reaktor I / O mencegat peristiwa dengan deskriptor fd , ia memanggil event handler yang sesuai, ke mana ia melewati fd , topeng bit dari peristiwa yang dihasilkan, dan penunjuk pengguna void .

 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 < nfds; i++) { int fd = events[i].data.fd; CallbackData *callback = g_hash_table_lookup(reactor->table, &fd); callback->callback(callback->arg, fd, events[i].events); } } } cleanup: free(events); return result; } 

Untuk meringkas, rantai panggilan fungsi dalam kode pengguna akan mengambil bentuk berikut:

Server berulir tunggal

Untuk menguji reaktor I / O di bawah beban tinggi, kami akan menulis server web HTTP sederhana untuk menanggapi setiap permintaan dengan gambar.

 <> <URI> < HTTP>CRLF < 1>CRLF < 2>CRLF < N>CRLF CRLF <> 

 < HTTP> < > < >CRLF < 1>CRLF < 2>CRLF < N>CRLF CRLF <> 

File http_server.c (server single-threaded) termasuk file common.h , yang berisi prototipe fungsi berikut:

 /* *  ,    ,    *    . */ 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); 

Fungsi makro SAFE_CALL() juga dijelaskan dan fungsi fail() didefinisikan. Makro membandingkan nilai ekspresi dengan kesalahan, dan jika kondisi terpenuhi, ia memanggil fungsi fail() :

 #define SAFE_CALL(call, error) \ do { \ if ((call) == error) { \ fail("%s", #call); \ } \ } while (false) 

Fungsi fail() mencetak argumen yang diteruskan ke terminal (seperti printf() ) dan mengakhiri program dengan kode 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, ": %s\n", strerror(errno)); exit(EXIT_FAILURE); } 

Fungsi new_server() mengembalikan deskriptor file dari soket "server" yang dibuat oleh socket() panggilan sistem socket() , bind() dan listen() dan mampu menerima koneksi masuk dalam mode non-pemblokiran.

 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; } 

• Perhatikan bahwa soket pada awalnya dibuat dalam mode non-pemblokiran menggunakan bendera SOCK_NONBLOCK , sehingga pada fungsi on_accept() (baca lebih lanjut), panggilan sistem accept() tidak menghentikan eksekusi aliran.
• Jika reuse_port true , maka fungsi ini akan mengonfigurasi soket dengan opsi SO_REUSEPORT menggunakan setsockopt() untuk menggunakan port yang sama di lingkungan multi-threaded (lihat bagian "Server multi-threaded").

on_accept() event on_accept() dipanggil setelah OS menghasilkan event EPOLLIN , dalam hal ini berarti bahwa koneksi baru dapat diterima. on_accept() menerima koneksi baru, mengalihkannya ke mode non-blocking dan mendaftar dengan event handler on_recv() di reaktor I / O.

 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); } 

Handler event on_recv() dipanggil setelah OS menghasilkan event EPOLLIN , dalam hal ini artinya koneksi yang didaftarkan oleh on_accept() siap menerima data.

on_recv() membaca data dari koneksi sampai permintaan HTTP penuh telah diterima, kemudian register handler on_send() untuk mengirim respons HTTP. Jika klien terputus, soket deregister dan ditutup dengan close() .

 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); } } 

Handler event on_send() dipanggil setelah OS menghasilkan event EPOLLOUT , yang berarti koneksi yang terdaftar oleh on_recv() siap untuk mengirim data. Fungsi ini mengirimkan respons HTTP yang berisi HTML dengan gambar ke klien, dan kemudian mengubah pengendali acara menjadi on_recv() lagi.

 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); } 

Dan akhirnya, dalam file http_server.c , dalam fungsi main() , kami membuat reaktor I / O menggunakan reactor_new() , membuat soket server dan mendaftarkannya, mulai reaktor menggunakan reactor_run() tepat satu menit, dan kemudian lepaskan sumber daya dan keluar dari program.

 #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); } 

Periksa apakah semuanya berfungsi seperti yang diharapkan. Kami mengkompilasi ( chmod a+x compile.sh && ./compile.sh di root proyek) dan memulai server yang ditulis sendiri, buka http://127.0.0.1:18470 di browser dan amati apa yang diharapkan:

Pengukuran kinerja

 $ 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 

Kami mengukur kinerja server single-threaded. Mari kita buka dua terminal: di satu kita jalankan ./http_server , di yang lain - wrk . Setelah satu menit, statistik berikut akan ditampilkan di terminal kedua:

 $ 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 

Server single-threaded kami dapat memproses lebih dari 11 juta permintaan per menit, yang berasal dari 100 koneksi. Bukan hasil yang buruk, tetapi dapatkah ini diperbaiki?

Server multithreaded

Seperti disebutkan di atas, reaktor I / O dapat dibuat dalam aliran terpisah, sehingga memanfaatkan semua inti CPU. Mari kita terapkan pendekatan ini dalam praktik:

 #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); } } 

Sekarang setiap utas memiliki reaktornya sendiri :

 static Reactor *reactor; #pragma omp threadprivate(reactor) 

Perhatikan bahwa argumen ke new_server() true . Ini berarti bahwa kami menetapkan soket server ke opsi SO_REUSEPORT untuk menggunakannya di lingkungan multi-utas. Anda dapat membaca lebih lanjut di sini .

Jalankan kedua

Sekarang kita akan mengukur kinerja server multithreaded:

 $ 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 

Jumlah permintaan yang diproses dalam 1 menit meningkat ~ 3,28 kali! Tetapi sampai angka bulat, hanya ~ dua juta tidak cukup, mari kita coba memperbaikinya.

Pertama, lihat statistik yang dihasilkan oleh perf :

 $ 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 

Menggunakan afinitas CPU , mengkompilasi dengan -march=native , PGO , meningkatkan jumlah hit dalam cache , meningkatkan MAX_EVENTS dan menggunakan EPOLLET tidak memberikan peningkatan kinerja yang signifikan. Tetapi apa yang terjadi jika Anda meningkatkan jumlah koneksi simultan?

Statistik untuk 352 koneksi simultan:

 $ 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 

Hasil yang diinginkan diperoleh, dan disertai grafik yang menarik yang menunjukkan ketergantungan jumlah permintaan yang diproses dalam 1 menit pada jumlah koneksi:

Kami melihat bahwa setelah beberapa ratusan koneksi, jumlah permintaan yang diproses dari kedua server turun tajam (dalam versi multi-utas, ini lebih terlihat). Apakah ini terkait dengan implementasi tumpukan TCP / IP Linux? Jangan ragu untuk menulis asumsi Anda tentang perilaku grafik dan optimalisasi opsi multithreaded dan single-threaded di komentar.

Seperti disebutkan dalam komentar, tes kinerja ini tidak menunjukkan perilaku reaktor I / O pada beban nyata, karena hampir selalu server berinteraksi dengan database, menampilkan log, menggunakan kriptografi dengan TLS , dll., Akibatnya beban menjadi heterogen (dinamis). Pengujian bersama dengan komponen pihak ketiga akan dilakukan dalam artikel tentang proaktor I / O.

Kekurangan Reaktor I / O

Anda perlu memahami bahwa reaktor I / O bukan tanpa kekurangan, yaitu:

• Menggunakan reaktor I / O di lingkungan multi-ulir agak lebih sulit, karena Anda harus mengelola arus secara manual.
• Praktek menunjukkan bahwa dalam kebanyakan kasus bebannya heterogen, yang dapat mengarah pada fakta bahwa satu thread akan diletakkan sementara yang lain dimuat dengan pekerjaan.
• Jika satu pengendali acara memblokir aliran, pemilih sistem itu sendiri juga akan diblokir, yang dapat menyebabkan bug yang sulit ditangkap.

Masalah-masalah ini diselesaikan oleh pengawas I / O , seringkali dengan penjadwal yang mendistribusikan beban secara merata ke kumpulan thread, dan juga memiliki API yang lebih nyaman. Ini akan dibahas nanti di artikel saya yang lain.

Kesimpulan

Tentang ini, perjalanan kami dari teori langsung ke profiler knalpot berakhir.

Jangan memikirkan hal ini, karena ada banyak pendekatan lain yang sama menarik untuk menulis perangkat lunak jaringan dengan tingkat kenyamanan dan kecepatan yang berbeda. Menarik, menurut saya, tautan diberikan di bawah ini.

Sampai ketemu lagi!

Proyek menarik

• Si
  
  • libevent
  • libev
  • libuv
  • libevhtp
  • mengoceh
  • DPDK
  • netmap
  • Pf_ring
• Karat
  
  • Mio
  • Tokio
  • smoltcp

Apa lagi yang harus dibaca?

• https://linux.die.net/man/7/socket
• https://stackoverflow.com/questions/1050222/what-is-the-ddference-between-concurrency-and-parallelism
• http://www.kegel.com/c10k.html
• https://kernel.dk/io_uring.pdf
• https://aturon.imtqy.com/blog/2016/09/07/futures-design/
• https://tokio.rs/blog/2019-10-scheduler/
• https://www.artima.com/articles/io_design_patterns.html
• https://habr.com/en/post/183832/