source: benchmark/io/http/http_ring.cpp@ d1b70d4

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

httpc now supports attach feature.
  • 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 return;
259}
260
261static void handle_answer(struct io_uring * ring, struct request_t * in, int res) {
262 if( res < 0 ) {
263 int err = -res;
264 switch(err) {
265 case EPIPE:
266 case ECONNRESET:
267 close(in->fd);
268 free(in);
269 return;
270 default:
271 std::cerr << "answer error: (" << err << ") " << strerror(err) << std::endl;
272 exit(EXIT_FAILURE);
273 }
274 }
275
276 if( res >= in->length ) {
277 free(in);
278 return;
279 }
280
281 struct io_uring_sqe * sqe = get_sqe(ring);
282 io_uring_prep_write(sqe, in->fd, in->buff + res, in->length - res, 0);
283 io_uring_sqe_set_data(sqe, in);
284 submit(ring);
285 // std::cout << "Re-Submitted request: " << in << " (" << (void*)in->buffer << ")"<<std::endl;
286
287 ring_request(ring, in->fd);
288}
289
290//=========================================================
291extern "C" {
292extern int __io_uring_flush_sq(struct io_uring *ring);
293}
294
295void * proc_loop(void * arg) {
296 size_t count = 0;
297 struct options_t & opt = *(struct options_t *)arg;
298
299 struct io_uring * ring = opt.ring;
300
301 char endfd_buf[8];
302 ring_end(ring, opt.endfd, endfd_buf, 8);
303
304 ring_accept(ring, opt.acpt.sockfd, opt.acpt.addr, opt.acpt.addrlen, opt.acpt.flags);
305
306 bool done = false;
307 while(!done) {
308 struct io_uring_cqe *cqe;
309 int ret;
310 while(-EAGAIN == (ret = io_uring_wait_cqe_nr(ring, &cqe, 0))) {
311 ret = io_uring_submit_and_wait(ring, 1);
312 if (ret < 0) {
313 fprintf( stderr, "io_uring get error: (%d) %s\n", (int)-ret, strerror(-ret) );
314 exit(EXIT_FAILURE);
315 }
316 opt.result.subs += ret;
317 opt.result.cnts++;
318 }
319
320 if (ret < 0 && -EAGAIN != ret) {
321 fprintf( stderr, "io_uring peek error: (%d) %s\n", (int)-ret, strerror(-ret) );
322 exit(EXIT_FAILURE);
323 }
324
325 auto req = (struct request_t *)cqe->user_data;
326 // std::cout << req << " completed with " << cqe->res << std::endl;
327
328 switch(req->type) {
329 case EVENT_END:
330 done = true;
331 break;
332 case EVENT_ACCEPT:
333 handle_new_conn(ring, cqe->res);
334 free(req);
335 ring_accept(ring, opt.acpt.sockfd, opt.acpt.addr, opt.acpt.addrlen, opt.acpt.flags);
336 break;
337 case EVENT_REQUEST:
338 handle_request(ring, req, cqe->res);
339 break;
340 case EVENT_ANSWER:
341 handle_answer(ring, req, cqe->res);
342 break;
343 }
344
345 io_uring_cqe_seen(ring, cqe);
346 }
347
348 return (void*)count;
349}
350
351//=========================================================
352struct __attribute__((aligned(128))) aligned_ring {
353 struct io_uring storage;
354};
355
356#include <bit>
357
358#include <pthread.h>
359extern "C" {
360 #include <signal.h>
361 #include <sys/eventfd.h>
362 #include <sys/socket.h>
363 #include <netinet/in.h>
364}
365
366int main(int argc, char * argv[]) {
367 signal(SIGPIPE, SIG_IGN);
368
369 unsigned nthreads = 1;
370 unsigned port = 8800;
371 unsigned entries = 256;
372 unsigned backlog = 10;
373 bool attach = false;
374
375 //===================
376 // Arguments
377 int c;
378 while ((c = getopt (argc, argv, "t:p:e:b:a")) != -1) {
379 switch (c)
380 {
381 case 't':
382 nthreads = atoi(optarg);
383 break;
384 case 'p':
385 port = atoi(optarg);
386 break;
387 case 'e':
388 entries = atoi(optarg);
389 break;
390 case 'b':
391 backlog = atoi(optarg);
392 break;
393 case 'a':
394 attach = true;
395 break;
396 case '?':
397 default:
398 std::cerr << "Usage: -t <threads> -p <port> -e <entries> -b <backlog> -a" << std::endl;
399 return EXIT_FAILURE;
400 }
401 }
402
403 if( !std::ispow2(entries) ) {
404 unsigned v = entries;
405 v--;
406 v |= v >> 1;
407 v |= v >> 2;
408 v |= v >> 4;
409 v |= v >> 8;
410 v |= v >> 16;
411 v++;
412 std::cerr << "Warning: num_entries not a power of 2 (" << entries << ") raising to " << v << std::endl;
413 entries = v;
414 }
415
416 //===================
417 // End FD
418 int efd = eventfd(0, EFD_SEMAPHORE);
419 if (efd < 0) {
420 std::cerr << "eventfd error: (" << errno << ") " << strerror(errno) << std::endl;
421 exit(EXIT_FAILURE);
422 }
423
424 //===================
425 // Open Socket
426 std::cout << getpid() << " : Listening on port " << port << std::endl;
427 int server_fd = socket(AF_INET, SOCK_STREAM, 0);
428 if(server_fd < 0) {
429 std::cerr << "socket error: (" << errno << ") " << strerror(errno) << std::endl;
430 exit(EXIT_FAILURE);
431 }
432
433 int ret = 0;
434 struct sockaddr_in address;
435 int addrlen = sizeof(address);
436 memset( (char *)&address, '\0', addrlen );
437 address.sin_family = AF_INET;
438 address.sin_addr.s_addr = htonl(INADDR_ANY);
439 address.sin_port = htons( port );
440
441 int waited = 0;
442 while(true) {
443 ret = bind( server_fd, (struct sockaddr *)&address, sizeof(address) );
444 if(ret < 0) {
445 if(errno == EADDRINUSE) {
446 if(waited == 0) {
447 std::cerr << "Waiting for port" << std::endl;
448 } else {
449 std::cerr << "\r" << waited;
450 std::cerr.flush();
451 }
452 waited ++;
453 usleep( 1000000 );
454 continue;
455 }
456 std::cerr << "bind error: (" << errno << ") " << strerror(errno) << std::endl;
457 exit(EXIT_FAILURE);
458 }
459 break;
460 }
461
462 ret = listen( server_fd, backlog );
463 if(ret < 0) {
464 std::cerr << "listen error: (" << errno << ") " << strerror(errno) << std::endl;
465 exit(EXIT_FAILURE);
466 }
467
468 //===================
469 // Run Server Threads
470 std::cout << "Starting " << nthreads << " Threads";
471 if(attach) {
472 std::cout << " with attached Rings";
473 }
474 std::cout << std::endl;
475
476 aligned_ring thrd_rings[nthreads];
477 pthread_t thrd_hdls[nthreads];
478 options_t thrd_opts[nthreads];
479 for(unsigned i = 0; i < nthreads; i++) {
480 if(!attach || i == 0) {
481 io_uring_queue_init(entries, &thrd_rings[i].storage, 0);
482 }
483 else {
484 struct io_uring_params p;
485 memset(&p, 0, sizeof(p));
486 p.flags = IORING_SETUP_ATTACH_WQ;
487 p.wq_fd = thrd_rings[0].storage.ring_fd;
488 io_uring_queue_init_params(entries, &thrd_rings[i].storage, &p);
489 }
490
491 thrd_opts[i].acpt.sockfd = server_fd;
492 thrd_opts[i].acpt.addr = (struct sockaddr *)&address;
493 thrd_opts[i].acpt.addrlen = (socklen_t*)&addrlen;
494 thrd_opts[i].acpt.flags = 0;
495 thrd_opts[i].endfd = efd;
496 thrd_opts[i].ring = &thrd_rings[i].storage;
497
498 int ret = pthread_create(&thrd_hdls[i], nullptr, proc_loop, &thrd_opts[i]);
499 if (ret < 0) {
500 std::cerr << "pthread create error: (" << errno << ") " << strerror(errno) << std::endl;
501 exit(EXIT_FAILURE);
502 }
503 }
504
505 //===================
506 // Server Started
507 std::cout << "Server Started" << std::endl;
508 {
509 char buffer[128];
510 int ret;
511 do {
512 ret = read(STDIN_FILENO, buffer, 128);
513 if(ret < 0) {
514 std::cerr << "main read error: (" << errno << ") " << strerror(errno) << std::endl;
515 exit(EXIT_FAILURE);
516 }
517 else if(ret > 0) {
518 std::cout << "User inputed '";
519 std::cout.write(buffer, ret);
520 std::cout << "'" << std::endl;
521 }
522 } while(ret != 0);
523
524 std::cout << "Shutdown received" << std::endl;
525 }
526
527 //===================
528 (std::cout << "Sending Shutdown to Threads... ").flush();
529 ret = eventfd_write(efd, nthreads);
530 if (ret < 0) {
531 std::cerr << "eventfd close error: (" << errno << ") " << strerror(errno) << std::endl;
532 exit(EXIT_FAILURE);
533 }
534 std::cout << "done" << std::endl;
535
536 //===================
537 (std::cout << "Stopping Threads Done... ").flush();
538 size_t total = 0;
539 size_t count = 0;
540 for(unsigned i = 0; i < nthreads; i++) {
541 void * retval;
542 int ret = pthread_join(thrd_hdls[i], &retval);
543 if (ret < 0) {
544 std::cerr << "pthread create error: (" << errno << ") " << strerror(errno) << std::endl;
545 exit(EXIT_FAILURE);
546 }
547 // total += (size_t)retval;
548 total += thrd_opts[i].result.subs;
549 count += thrd_opts[i].result.cnts;
550
551 io_uring_queue_exit(thrd_opts[i].ring);
552 }
553 std::cout << "done" << std::endl;
554 std::cout << "Submit average: " << total << "/" << count << "(" << (((double)total) / count) << ")" << std::endl;
555
556 //===================
557 (std::cout << "Closing Socket... ").flush();
558 ret = shutdown( server_fd, SHUT_RD );
559 if( ret < 0 ) {
560 std::cerr << "shutdown socket error: (" << errno << ") " << strerror(errno) << std::endl;
561 exit(EXIT_FAILURE);
562 }
563
564 ret = close(server_fd);
565 if (ret < 0) {
566 std::cerr << "close socket error: (" << errno << ") " << strerror(errno) << std::endl;
567 exit(EXIT_FAILURE);
568 }
569 std::cout << "done" << std::endl;
570}
Note: See TracBrowser for help on using the repository browser.