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

arm-ehjacob/cs343-translationnew-ast-unique-expr
Last change on this file since f3e87af was f3e87af, checked in by Thierry Delisle <tdelisle@…>, 9 months 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.