source: benchmark/io/http/http_ring.cpp@ 7cc532b

ADT arm-eh ast-experimental enum forall-pointer-decay jacob/cs343-translation new-ast-unique-expr pthread-emulation qualifiedEnum stuck-waitfor-destruct
Last change on this file since 7cc532b was 7cc532b, checked in by Thierry Delisle <tdelisle@…>, 5 years ago

Fixed memory leak
  • Property mode set to 100644
File size: 14.2 KB
Line 
1#include <cstdio>
2#include <cstdlib>
3#include <cstring>
4
5#include <iostream>
6
7#include <signal.h>
8#include <unistd.h>
9#include <liburing.h>
10
11typedef enum {
12 EVENT_END,
13 EVENT_ACCEPT,
14 EVENT_REQUEST,
15 EVENT_ANSWER
16} event_t;
17
18struct __attribute__((aligned(128))) request_t {
19 event_t type;
20 int fd;
21 size_t length;
22 char * buff;
23 char data[0];
24
25 static struct request_t * create(event_t type, size_t extra) {
26 auto ret = (struct request_t *)malloc(sizeof(struct request_t) + extra);
27 ret->type = type;
28 ret->length = extra;
29 ret->buff = ret->data;
30 return ret;
31 }
32
33 static struct request_t * create(event_t type) {
34 return create(type, 0);
35 }
36};
37
38struct __attribute__((aligned(128))) options_t {
39 struct {
40 int sockfd;
41 struct sockaddr *addr;
42 socklen_t *addrlen;
43 int flags;
44 } acpt;
45
46 int endfd;
47 struct io_uring * ring;
48
49 struct {
50 size_t subs = 0;
51 size_t cnts = 0;
52 } result;
53};
54
55//=========================================================
56static struct io_uring_sqe * get_sqe(struct io_uring * ring) {
57 struct io_uring_sqe * sqe = io_uring_get_sqe(ring);
58 if(!sqe) {
59 std::cerr << "Insufficient entries in ring" << std::endl;
60 exit(EXIT_FAILURE);
61 }
62 return sqe;
63}
64
65static void submit(struct io_uring * ) {
66 // io_uring_submit(ring);
67}
68
69//=========================================================
70static void ring_end(struct io_uring * ring, int fd, char * buffer, size_t len) {
71 struct io_uring_sqe * sqe = get_sqe(ring);
72 io_uring_prep_read(sqe, fd, buffer, len, 0);
73 io_uring_sqe_set_data(sqe, request_t::create(EVENT_END));
74 submit(ring);
75}
76
77static void ring_accept(struct io_uring * ring, int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags) {
78 auto req = request_t::create(EVENT_ACCEPT);
79 struct io_uring_sqe * sqe = get_sqe(ring);
80 io_uring_prep_accept(sqe, sockfd, addr, addrlen, flags);
81 io_uring_sqe_set_data(sqe, req);
82 submit(ring);
83 // std::cout << "Submitted accept: " << req << std::endl;
84}
85
86static void ring_request(struct io_uring * ring, int fd) {
87 size_t size = 1024;
88 auto req = request_t::create(EVENT_REQUEST, size);
89 req->fd = fd;
90
91 struct io_uring_sqe * sqe = get_sqe(ring);
92 io_uring_prep_read(sqe, fd, req->buff, size, 0);
93 io_uring_sqe_set_data(sqe, req);
94 submit(ring);
95 // std::cout << "Submitted request: " << req << " (" << (void*)req->buffer << ")"<<std::endl;
96}
97
98//=========================================================
99enum HttpCode {
100 OK200 = 0,
101 E400,
102 E404,
103 E405,
104 E408,
105 E413,
106 E414,
107 KNOWN_CODES
108};
109
110const char * http_msgs[] = {
111 "HTTP/1.1 200 OK\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: %zu \n\n%s",
112 "HTTP/1.1 400 Bad Request\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
113 "HTTP/1.1 404 Not Found\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
114 "HTTP/1.1 405 Method Not Allowed\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
115 "HTTP/1.1 408 Request Timeout\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
116 "HTTP/1.1 413 Payload Too Large\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
117 "HTTP/1.1 414 URI Too Long\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
118};
119
120static_assert( KNOWN_CODES == (sizeof(http_msgs ) / sizeof(http_msgs [0])));
121
122const int http_codes[] = {
123 200,
124 400,
125 404,
126 405,
127 408,
128 413,
129 414,
130};
131
132static_assert( KNOWN_CODES == (sizeof(http_codes) / sizeof(http_codes[0])));
133
134int code_val(HttpCode code) {
135 return http_codes[code];
136}
137
138static void ring_answer(struct io_uring * ring, int fd, HttpCode code) {
139 size_t size = 256;
140 auto req = request_t::create(EVENT_ANSWER, size);
141 req->fd = fd;
142
143 const char * fmt = http_msgs[code];
144 const char * date = "";
145 size = snprintf(req->buff, size, fmt, date, size);
146
147 struct io_uring_sqe * sqe = get_sqe(ring);
148 io_uring_prep_write(sqe, fd, req->buff, size, 0);
149 io_uring_sqe_set_data(sqe, req);
150 submit(ring);
151 // std::cout << "Submitted good answer: " << req << " (" << (void*)req->buffer << ")"<<std::endl;
152}
153
154static void ring_answer(struct io_uring * ring, int fd, const std::string &) {
155 // size_t size = 256;
156 // auto req = request_t::create(EVENT_ANSWER, size);
157 // req->fd = fd;
158
159 // const char * fmt = http_msgs[OK200];
160 // const char * date = "";
161 // size_t len = snprintf(req->buffer, size, fmt, date, ans.size(), ans.c_str());
162 // req->length = len;
163
164 // struct io_uring_sqe * sqe = get_sqe(ring);
165 // io_uring_prep_write(sqe, fd, req->buffer, len, 0);
166 // io_uring_sqe_set_data(sqe, req);
167 // submit(ring);
168 // std::cout << "Submitted good answer: " << req << " (" << (void*)req->buffer << ")"<<std::endl;
169
170
171 static const char* RESPONSE = "HTTP/1.1 200 OK\r\n" \
172 "Content-Length: 15\r\n" \
173 "Content-Type: text/html\r\n" \
174 "Connection: keep-alive\r\n" \
175 "Server: testserver\r\n" \
176 "\r\n" \
177 "Hello, World!\r\n";
178
179 static const size_t RLEN = strlen(RESPONSE);
180
181 size_t size = 256;
182 auto req = request_t::create(EVENT_ANSWER, size);
183 req->fd = fd;
184 req->buff = (char*)RESPONSE;
185 req->length = RLEN;
186
187 // const char * fmt = http_msgs[OK200];
188 // const char * date = "";
189 // size_t len = snprintf(req->buffer, size, fmt, date, ans.size(), ans.c_str());
190 // req->length = len;
191
192 struct io_uring_sqe * sqe = get_sqe(ring);
193 io_uring_prep_write(sqe, fd, RESPONSE, RLEN, 0);
194 io_uring_sqe_set_data(sqe, req);
195 submit(ring);
196}
197
198//=========================================================
199static void handle_new_conn(struct io_uring * ring, int fd) {
200 if( fd < 0 ) {
201 int err = -fd;
202 if( err == ECONNABORTED ) return;
203 std::cerr << "accept error: (" << errno << ") " << strerror(errno) << std::endl;
204 exit(EXIT_FAILURE);
205 }
206
207 ring_request(ring, fd);
208}
209
210static void handle_request(struct io_uring * ring, struct request_t * in, int res) {
211 if( res < 0 ) {
212 int err = -res;
213 switch(err) {
214 case EPIPE:
215 case ECONNRESET:
216 close(in->fd);
217 free(in);
218 return;
219 default:
220 std::cerr << "request error: (" << err << ") " << strerror(err) << std::endl;
221 exit(EXIT_FAILURE);
222 }
223 }
224
225 if(res == 0) {
226 close(in->fd);
227 free(in);
228 return;
229 }
230
231 const char * it = in->buff;
232 if( !strstr( it, "\r\n\r\n" ) ) {
233 std::cout << "Incomplete request" << std::endl;
234 close(in->fd);
235 free(in);
236 return;
237 }
238
239 it = in->buff;
240 const std::string reply = "Hello, World!\n";
241 int ret = memcmp(it, "GET ", 4);
242 if( ret != 0 ) {
243 ring_answer(ring, in->fd, E400);
244 goto NEXT;
245 }
246
247 it += 4;
248 ret = memcmp(it, "/plaintext", 10);
249 if( ret != 0 ) {
250 ring_answer(ring, in->fd, E404);
251 goto NEXT;
252 }
253
254 ring_answer(ring, in->fd, reply);
255
256 NEXT:
257 ring_request(ring, in->fd);
258 free(in);
259 return;
260}
261
262static void handle_answer(struct io_uring * ring, struct request_t * in, int res) {
263 if( res < 0 ) {
264 int err = -res;
265 switch(err) {
266 case EPIPE:
267 case ECONNRESET:
268 close(in->fd);
269 free(in);
270 return;
271 default:
272 std::cerr << "answer error: (" << err << ") " << strerror(err) << std::endl;
273 exit(EXIT_FAILURE);
274 }
275 }
276
277 if( res >= in->length ) {
278 free(in);
279 return;
280 }
281
282 struct io_uring_sqe * sqe = get_sqe(ring);
283 io_uring_prep_write(sqe, in->fd, in->buff + res, in->length - res, 0);
284 io_uring_sqe_set_data(sqe, in);
285 submit(ring);
286 // std::cout << "Re-Submitted request: " << in << " (" << (void*)in->buffer << ")"<<std::endl;
287
288 ring_request(ring, in->fd);
289}
290
291//=========================================================
292extern "C" {
293extern int __io_uring_flush_sq(struct io_uring *ring);
294}
295
296void * proc_loop(void * arg) {
297 size_t count = 0;
298 struct options_t & opt = *(struct options_t *)arg;
299
300 struct io_uring * ring = opt.ring;
301
302 char endfd_buf[8];
303 ring_end(ring, opt.endfd, endfd_buf, 8);
304
305 ring_accept(ring, opt.acpt.sockfd, opt.acpt.addr, opt.acpt.addrlen, opt.acpt.flags);
306
307 bool done = false;
308 while(!done) {
309 struct io_uring_cqe *cqe;
310 int ret;
311 while(-EAGAIN == (ret = io_uring_wait_cqe_nr(ring, &cqe, 0))) {
312 ret = io_uring_submit_and_wait(ring, 1);
313 if (ret < 0) {
314 fprintf( stderr, "io_uring get error: (%d) %s\n", (int)-ret, strerror(-ret) );
315 exit(EXIT_FAILURE);
316 }
317 opt.result.subs += ret;
318 opt.result.cnts++;
319 }
320
321 if (ret < 0 && -EAGAIN != ret) {
322 fprintf( stderr, "io_uring peek error: (%d) %s\n", (int)-ret, strerror(-ret) );
323 exit(EXIT_FAILURE);
324 }
325
326 auto req = (struct request_t *)cqe->user_data;
327 // std::cout << req << " completed with " << cqe->res << std::endl;
328
329 switch(req->type) {
330 case EVENT_END:
331 done = true;
332 break;
333 case EVENT_ACCEPT:
334 handle_new_conn(ring, cqe->res);
335 free(req);
336 ring_accept(ring, opt.acpt.sockfd, opt.acpt.addr, opt.acpt.addrlen, opt.acpt.flags);
337 break;
338 case EVENT_REQUEST:
339 handle_request(ring, req, cqe->res);
340 break;
341 case EVENT_ANSWER:
342 handle_answer(ring, req, cqe->res);
343 break;
344 }
345
346 io_uring_cqe_seen(ring, cqe);
347 }
348
349 return (void*)count;
350}
351
352//=========================================================
353struct __attribute__((aligned(128))) aligned_ring {
354 struct io_uring storage;
355};
356
357#include <bit>
358
359#include <pthread.h>
360extern "C" {
361 #include <signal.h>
362 #include <sys/eventfd.h>
363 #include <sys/socket.h>
364 #include <netinet/in.h>
365}
366
367int main(int argc, char * argv[]) {
368 signal(SIGPIPE, SIG_IGN);
369
370 unsigned nthreads = 1;
371 unsigned port = 8800;
372 unsigned entries = 256;
373 unsigned backlog = 10;
374 bool attach = false;
375
376 //===================
377 // Arguments
378 int c;
379 while ((c = getopt (argc, argv, "t:p:e:b:a")) != -1) {
380 switch (c)
381 {
382 case 't':
383 nthreads = atoi(optarg);
384 break;
385 case 'p':
386 port = atoi(optarg);
387 break;
388 case 'e':
389 entries = atoi(optarg);
390 break;
391 case 'b':
392 backlog = atoi(optarg);
393 break;
394 case 'a':
395 attach = true;
396 break;
397 case '?':
398 default:
399 std::cerr << "Usage: -t <threads> -p <port> -e <entries> -b <backlog> -a" << std::endl;
400 return EXIT_FAILURE;
401 }
402 }
403
404 if( !std::ispow2(entries) ) {
405 unsigned v = entries;
406 v--;
407 v |= v >> 1;
408 v |= v >> 2;
409 v |= v >> 4;
410 v |= v >> 8;
411 v |= v >> 16;
412 v++;
413 std::cerr << "Warning: num_entries not a power of 2 (" << entries << ") raising to " << v << std::endl;
414 entries = v;
415 }
416
417 //===================
418 // End FD
419 int efd = eventfd(0, EFD_SEMAPHORE);
420 if (efd < 0) {
421 std::cerr << "eventfd error: (" << errno << ") " << strerror(errno) << std::endl;
422 exit(EXIT_FAILURE);
423 }
424
425 //===================
426 // Open Socket
427 std::cout << getpid() << " : Listening on port " << port << std::endl;
428 int server_fd = socket(AF_INET, SOCK_STREAM, 0);
429 if(server_fd < 0) {
430 std::cerr << "socket error: (" << errno << ") " << strerror(errno) << std::endl;
431 exit(EXIT_FAILURE);
432 }
433
434 int ret = 0;
435 struct sockaddr_in address;
436 int addrlen = sizeof(address);
437 memset( (char *)&address, '\0', addrlen );
438 address.sin_family = AF_INET;
439 address.sin_addr.s_addr = htonl(INADDR_ANY);
440 address.sin_port = htons( port );
441
442 int waited = 0;
443 while(true) {
444 ret = bind( server_fd, (struct sockaddr *)&address, sizeof(address) );
445 if(ret < 0) {
446 if(errno == EADDRINUSE) {
447 if(waited == 0) {
448 std::cerr << "Waiting for port" << std::endl;
449 } else {
450 std::cerr << "\r" << waited;
451 std::cerr.flush();
452 }
453 waited ++;
454 usleep( 1000000 );
455 continue;
456 }
457 std::cerr << "bind error: (" << errno << ") " << strerror(errno) << std::endl;
458 exit(EXIT_FAILURE);
459 }
460 break;
461 }
462
463 ret = listen( server_fd, backlog );
464 if(ret < 0) {
465 std::cerr << "listen error: (" << errno << ") " << strerror(errno) << std::endl;
466 exit(EXIT_FAILURE);
467 }
468
469 //===================
470 // Run Server Threads
471 std::cout << "Starting " << nthreads << " Threads";
472 if(attach) {
473 std::cout << " with attached Rings";
474 }
475 std::cout << std::endl;
476
477 aligned_ring thrd_rings[nthreads];
478 pthread_t thrd_hdls[nthreads];
479 options_t thrd_opts[nthreads];
480 for(unsigned i = 0; i < nthreads; i++) {
481 if(!attach || i == 0) {
482 io_uring_queue_init(entries, &thrd_rings[i].storage, 0);
483 }
484 else {
485 struct io_uring_params p;
486 memset(&p, 0, sizeof(p));
487 p.flags = IORING_SETUP_ATTACH_WQ;
488 p.wq_fd = thrd_rings[0].storage.ring_fd;
489 io_uring_queue_init_params(entries, &thrd_rings[i].storage, &p);
490 }
491
492 thrd_opts[i].acpt.sockfd = server_fd;
493 thrd_opts[i].acpt.addr = (struct sockaddr *)&address;
494 thrd_opts[i].acpt.addrlen = (socklen_t*)&addrlen;
495 thrd_opts[i].acpt.flags = 0;
496 thrd_opts[i].endfd = efd;
497 thrd_opts[i].ring = &thrd_rings[i].storage;
498
499 int ret = pthread_create(&thrd_hdls[i], nullptr, proc_loop, &thrd_opts[i]);
500 if (ret < 0) {
501 std::cerr << "pthread create error: (" << errno << ") " << strerror(errno) << std::endl;
502 exit(EXIT_FAILURE);
503 }
504 }
505
506 //===================
507 // Server Started
508 std::cout << "Server Started" << std::endl;
509 {
510 char buffer[128];
511 int ret;
512 do {
513 ret = read(STDIN_FILENO, buffer, 128);
514 if(ret < 0) {
515 std::cerr << "main read error: (" << errno << ") " << strerror(errno) << std::endl;
516 exit(EXIT_FAILURE);
517 }
518 else if(ret > 0) {
519 std::cout << "User inputed '";
520 std::cout.write(buffer, ret);
521 std::cout << "'" << std::endl;
522 }
523 } while(ret != 0);
524
525 std::cout << "Shutdown received" << std::endl;
526 }
527
528 //===================
529 (std::cout << "Sending Shutdown to Threads... ").flush();
530 ret = eventfd_write(efd, nthreads);
531 if (ret < 0) {
532 std::cerr << "eventfd close error: (" << errno << ") " << strerror(errno) << std::endl;
533 exit(EXIT_FAILURE);
534 }
535 std::cout << "done" << std::endl;
536
537 //===================
538 (std::cout << "Stopping Threads Done... ").flush();
539 size_t total = 0;
540 size_t count = 0;
541 for(unsigned i = 0; i < nthreads; i++) {
542 void * retval;
543 int ret = pthread_join(thrd_hdls[i], &retval);
544 if (ret < 0) {
545 std::cerr << "pthread create error: (" << errno << ") " << strerror(errno) << std::endl;
546 exit(EXIT_FAILURE);
547 }
548 // total += (size_t)retval;
549 total += thrd_opts[i].result.subs;
550 count += thrd_opts[i].result.cnts;
551
552 io_uring_queue_exit(thrd_opts[i].ring);
553 }
554 std::cout << "done" << std::endl;
555 std::cout << "Submit average: " << total << "/" << count << "(" << (((double)total) / count) << ")" << std::endl;
556
557 //===================
558 (std::cout << "Closing Socket... ").flush();
559 ret = shutdown( server_fd, SHUT_RD );
560 if( ret < 0 ) {
561 std::cerr << "shutdown socket error: (" << errno << ") " << strerror(errno) << std::endl;
562 exit(EXIT_FAILURE);
563 }
564
565 ret = close(server_fd);
566 if (ret < 0) {
567 std::cerr << "close socket error: (" << errno << ") " << strerror(errno) << std::endl;
568 exit(EXIT_FAILURE);
569 }
570 std::cout << "done" << std::endl;
571}
Note: See TracBrowser for help on using the repository browser.