Changeset 5407cdc for benchmark/io/http/http_ring.cpp

Timestamp:

Apr 28, 2021, 4:56:50 PM (5 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:

8d66610

Parents:

feacef9 (diff), b7fd2db6 (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Message:

Merge branch 'master' of plg.uwaterloo.ca:software/cfa/cfa-cc

File:

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

benchmark/io/http/http_ring.cpp

@@ -20 +20 @@
                 socklen_t *addrlen;
                 int flags;
+                unsigned cnt;
         } acpt;
 
@@ -67 +68 @@
 thread_local stats_block_t stats;
 stats_block_t global_stats;
+
+thread_local struct __attribute__((aligned(128))) {
+        size_t to_submit = 0;
+} local;
 
 // Get an array of current connections
@@ -192 +197 @@
         static void submit(struct io_uring * ring, struct io_uring_sqe * sqe, connection * conn) {
                 (void)ring;
+                local.to_submit++;
                 #ifdef USE_ASYNC
                         io_uring_sqe_set_flags(sqe, IOSQE_ASYNC);
@@ -406 +412 @@
                 switch(state) {
                 case ACCEPTING:
-                        connection::accept(ring, opt);
+                        // connection::accept(ring, opt);
                         newconn(ring, res);
                         break;
@@ -420 +426 @@
 
 //=========================================================
+extern "C" {
+        #include <sys/eventfd.h>  // use for termination
+}
+
 // Main loop of the WebServer
 // Effectively uses one thread_local copy of everything per kernel thread
@@ -427 +437 @@
         struct io_uring * ring = opt.ring;
 
+        int blockfd = eventfd(0, 0);
+        if (blockfd < 0) {
+                fprintf( stderr, "eventfd create error: (%d) %s\n", (int)errno, strerror(errno) );
+                exit(EXIT_FAILURE);
+        }
+
+        int ret = io_uring_register_eventfd(ring, blockfd);
+        if (ret < 0) {
+                fprintf( stderr, "io_uring S&W error: (%d) %s\n", (int)-ret, strerror(-ret) );
+                exit(EXIT_FAILURE);
+        }
+
         // Track the shutdown using a event_fd
         char endfd_buf[8];
@@ -433 +455 @@
         // Accept our first connection
         // May not take effect until io_uring_submit_and_wait
-        connection::accept(ring, opt);
+        for(unsigned i = 0; i < opt.acpt.cnt; i++) {
+                connection::accept(ring, opt);
+        }
 
         int reset = 1;       // Counter to print stats once in a while
@@ -441 +465 @@
         while(!done) {
                 // Submit all the answers we have and wait for responses
-                int ret = io_uring_submit_and_wait(ring, 1);
+                int ret = io_uring_submit(ring);
+                local.to_submit = 0;
 
                 // check errors
@@ -452 +477 @@
                 sqes += ret;
                 call++;
+
+
+                eventfd_t val;
+                ret = eventfd_read(blockfd, &val);
+
+                // check errors
+                if (ret < 0) {
+                        fprintf( stderr, "eventfd read error: (%d) %s\n", (int)errno, strerror(errno) );
+                        exit(EXIT_FAILURE);
+                }
 
                 struct io_uring_cqe *cqe;
@@ -463 +498 @@
                                 break;
                         }
+
+                        if(local.to_submit > 30) break;
 
                         auto req = (class connection *)cqe->user_data;
@@ -509 +546 @@
         #include <pthread.h>      // for pthreads
         #include <signal.h>       // for signal(SIGPIPE, SIG_IGN);
-        #include <sys/eventfd.h>  // use for termination
         #include <sys/socket.h>   // for sockets in general
         #include <netinet/in.h>   // for sockaddr_in, AF_INET
@@ -528 +564 @@
         unsigned entries = 256;     // number of entries per ring/kernel thread
         unsigned backlog = 262144;  // backlog argument to listen
+        unsigned preaccept = 1;     // start by accepting X per threads
         bool attach = false;        // Whether or not to attach all the rings
         bool sqpoll = false;        // Whether or not to use SQ Polling
@@ -534 +571 @@
         // Arguments Parsing
         int c;
-        while ((c = getopt (argc, argv, "t:p:e:b:aS")) != -1) {
+        while ((c = getopt (argc, argv, "t:p:e:b:c:aS")) != -1) {
                 switch (c)
                 {
@@ -548 +585 @@
                 case 'b':
                         backlog = atoi(optarg);
+                        break;
+                case 'c':
+                        preaccept = atoi(optarg);
                         break;
                 case 'a':
@@ -681 +721 @@
                 thrd_opts[i].acpt.addrlen = (socklen_t*)&addrlen;
                 thrd_opts[i].acpt.flags   = 0;
+                thrd_opts[i].acpt.cnt     = preaccept;
                 thrd_opts[i].endfd        = efd;
                 thrd_opts[i].ring         = &thrd_rings[i].storage;