Index: benchmark/io/http/protocol.cfa =================================================================== --- benchmark/io/http/protocol.cfa (revision 4087baf4c14f5e44023d455a54e679b6e8c2ad42) +++ benchmark/io/http/protocol.cfa (revision 3f39009bb2c72295a552596d5e31d2d2d71bf9ea) @@ -11,4 +11,6 @@ #include #include +#include +#include #include @@ -26,4 +28,5 @@ #define PLAINTEXT_MEMCPY #define PLAINTEXT_NOCOPY +#define LINKED_IO struct https_msg_str { @@ -53,10 +56,10 @@ } -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,11 +80,15 @@ } -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,46 +142,5 @@ } - -[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,4 +173,367 @@ } +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,5 +580,5 @@ 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,5 +593,5 @@ this.idx = (this.idx + 1) % 2; - if( options.log ) sout | "Date thread sleeping"; + // if( options.log ) sout | "Date thread sleeping"; sleep(1`s); Index: benchmark/io/http/protocol.hfa =================================================================== --- benchmark/io/http/protocol.hfa (revision 4087baf4c14f5e44023d455a54e679b6e8c2ad42) +++ benchmark/io/http/protocol.hfa (revision 3f39009bb2c72295a552596d5e31d2d2d71bf9ea) @@ -16,9 +16,7 @@ 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 ); Index: benchmark/io/http/worker.cfa =================================================================== --- benchmark/io/http/worker.cfa (revision 4087baf4c14f5e44023d455a54e679b6e8c2ad42) +++ benchmark/io/http/worker.cfa (revision 3f39009bb2c72295a552596d5e31d2d2d71bf9ea) @@ -122,10 +122,6 @@ } - // 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,5 +130,4 @@ if( options.log ) sout | "=== Connection closed ==="; - close(fd); continue CONNECTION; }