source: benchmark/io/http/http_ring.cpp @ 3acbf89

ADTarm-ehast-experimentalenumforall-pointer-decayjacob/cs343-translationnew-ast-unique-exprpthread-emulationqualifiedEnum
Last change on this file since 3acbf89 was 3acbf89, checked in by Thierry Delisle <tdelisle@…>, 3 years ago

Tentative improvement to batch more requests together.

  • Property mode set to 100644
File size: 13.7 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        unsigned entries;
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_storage;
300        struct io_uring * ring = &ring_storage;
301        io_uring_queue_init(opt.entries, ring, 0);
302
303        char endfd_buf[8];
304        ring_end(ring, opt.endfd, endfd_buf, 8);
305
306        ring_accept(ring, opt.acpt.sockfd, opt.acpt.addr, opt.acpt.addrlen, opt.acpt.flags);
307
308        bool done = false;
309        while(!done) {
310                struct io_uring_cqe *cqe;
311                int ret;
312                while(-EAGAIN == (ret = io_uring_wait_cqe_nr(ring, &cqe, 0))) {
313                        ret = io_uring_submit_and_wait(ring, 1);
314                        if (ret < 0) {
315                                fprintf( stderr, "io_uring get error: (%d) %s\n", (int)-ret, strerror(-ret) );
316                                exit(EXIT_FAILURE);
317                        }
318                        opt.result.subs += ret;
319                        opt.result.cnts++;
320                }
321
322                if (ret < 0 && -EAGAIN != ret) {
323                        fprintf( stderr, "io_uring peek error: (%d) %s\n", (int)-ret, strerror(-ret) );
324                        exit(EXIT_FAILURE);
325                }
326
327                auto req = (struct request_t *)cqe->user_data;
328                // std::cout << req << " completed with " << cqe->res << std::endl;
329
330                switch(req->type) {
331                        case EVENT_END:
332                                done = true;
333                                break;
334                        case EVENT_ACCEPT:
335                                handle_new_conn(ring, cqe->res);
336                                free(req);
337                                ring_accept(ring, opt.acpt.sockfd, opt.acpt.addr, opt.acpt.addrlen, opt.acpt.flags);
338                                break;
339                        case EVENT_REQUEST:
340                                handle_request(ring, req, cqe->res);
341                                break;
342                        case EVENT_ANSWER:
343                                handle_answer(ring, req, cqe->res);
344                                break;
345                }
346
347                io_uring_cqe_seen(ring, cqe);
348        }
349
350        io_uring_queue_exit(ring);
351
352        return (void*)count;
353}
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
374        //===================
375        // Arguments
376        int c;
377        while ((c = getopt (argc, argv, "t:p:e:b:")) != -1) {
378                switch (c)
379                {
380                case 't':
381                        nthreads = atoi(optarg);
382                        break;
383                case 'p':
384                        port = atoi(optarg);
385                        break;
386                case 'e':
387                        entries = atoi(optarg);
388                        break;
389                case 'b':
390                        backlog = atoi(optarg);
391                        break;
392                case '?':
393                default:
394                        std::cerr << "Usage: -t <threads> -p <port> -e <entries> -b <backlog>" << std::endl;
395                        return EXIT_FAILURE;
396                }
397        }
398
399        if( !std::ispow2(entries) ) {
400                unsigned v = entries;
401                v--;
402                v |= v >> 1;
403                v |= v >> 2;
404                v |= v >> 4;
405                v |= v >> 8;
406                v |= v >> 16;
407                v++;
408                std::cerr << "Warning: num_entries not a power of 2 (" << entries << ") raising to " << v << std::endl;
409                entries = v;
410        }
411
412        //===================
413        // End FD
414        int efd = eventfd(0, EFD_SEMAPHORE);
415        if (efd < 0) {
416                std::cerr << "eventfd error: (" << errno << ") " << strerror(errno) << std::endl;
417                exit(EXIT_FAILURE);
418        }
419
420        //===================
421        // Open Socket
422        std::cout << getpid() << " : Listening on port " << port << std::endl;
423        int server_fd = socket(AF_INET, SOCK_STREAM, 0);
424        if(server_fd < 0) {
425                std::cerr << "socket error: (" << errno << ") " << strerror(errno) << std::endl;
426                exit(EXIT_FAILURE);
427        }
428
429        int ret = 0;
430        struct sockaddr_in address;
431        int addrlen = sizeof(address);
432        memset( (char *)&address, '\0', addrlen );
433        address.sin_family = AF_INET;
434        address.sin_addr.s_addr = htonl(INADDR_ANY);
435        address.sin_port = htons( port );
436
437        int waited = 0;
438        while(true) {
439                ret = bind( server_fd, (struct sockaddr *)&address, sizeof(address) );
440                if(ret < 0) {
441                        if(errno == EADDRINUSE) {
442                                if(waited == 0) {
443                                        std::cerr << "Waiting for port" << std::endl;
444                                } else {
445                                        std::cerr << "\r" << waited;
446                                        std::cerr.flush();
447                                }
448                                waited ++;
449                                usleep( 1000000 );
450                                continue;
451                        }
452                        std::cerr << "bind error: (" << errno << ") " << strerror(errno) << std::endl;
453                        exit(EXIT_FAILURE);
454                }
455                break;
456        }
457
458        ret = listen( server_fd, backlog );
459        if(ret < 0) {
460                std::cerr << "listen error: (" << errno << ") " << strerror(errno) << std::endl;
461                exit(EXIT_FAILURE);
462        }
463
464        //===================
465        // Run Server Threads
466        std::cout << "Starting " << nthreads << " Threads" << std::endl;
467        pthread_t thrd_hdls[nthreads];
468        options_t thrd_opts[nthreads];
469        for(unsigned i = 0; i < nthreads; i++) {
470                thrd_opts[i].acpt.sockfd  = server_fd;
471                thrd_opts[i].acpt.addr    = (struct sockaddr *)&address;
472                thrd_opts[i].acpt.addrlen = (socklen_t*)&addrlen;
473                thrd_opts[i].acpt.flags   = 0;
474                thrd_opts[i].endfd   = efd;
475                thrd_opts[i].entries = entries;
476
477                int ret = pthread_create(&thrd_hdls[i], nullptr, proc_loop, &thrd_opts[i]);
478                if (ret < 0) {
479                        std::cerr << "pthread create error: (" << errno << ") " << strerror(errno) << std::endl;
480                        exit(EXIT_FAILURE);
481                }
482        }
483
484        //===================
485        // Server Started
486        std::cout << "Server Started" << std::endl;
487        {
488                char buffer[128];
489                int ret;
490                do {
491                        ret = read(STDIN_FILENO, buffer, 128);
492                        if(ret < 0) {
493                                std::cerr << "main read error: (" << errno << ") " << strerror(errno) << std::endl;
494                                exit(EXIT_FAILURE);
495                        }
496                        else if(ret > 0) {
497                                std::cout << "User inputed '";
498                                std::cout.write(buffer, ret);
499                                std::cout << "'" << std::endl;
500                        }
501                } while(ret != 0);
502
503                std::cout << "Shutdown received" << std::endl;
504        }
505
506        //===================
507        (std::cout << "Sending Shutdown to Threads... ").flush();
508        ret = eventfd_write(efd, nthreads);
509        if (ret < 0) {
510                std::cerr << "eventfd close error: (" << errno << ") " << strerror(errno) << std::endl;
511                exit(EXIT_FAILURE);
512        }
513        std::cout << "done" << std::endl;
514
515        //===================
516        (std::cout << "Stopping Threads Done... ").flush();
517        size_t total = 0;
518        size_t count = 0;
519        for(unsigned i = 0; i < nthreads; i++) {
520                void * retval;
521                int ret = pthread_join(thrd_hdls[i], &retval);
522                if (ret < 0) {
523                        std::cerr << "pthread create error: (" << errno << ") " << strerror(errno) << std::endl;
524                        exit(EXIT_FAILURE);
525                }
526                // total += (size_t)retval;
527                total += thrd_opts[i].result.subs;
528                count += thrd_opts[i].result.cnts;
529        }
530        std::cout << "done" << std::endl;
531        std::cout << "Submit average: " << total << "/" << count << "(" << (((double)total) / count) << ")" << std::endl;
532
533        //===================
534        (std::cout << "Closing Socket... ").flush();
535        ret = shutdown( server_fd, SHUT_RD );
536        if( ret < 0 ) {
537                std::cerr << "shutdown socket error: (" << errno << ") " << strerror(errno) << std::endl;
538                exit(EXIT_FAILURE);
539        }
540
541        ret = close(server_fd);
542        if (ret < 0) {
543                std::cerr << "close socket error: (" << errno << ") " << strerror(errno) << std::endl;
544                exit(EXIT_FAILURE);
545        }
546        std::cout << "done" << std::endl;
547}
Note: See TracBrowser for help on using the repository browser.