Changeset 3f39009 for benchmark

Timestamp:

Oct 29, 2021, 5:21:35 PM (2 years ago)

Author:

Thierry Delisle <tdelisle@…>

Branches:

ADT, ast-experimental, enum, forall-pointer-decay, master, pthread-emulation, qualifiedEnum

Children:

42b7fa5f

Parents:

4087baf

Message:

Reimplemnted how splice is handled.

Location:

benchmark/io/http

Files:

• protocol.cfa (modified) (8 diffs)
• protocol.hfa (modified) (1 diff)
• worker.cfa (modified) (2 diffs)

benchmark/io/http/protocol.cfa

@@ -11 +11 @@
 #include <fstream.hfa>
 #include <iofwd.hfa>
+#include <io/types.hfa>
+#include <mutex_stmt.hfa>
 
 #include <assert.h>
@@ -26 +28 @@
 #define PLAINTEXT_MEMCPY
 #define PLAINTEXT_NOCOPY
+#define LINKED_IO
 
 struct https_msg_str {
@@ -53 +56 @@
 }
 
-static inline int answer( int fd, const char * it, int len) {
+static inline int answer( int fd, const char * it, int len ) {
         while(len > 0) {
                 // Call write
                 int ret = cfa_send(fd, it, len, 0, CFA_IO_LAZY);
                 if( ret < 0 ) {
-                        if( errno == ECONNRESET || errno == EPIPE ) return -ECONNRESET;
+                        if( errno == ECONNRESET || errno == EPIPE ) { close(fd); return -ECONNRESET; }
                         if( errno == EAGAIN || errno == EWOULDBLOCK) return -EAGAIN;
 
@@ -77 +80 @@
 }
 
-int answer_header( int fd, size_t size ) {
-        char buffer[512];
-        char * it = buffer;
+static int fill_header(char * it, size_t size) {
         memcpy(it, http_msgs[OK200]->msg, http_msgs[OK200]->len);
         it += http_msgs[OK200]->len;
         int len = http_msgs[OK200]->len;
         len += snprintf(it, 512 - len, "%d \n\n", size);
+        return len;
+}
+
+static int answer_header( int fd, size_t size ) {
+        char buffer[512];
+        int len = fill_header(buffer, size);
         return answer( fd, buffer, len );
 }
@@ -135 +142 @@
 }
 
-
-[HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len) {
-        char * it = buffer;
-        size_t count = len - 1;
-        int rlen = 0;
-        READ:
-        for() {
-                int ret = cfa_recv(fd, (void*)it, count, 0, CFA_IO_LAZY);
-                // int ret = read(fd, (void*)it, count);
-                if(ret == 0 ) return [OK200, true, 0, 0];
-                if(ret < 0 ) {
-                        if( errno == EAGAIN || errno == EWOULDBLOCK) continue READ;
-                        if( errno == ECONNRESET ) return [E408, true, 0, 0];
-                        if( errno == EPIPE ) return [E408, true, 0, 0];
-                        abort( "read error: (%d) %s\n", (int)errno, strerror(errno) );
-                }
-                it[ret + 1] = '\0';
-                rlen += ret;
-
-                if( strstr( it, "\r\n\r\n" ) ) break;
-
-                it += ret;
-                count -= ret;
-
-                if( count < 1 ) return [E414, false, 0, 0];
-        }
-
-        if( options.log ) {
-                write(sout, buffer, rlen);
-                sout | nl;
-        }
-
-        it = buffer;
-        int ret = memcmp(it, "GET /", 5);
-        if( ret != 0 ) return [E400, false, 0, 0];
-        it += 5;
-
-        char * end = strstr( it, " " );
-        return [OK200, false, it, end - it];
-}
-
-int sendfile( int pipe[2], int fd, int ans_fd, size_t count ) {
+static int sendfile( int pipe[2], int fd, int ans_fd, size_t count ) {
         unsigned sflags = SPLICE_F_MOVE; // | SPLICE_F_MORE;
         off_t offset = 0;
@@ -207 +173 @@
 }
 
+static void zero_sqe(struct io_uring_sqe * sqe) {
+        sqe->flags = 0;
+        sqe->ioprio = 0;
+        sqe->fd = 0;
+        sqe->off = 0;
+        sqe->addr = 0;
+        sqe->len = 0;
+        sqe->fsync_flags = 0;
+        sqe->__pad2[0] = 0;
+        sqe->__pad2[1] = 0;
+        sqe->__pad2[2] = 0;
+        sqe->fd = 0;
+        sqe->off = 0;
+        sqe->addr = 0;
+        sqe->len = 0;
+}
+
+enum FSM_STATE {
+        Initial,
+        Retry,
+        Error,
+        Done,
+};
+
+struct FSM_Result {
+        FSM_STATE state;
+        int error;
+};
+
+static inline void ?{}(FSM_Result & this) { this.state = Initial; this.error = 0; }
+static inline bool is_error(FSM_Result & this) { return Error == this.state; }
+static inline bool is_done(FSM_Result & this) { return Done == this.state; }
+
+static inline int error(FSM_Result & this, int error) {
+        this.error = error;
+        this.state = Error;
+        return error;
+}
+
+static inline int done(FSM_Result & this) {
+        this.state = Done;
+        return 0;
+}
+
+static inline int retry(FSM_Result & this) {
+        this.state = Retry;
+        return 0;
+}
+
+static inline int need(FSM_Result & this) {
+        switch(this.state) {
+                case Initial:
+                case Retry:
+                        return 1;
+                case Error:
+                        if(this.error == 0) mutex(serr) serr | "State marked error but code is 0";
+                case Done:
+                        return 0;
+        }
+}
+
+// Generator that handles sending the header
+generator header_g {
+        io_future_t f;
+        const char * next;
+        int fd; size_t len;
+        FSM_Result res;
+};
+
+static inline void ?{}(header_g & this, int fd, const char * it, size_t len ) {
+        this.next = it;
+        this.fd = fd;
+        this.len = len;
+}
+
+static inline void fill(header_g & this, struct io_uring_sqe * sqe) {
+        zero_sqe(sqe);
+        sqe->opcode = IORING_OP_SEND;
+        sqe->user_data = (uintptr_t)&this.f;
+        sqe->flags = IOSQE_IO_LINK;
+        sqe->fd = this.fd;
+        sqe->addr = (uintptr_t)this.next;
+        sqe->len = this.len;
+}
+
+static inline int error(header_g & this, int error) {
+        int ret = close(this.fd);
+        if( ret != 0 ) {
+                mutex(serr) serr | "Failed to close fd" | errno;
+        }
+        return error(this.res, error);
+}
+
+static inline int wait_and_process(header_g & this) {
+        wait(this.f);
+
+        // Did something crazy happen?
+        if(this.f.result > this.len) {
+                mutex(serr) serr | "HEADER sent too much!";
+                return error(this, -ERANGE);
+        }
+
+        // Something failed?
+        if(this.f.result < 0) {
+                int error = -this.f.result;
+                if( error == ECONNRESET ) return error(this, -ECONNRESET);
+                if( error == EPIPE ) return error(this, -EPIPE);
+                if( error == ECANCELED ) {
+                        mutex(serr) serr | "HEADER was cancelled, WTF!";
+                        return error(this, -ECONNRESET);
+                }
+                if( error == EAGAIN || error == EWOULDBLOCK) {
+                        mutex(serr) serr | "HEADER got eagain, WTF!";
+                        return error(this, -ECONNRESET);
+                }
+        }
+
+        // Done?
+        if(this.f.result == this.len) {
+                return done(this.res);
+        }
+
+        // It must be a Short read
+        this.len  -= this.f.result;
+        this.next += this.f.result;
+        reset(this.f);
+        return retry(this.res);
+}
+
+// Generator that handles splicing in a file
+struct splice_in_t {
+        io_future_t f;
+        int fd; int pipe; size_t len; off_t off;
+        FSM_Result res;
+};
+
+static inline void ?{}(splice_in_t & this, int fd, int pipe, size_t len) {
+        this.fd = fd;
+        this.pipe = pipe;
+        this.len = len;
+        this.off = 0;
+}
+
+static inline void fill(splice_in_t & this, struct io_uring_sqe * sqe) {
+        zero_sqe(sqe);
+        sqe->opcode = IORING_OP_SPLICE;
+        sqe->user_data = (uintptr_t)&this.f;
+        sqe->flags = 0;
+        sqe->splice_fd_in = this.fd;
+        sqe->splice_off_in = this.off;
+        sqe->fd = this.pipe;
+        sqe->off = (__u64)-1;
+        sqe->len = this.len;
+        sqe->splice_flags = SPLICE_F_MOVE;
+}
+
+static inline int wait_and_process(splice_in_t & this) {
+        wait(this.f);
+
+        // Did something crazy happen?
+        if(this.f.result > this.len) {
+                mutex(serr) serr | "SPLICE IN spliced too much!";
+                return error(this.res, -ERANGE);
+        }
+
+        // Something failed?
+        if(this.f.result < 0) {
+                int error = -this.f.result;
+                if( error == ECONNRESET ) return error(this.res, -ECONNRESET);
+                if( error == EPIPE ) return error(this.res, -EPIPE);
+                if( error == ECANCELED ) {
+                        mutex(serr) serr | "SPLICE IN was cancelled, WTF!";
+                        return error(this.res, -ECONNRESET);
+                }
+                if( error == EAGAIN || error == EWOULDBLOCK) {
+                        mutex(serr) serr | "SPLICE IN got eagain, WTF!";
+                        return error(this.res, -ECONNRESET);
+                }
+        }
+
+        // Done?
+        if(this.f.result == this.len) {
+                return done(this.res);
+        }
+
+        // It must be a Short read
+        this.len -= this.f.result;
+        this.off += this.f.result;
+        reset(this.f);
+        return retry(this.res);
+}
+
+generator splice_out_g {
+        io_future_t f;
+        int pipe; int fd; size_t len;
+        FSM_Result res;
+};
+
+static inline void ?{}(splice_out_g & this, int pipe, int fd, size_t len) {
+        this.pipe = pipe;
+        this.fd = fd;
+        this.len = len;
+}
+
+static inline void fill(splice_out_g & this, struct io_uring_sqe * sqe) {
+        zero_sqe(sqe);
+        sqe->opcode = IORING_OP_SPLICE;
+        sqe->user_data = (uintptr_t)&this.f;
+        sqe->flags = 0;
+        sqe->splice_fd_in = this.pipe;
+        sqe->splice_off_in = (__u64)-1;
+        sqe->fd = this.fd;
+        sqe->off = (__u64)-1;
+        sqe->len = this.len;
+        sqe->splice_flags = SPLICE_F_MOVE;
+}
+
+static inline int error(splice_out_g & this, int error) {
+        int ret = close(this.fd);
+        if( ret != 0 ) {
+                mutex(serr) serr | "Failed to close fd" | errno;
+        }
+        return error(this.res, error);
+}
+
+static inline void wait_and_process(splice_out_g & this) {
+        wait(this.f);
+
+        // Did something crazy happen?
+        if(this.f.result > this.len) {
+                mutex(serr) serr | "SPLICE OUT spliced too much!";
+                return error(this.res, -ERANGE);
+        }
+
+        // Something failed?
+        if(this.f.result < 0) {
+                int error = -this.f.result;
+                if( error == ECONNRESET ) return error(this, -ECONNRESET);
+                if( error == EPIPE ) return error(this, -EPIPE);
+                if( error == ECANCELED ) {
+                        this.f.result = 0;
+                        goto SHORT_WRITE;
+                }
+                if( error == EAGAIN || error == EWOULDBLOCK) {
+                        mutex(serr) serr | "SPLICE OUT got eagain, WTF!";
+                        return error(this, -ECONNRESET);
+                }
+        }
+
+        // Done?
+        if(this.f.result == this.len) {
+                return done(this.res);
+        }
+
+SHORT_WRITE:
+        // It must be a Short Write
+        this.len -= this.f.result;
+        reset(this.f);
+        return retry(this.res);
+}
+
+int answer_sendfile( int pipe[2], int fd, int ans_fd, size_t fsize ) {
+        #if defined(LINKED_IO)
+                char buffer[512];
+                int len = fill_header(buffer, fsize);
+                header_g header = { fd, buffer, len };
+                splice_in_t splice_in = { ans_fd, pipe[1], fsize };
+                splice_out_g splice_out = { pipe[0], fd, fsize };
+
+                RETRY_LOOP: for() {
+                        int have = need(header.res) + need(splice_in.res) + 1;
+                        int idx = 0;
+                        struct io_uring_sqe * sqes[3];
+                        __u32 idxs[3];
+                        struct $io_context * ctx = cfa_io_allocate(sqes, idxs, have);
+
+                        if(need(splice_in.res)) { fill(splice_in, sqes[idx++]); }
+                        if(need(   header.res)) { fill(header   , sqes[idx++]); }
+                        fill(splice_out, sqes[idx]);
+
+                        // Submit everything
+                        asm volatile("": : :"memory");
+                        cfa_io_submit( ctx, idxs, have, false );
+
+                        // wait for the results
+                        // Always wait for splice-in to complete as
+                        // we may need to kill the connection if it fails
+                        // If it already completed, this is a no-op
+                        wait_and_process(splice_in);
+
+                        if(is_error(splice_in.res)) {
+                                mutex(serr) serr | "SPLICE IN failed with" | splice_in.res.error;
+                                close(fd);
+                        }
+
+                        // Process the other 2
+                        wait_and_process(header);
+                        wait_and_process(splice_out);
+
+                        if(is_done(splice_out.res)) {
+                                break RETRY_LOOP;
+                        }
+
+                        // We need to wait for the completion if
+                        // - both completed
+                        // - the header failed
+                        // -
+
+                        if(  is_error(header.res)
+                          || is_error(splice_in.res)
+                          || is_error(splice_out.res)) {
+                                return -ECONNRESET;
+                        }
+                }
+
+                return len + fsize;
+        #else
+                int ret = answer_header(fd, fsize);
+                if( ret < 0 ) return ret;
+                return sendfile(pipe, fd, ans_fd, fsize);
+        #endif
+}
+
+[HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len) {
+        char * it = buffer;
+        size_t count = len - 1;
+        int rlen = 0;
+        READ:
+        for() {
+                int ret = cfa_recv(fd, (void*)it, count, 0, CFA_IO_LAZY);
+                // int ret = read(fd, (void*)it, count);
+                if(ret == 0 ) return [OK200, true, 0, 0];
+                if(ret < 0 ) {
+                        if( errno == EAGAIN || errno == EWOULDBLOCK) continue READ;
+                        if( errno == ECONNRESET ) { close(fd); return [E408, true, 0, 0]; }
+                        if( errno == EPIPE ) { close(fd); return [E408, true, 0, 0]; }
+                        abort( "read error: (%d) %s\n", (int)errno, strerror(errno) );
+                }
+                it[ret + 1] = '\0';
+                rlen += ret;
+
+                if( strstr( it, "\r\n\r\n" ) ) break;
+
+                it += ret;
+                count -= ret;
+
+                if( count < 1 ) return [E414, false, 0, 0];
+        }
+
+        if( options.log ) {
+                write(sout, buffer, rlen);
+                sout | nl;
+        }
+
+        it = buffer;
+        int ret = memcmp(it, "GET /", 5);
+        if( ret != 0 ) return [E400, false, 0, 0];
+        it += 5;
+
+        char * end = strstr( it, " " );
+        return [OK200, false, it, end - it];
+}
+
 //=============================================================================================
 
@@ -251 +580 @@
                 Time now = timeHiRes();
                 strftime( buff, 100, "%a, %d %b %Y %H:%M:%S %Z", now );
-                if( options.log ) sout | "Updated date to '" | buff | "'";
+                // if( options.log ) sout | "Updated date to '" | buff | "'";
 
                 for(i; KNOWN_CODES) {
@@ -264 +593 @@
                 this.idx = (this.idx + 1) % 2;
 
-                if( options.log ) sout | "Date thread sleeping";
+                // if( options.log ) sout | "Date thread sleeping";
 
                 sleep(1`s);

benchmark/io/http/protocol.hfa

@@ -16 +16 @@
 
 int answer_error( int fd, HttpCode code );
-int answer_header( int fd, size_t size );
 int answer_plaintext( int fd );
 int answer_empty( int fd );
+int answer_sendfile( int pipe[2], int fd, int ans_fd, size_t count );
 
 [HttpCode code, bool closed, * const char file, size_t len] http_read(int fd, []char buffer, size_t len);
-
-int sendfile( int pipe[2], int fd, int ans_fd, size_t count );

benchmark/io/http/worker.cfa

@@ -122 +122 @@
                         }
 
-                        // Send the header
-                        int ret = answer_header(fd, count);
-                        if( ret == -ECONNRESET ) break REQUEST;
-
                         // Send the desired file
-                        ret = sendfile( this.pipe, fd, ans_fd, count);
+                        int ret = answer_sendfile( this.pipe, fd, ans_fd, count);
                         if( ret == -ECONNRESET ) break REQUEST;
 
@@ -134 +130 @@
 
                 if( options.log ) sout | "=== Connection closed ===";
-                close(fd);
                 continue CONNECTION;
         }