Changeset 3acbf89 for benchmark/io/http/http_ring.cpp

Timestamp:

Jan 19, 2021, 2:40:19 PM (3 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, arm-eh, ast-experimental, enum, forall-pointer-decay, jacob/cs343-translation, master, new-ast-unique-expr, pthread-emulation, qualifiedEnum

Children:

f3e87af

Parents:

efdfdee

Message:

Tentative improvement to batch more requests together.

File:

• benchmark/io/http/http_ring.cpp (modified) (17 diffs)

benchmark/io/http/http_ring.cpp

@@ -16 +16 @@
 } event_t;
 
-struct request_t {
+struct __attribute__((aligned(128))) request_t {
         event_t type;
         int fd;
         size_t length;
-        char buffer[0];
+        char * buff;
+        char data[0];
 
         static struct request_t * create(event_t type, size_t extra) {
@@ -26 +27 @@
                 ret->type = type;
                 ret->length = extra;
+                ret->buff = ret->data;
                 return ret;
         }
@@ -34 +36 @@
 };
 
-struct options_t {
+struct __attribute__((aligned(128))) options_t {
         struct {
                 int sockfd;
@@ -44 +46 @@
         int endfd;
         unsigned entries;
+
+        struct {
+                size_t subs = 0;
+                size_t cnts = 0;
+        } result;
 };
+
+//=========================================================
+static struct io_uring_sqe * get_sqe(struct io_uring * ring) {
+        struct io_uring_sqe * sqe = io_uring_get_sqe(ring);
+        if(!sqe) {
+                std::cerr << "Insufficient entries in ring" << std::endl;
+                exit(EXIT_FAILURE);
+        }
+        return sqe;
+}
+
+static void submit(struct io_uring * ) {
+        // io_uring_submit(ring);
+}
 
 //=========================================================
 static void ring_end(struct io_uring * ring, int fd, char * buffer, size_t len) {
-        struct io_uring_sqe * sqe = io_uring_get_sqe(ring);
+        struct io_uring_sqe * sqe = get_sqe(ring);
         io_uring_prep_read(sqe, fd, buffer, len, 0);
         io_uring_sqe_set_data(sqe, request_t::create(EVENT_END));
-        io_uring_submit(ring);
+        submit(ring);
 }
 
 static void ring_accept(struct io_uring * ring, int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags) {
         auto req = request_t::create(EVENT_ACCEPT);
-        struct io_uring_sqe * sqe = io_uring_get_sqe(ring);
+        struct io_uring_sqe * sqe = get_sqe(ring);
         io_uring_prep_accept(sqe, sockfd, addr, addrlen, flags);
         io_uring_sqe_set_data(sqe, req);
-        io_uring_submit(ring);
-        std::cout << "Submitted accept: " << req << std::endl;
+        submit(ring);
+        // std::cout << "Submitted accept: " << req << std::endl;
 }
 
@@ -68 +89 @@
         req->fd = fd;
 
-        struct io_uring_sqe * sqe = io_uring_get_sqe(ring);
-        io_uring_prep_read(sqe, fd, req->buffer, size, 0);
+        struct io_uring_sqe * sqe = get_sqe(ring);
+        io_uring_prep_read(sqe, fd, req->buff, size, 0);
         io_uring_sqe_set_data(sqe, req);
-        io_uring_submit(ring);
-        std::cout << "Submitted request: " << req << " (" << (void*)req->buffer << ")"<<std::endl;
+        submit(ring);
+        // std::cout << "Submitted request: " << req << " (" << (void*)req->buffer << ")"<<std::endl;
 }
 
@@ -122 +143 @@
         const char * fmt = http_msgs[code];
         const char * date = "";
-        size = snprintf(req->buffer, size, fmt, date, size);
-
-        struct io_uring_sqe * sqe = io_uring_get_sqe(ring);
-        io_uring_prep_write(sqe, fd, req->buffer, size, 0);
+        size = snprintf(req->buff, size, fmt, date, size);
+
+        struct io_uring_sqe * sqe = get_sqe(ring);
+        io_uring_prep_write(sqe, fd, req->buff, size, 0);
         io_uring_sqe_set_data(sqe, req);
-        io_uring_submit(ring);
-        std::cout << "Submitted good answer: " << req << " (" << (void*)req->buffer << ")"<<std::endl;
-}
-
-static void ring_answer(struct io_uring * ring, int fd, const std::string & ans) {
+        submit(ring);
+        // std::cout << "Submitted good answer: " << req << " (" << (void*)req->buffer << ")"<<std::endl;
+}
+
+static void ring_answer(struct io_uring * ring, int fd, const std::string &) {
+        // size_t size = 256;
+        // auto req = request_t::create(EVENT_ANSWER, size);
+        // req->fd = fd;
+
+        // const char * fmt = http_msgs[OK200];
+        // const char * date = "";
+        // size_t len = snprintf(req->buffer, size, fmt, date, ans.size(), ans.c_str());
+        // req->length = len;
+
+        // struct io_uring_sqe * sqe = get_sqe(ring);
+        // io_uring_prep_write(sqe, fd, req->buffer, len, 0);
+        // io_uring_sqe_set_data(sqe, req);
+        // submit(ring);
+        // std::cout << "Submitted good answer: " << req << " (" << (void*)req->buffer << ")"<<std::endl;
+
+
+        static const char* RESPONSE = "HTTP/1.1 200 OK\r\n" \
+                                                "Content-Length: 15\r\n" \
+                                                "Content-Type: text/html\r\n" \
+                                                "Connection: keep-alive\r\n" \
+                                                "Server: testserver\r\n" \
+                                                "\r\n" \
+                                                "Hello, World!\r\n";
+
+        static const size_t RLEN = strlen(RESPONSE);
+
         size_t size = 256;
         auto req = request_t::create(EVENT_ANSWER, size);
         req->fd = fd;
-
-        const char * fmt = http_msgs[OK200];
-        const char * date = "";
-        size_t len = snprintf(req->buffer, size, fmt, date, ans.size(), ans.c_str());
-        req->length = len;
-
-        struct io_uring_sqe * sqe = io_uring_get_sqe(ring);
-        io_uring_prep_write(sqe, fd, req->buffer, len, 0);
+        req->buff = (char*)RESPONSE;
+        req->length = RLEN;
+
+        // const char * fmt = http_msgs[OK200];
+        // const char * date = "";
+        // size_t len = snprintf(req->buffer, size, fmt, date, ans.size(), ans.c_str());
+        // req->length = len;
+
+        struct io_uring_sqe * sqe = get_sqe(ring);
+        io_uring_prep_write(sqe, fd, RESPONSE, RLEN, 0);
         io_uring_sqe_set_data(sqe, req);
-        io_uring_submit(ring);
-        std::cout << "Submitted good answer: " << req << " (" << (void*)req->buffer << ")"<<std::endl;
+        submit(ring);
 }
 
@@ -170 +218 @@
                                 return;
                         default:
-                                std::cerr << "answer error: (" << err << ") " << strerror(err) << std::endl;
+                                std::cerr << "request error: (" << err << ") " << strerror(err) << std::endl;
                                 exit(EXIT_FAILURE);
                 }
@@ -181 +229 @@
         }
 
-        char * it = in->buffer;
+        const char * it = in->buff;
         if( !strstr( it, "\r\n\r\n" ) ) {
                 std::cout << "Incomplete request" << std::endl;
@@ -189 +237 @@
         }
 
-        it = in->buffer;
+        it = in->buff;
         const std::string reply = "Hello, World!\n";
         int ret = memcmp(it, "GET ", 4);
@@ -231 +279 @@
         }
 
-        struct io_uring_sqe * sqe = io_uring_get_sqe(ring);
-        io_uring_prep_write(sqe, in->fd, in->buffer + res, in->length - res, 0);
+        struct io_uring_sqe * sqe = get_sqe(ring);
+        io_uring_prep_write(sqe, in->fd, in->buff + res, in->length - res, 0);
         io_uring_sqe_set_data(sqe, in);
-        io_uring_submit(ring);
-        std::cout << "Re-Submitted request: " << in << " (" << (void*)in->buffer << ")"<<std::endl;
+        submit(ring);
+        // std::cout << "Re-Submitted request: " << in << " (" << (void*)in->buffer << ")"<<std::endl;
 
         ring_request(ring, in->fd);
@@ -241 +289 @@
 
 //=========================================================
+extern "C" {
+extern int __io_uring_flush_sq(struct io_uring *ring);
+}
+
 void * proc_loop(void * arg) {
-        const struct options_t & opt = *(const struct options_t *)arg;
+        size_t count = 0;
+        struct options_t & opt = *(struct options_t *)arg;
 
         struct io_uring ring_storage;
@@ -256 +309 @@
         while(!done) {
                 struct io_uring_cqe *cqe;
-                int ret = io_uring_wait_cqe(ring, &cqe);
-                if (ret < 0) {
-                        fprintf( stderr, "io_uring error: (%d) %s\n", (int)-ret, strerror(-ret) );
+                int ret;
+                while(-EAGAIN == (ret = io_uring_wait_cqe_nr(ring, &cqe, 0))) {
+                        ret = io_uring_submit_and_wait(ring, 1);
+                        if (ret < 0) {
+                                fprintf( stderr, "io_uring get error: (%d) %s\n", (int)-ret, strerror(-ret) );
+                                exit(EXIT_FAILURE);
+                        }
+                        opt.result.subs += ret;
+                        opt.result.cnts++;
+                }
+
+                if (ret < 0 && -EAGAIN != ret) {
+                        fprintf( stderr, "io_uring peek error: (%d) %s\n", (int)-ret, strerror(-ret) );
                         exit(EXIT_FAILURE);
                 }
 
                 auto req = (struct request_t *)cqe->user_data;
-                std::cout << req << " completed with " << cqe->res << std::endl;
+                // std::cout << req << " completed with " << cqe->res << std::endl;
 
                 switch(req->type) {
@@ -287 +350 @@
         io_uring_queue_exit(ring);
 
-        return NULL;
+        return (void*)count;
 }
 
 //=========================================================
+#include <bit>
+
 #include <pthread.h>
 extern "C" {
@@ -299 +364 @@
 }
 
-int main() {
+int main(int argc, char * argv[]) {
         signal(SIGPIPE, SIG_IGN);
 
@@ -306 +371 @@
         unsigned entries = 256;
         unsigned backlog = 10;
+
+        //===================
+        // Arguments
+        int c;
+        while ((c = getopt (argc, argv, "t:p:e:b:")) != -1) {
+                switch (c)
+                {
+                case 't':
+                        nthreads = atoi(optarg);
+                        break;
+                case 'p':
+                        port = atoi(optarg);
+                        break;
+                case 'e':
+                        entries = atoi(optarg);
+                        break;
+                case 'b':
+                        backlog = atoi(optarg);
+                        break;
+                case '?':
+                default:
+                        std::cerr << "Usage: -t <threads> -p <port> -e <entries> -b <backlog>" << std::endl;
+                        return EXIT_FAILURE;
+                }
+        }
+
+        if( !std::ispow2(entries) ) {
+                unsigned v = entries;
+                v--;
+                v |= v >> 1;
+                v |= v >> 2;
+                v |= v >> 4;
+                v |= v >> 8;
+                v |= v >> 16;
+                v++;
+                std::cerr << "Warning: num_entries not a power of 2 (" << entries << ") raising to " << v << std::endl;
+                entries = v;
+        }
 
         //===================
@@ -412 +515 @@
         //===================
         (std::cout << "Stopping Threads Done... ").flush();
+        size_t total = 0;
+        size_t count = 0;
         for(unsigned i = 0; i < nthreads; i++) {
                 void * retval;
@@ -419 +524 @@
                         exit(EXIT_FAILURE);
                 }
+                // total += (size_t)retval;
+                total += thrd_opts[i].result.subs;
+                count += thrd_opts[i].result.cnts;
         }
         std::cout << "done" << std::endl;
+        std::cout << "Submit average: " << total << "/" << count << "(" << (((double)total) / count) << ")" << std::endl;
 
         //===================