Changes in / [8e48fca4:e58e423]

Files:

• benchmark/io/http/filecache.cfa (modified) (1 diff)
• benchmark/io/http/http_ring.cpp (modified) (1 diff)
• benchmark/io/http/main.cfa (modified) (1 diff)
• benchmark/io/http/options.cfa (modified) (5 diffs)
• benchmark/io/http/options.hfa (modified) (1 diff)
• benchmark/io/http/protocol.cfa (modified) (9 diffs)
• benchmark/io/http/protocol.hfa (modified) (1 diff)
• benchmark/io/http/worker.cfa (modified) (2 diffs)
• libcfa/src/concurrency/io.cfa (modified) (2 diffs)
• libcfa/src/concurrency/io/types.hfa (modified) (1 diff)
• libcfa/src/concurrency/kernel.cfa (modified) (6 diffs)

benchmark/io/http/filecache.cfa

@@ -185 +185 @@
         sout | "Filled cache from path \"" | path | "\" with" | fcount | "files";
         if( conflicts > 0 ) {
-                sout | "Found" | conflicts | "conflicts (seed: " | options.file_cache.hash_seed | ")";
+                sout | "Found" | conflicts | "conflicts (size: " | file_cache.size | ", seed: " | options.file_cache.hash_seed | ")";
                 #if defined(REJECT_CONFLICTS)
                         abort("Conflicts found in the cache");

benchmark/io/http/http_ring.cpp

@@ -118 +118 @@
 // Get a fix reply based on the return code
 const char * http_msgs[] = {
-        "HTTP/1.1 200 OK\r\nServer: HttoForall\r\nContent-Type: text/plain\r\nContent-Length: 15\r\nConnection: keep-alive\r\n\r\nHello, World!\r\n",
-        "HTTP/1.1 400 Bad Request\r\nServer: HttoForall\r\nContent-Type: text/plain\r\nContent-Length: 0 \r\n\r\n",
-        "HTTP/1.1 404 Not Found\r\nServer: HttoForall\r\nContent-Type: text/plain\r\nContent-Length: 0 \r\n\r\n",
-        "HTTP/1.1 405 Method Not \r\nServer: HttoForall\r\nContent-Type: text/plain\r\nContent-Length: 0 \r\n\r\n",
-        "HTTP/1.1 408 Request Timeout\r\nServer: HttoForall\r\nContent-Type: text/plain\r\nContent-Length: 0 \r\n\r\n",
-        "HTTP/1.1 413 Payload Too Large\r\nServer: HttoForall\r\nContent-Type: text/plain\r\nContent-Length: 0 \r\n\r\n",
-        "HTTP/1.1 414 URI Too Long\r\nServer: HttoForall\r\nContent-Type: text/plain\r\nContent-Length: 0 \r\n\r\n",
+        "HTTP/1.1 200 OK\r\nServer: HttpForall\r\nContent-Type: text/plain\r\nContent-Length: 15\r\nConnection: keep-alive\r\n\r\nHello, World!\r\n",
+        "HTTP/1.1 400 Bad Request\r\nServer: HttpForall\r\nContent-Type: text/plain\r\nContent-Length: 0 \r\n\r\n",
+        "HTTP/1.1 404 Not Found\r\nServer: HttpForall\r\nContent-Type: text/plain\r\nContent-Length: 0 \r\n\r\n",
+        "HTTP/1.1 405 Method Not \r\nServer: HttpForall\r\nContent-Type: text/plain\r\nContent-Length: 0 \r\n\r\n",
+        "HTTP/1.1 408 Request Timeout\r\nServer: HttpForall\r\nContent-Type: text/plain\r\nContent-Length: 0 \r\n\r\n",
+        "HTTP/1.1 413 Payload Too Large\r\nServer: HttpForall\r\nContent-Type: text/plain\r\nContent-Length: 0 \r\n\r\n",
+        "HTTP/1.1 414 URI Too Long\r\nServer: HttpForall\r\nContent-Type: text/plain\r\nContent-Length: 0 \r\n\r\n",
 };
 static_assert( KNOWN_CODES == (sizeof(http_msgs) / sizeof(http_msgs[0])) );

benchmark/io/http/main.cfa

@@ -212 +212 @@
                                 }
                                 sout | nl;
+                                if(!options.interactive) park();
                                 {
                                         char buffer[128];

benchmark/io/http/options.cfa

@@ -21 +21 @@
         false, // log
         false, // stats
+        true, // interactive
+        0, // redirect
+        0, // redirect
 
         { // file_cache
@@ -52 +55 @@
         // bool sqkpoll = false;
         // bool iokpoll = false;
-        unsigned nentries = 16;
+        unsigned nentries = 0;
         bool isolate = false;
 
@@ -62 +65 @@
                 {'\0', "isolate",        "Create one cluster per processor", isolate, parse_settrue},
                 {'\0', "log",            "Enable logs", options.log, parse_settrue},
+                {'\0', "sout",           "Redirect standard out to file", options.reopen_stdout},
+                {'\0', "serr",           "Redirect standard error to file", options.reopen_stderr},
                 {'\0', "stats",          "Enable statistics", options.stats, parse_settrue},
+                {'\0', "shell",          "Disable interactive mode", options.interactive, parse_setfalse},
                 {'\0', "accept-backlog", "Maximum number of pending accepts", options.socket.backlog},
                 {'\0', "request_len",    "Maximum number of bytes in the http request, requests with more data will be answered with Http Code 414", options.socket.buflen},
@@ -79 +85 @@
         parse_args( argc, argv, opt, opt_cnt, "[OPTIONS]... [PATH]\ncforall http server", left );
 
-        if( !is_pow2(nentries) ) {
+        if( nentries != 0 && !is_pow2(nentries) ) {
                 unsigned v = nentries;
                 v--;
@@ -131 +137 @@
 
         options.file_cache.path = path;
+
+        if( options.reopen_stdout && options.reopen_stderr && 0 == strcmp(options.reopen_stdout, options.reopen_stderr) ) {
+                serr | "Redirect sout and serr to the same file is not supported";
+                exit(EXIT_FAILURE);
+        }
+
+        if( options.reopen_stdout ) {
+                sout | "redirecting sout to '" | options.reopen_stdout | "'";
+                FILE  * ret = freopen( options.reopen_stdout, "w", stdout);
+                if( ret == 0p ) {
+                        serr | "Failed to redirect sout to '" | options.reopen_stdout | "'";
+                        exit(EXIT_FAILURE);
+                }
+        }
+
+        if( options.reopen_stderr ) {
+                sout | "redirecting serr to '" | options.reopen_stderr | "'";
+                FILE  * ret = freopen( options.reopen_stderr, "w", stderr);
+                if( ret == 0p ) {
+                        serr | "Failed to redirect serr to '" | options.reopen_stderr | "'";
+                        exit(EXIT_FAILURE);
+                }
+        }
 }

benchmark/io/http/options.hfa

@@ -10 +10 @@
         bool log;
         bool stats;
+        bool interactive;
+        const char * reopen_stdout;
+        const char * reopen_stderr;
 
         struct {

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 ) { close(fd); 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];
+}
+
 //=============================================================================================
 
@@ -214 +543 @@
 
 const char * original_http_msgs[] = {
-        "HTTP/1.1 200 OK\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: ",
-        "HTTP/1.1 200 OK\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 15\n\nHello, World!\n\n",
-        "HTTP/1.1 400 Bad Request\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
-        "HTTP/1.1 404 Not Found\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
-        "HTTP/1.1 405 Method Not Allowed\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
-        "HTTP/1.1 408 Request Timeout\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
-        "HTTP/1.1 413 Payload Too Large\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
-        "HTTP/1.1 414 URI Too Long\nServer: HttoForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
+        "HTTP/1.1 200 OK\nServer: HttpForall\nDate: %s \nContent-Type: text/plain\nContent-Length: ",
+        "HTTP/1.1 200 OK\nServer: HttpForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 15\n\nHello, World!\n\n",
+        "HTTP/1.1 400 Bad Request\nServer: HttpForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
+        "HTTP/1.1 404 Not Found\nServer: HttpForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
+        "HTTP/1.1 405 Method Not Allowed\nServer: HttpForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
+        "HTTP/1.1 408 Request Timeout\nServer: HttpForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
+        "HTTP/1.1 413 Payload Too Large\nServer: HttpForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
+        "HTTP/1.1 414 URI Too Long\nServer: HttpForall\nDate: %s \nContent-Type: text/plain\nContent-Length: 0 \n\n",
 };
 
@@ -251 +580 @@
                 Time now = timeHiRes();
                 strftime( buff, 100, "%a, %d %b %Y %H:%M:%S %Z", now );
-                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;
 
-                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;
         }

libcfa/src/concurrency/io.cfa

@@ -183 +183 @@
                 ctx.proc->io.pending = false;
 
-                ready_schedule_lock();
                 __cfa_io_drain( proc );
-                ready_schedule_unlock();
                 // for(i; 2) {
                 //      unsigned idx = proc->rdq.id + i;
@@ -311 +309 @@
                 // Make the sqes visible to the submitter
                 __atomic_store_n(sq.kring.tail, tail + have, __ATOMIC_RELEASE);
-                sq.to_submit++;
+                sq.to_submit += have;
 
                 ctx->proc->io.pending = true;
                 ctx->proc->io.dirty   = true;
                 if(sq.to_submit > 30 || !lazy) {
+                        ready_schedule_lock();
                         __cfa_io_flush( ctx->proc );
+                        ready_schedule_unlock();
                 }
         }

libcfa/src/concurrency/io/types.hfa

@@ -188 +188 @@
                 return wait(this.self);
         }
+
+        void reset( io_future_t & this ) {
+                return reset(this.self);
+        }
 }

libcfa/src/concurrency/kernel.cfa

@@ -196 +196 @@
 
                         if( !readyThread ) {
+                                ready_schedule_lock();
                                 __cfa_io_flush( this );
+                                ready_schedule_unlock();
+
                                 readyThread = __next_thread_slow( this->cltr );
                         }
@@ -277 +280 @@
 
                         if(this->io.pending && !this->io.dirty) {
+                                ready_schedule_lock();
                                 __cfa_io_flush( this );
+                                ready_schedule_unlock();
                         }
 
@@ -317 +322 @@
 
                                 // Don't block if we are done
-                                if( __atomic_load_n(&this->do_terminate, __ATOMIC_SEQ_CST) ) break MAIN_LOOP;
+                                if( __atomic_load_n(&this->do_terminate, __ATOMIC_SEQ_CST) ) {
+                                        ready_schedule_unlock();
+                                        break MAIN_LOOP;
+                                }
 
                                 __STATS( __tls_stats()->ready.sleep.halts++; )
@@ -939 +947 @@
                         /* paranoid */ verifyf( it, "Unexpected null iterator, at index %u of %u\n", i, count);
                         /* paranoid */ verify( it->local_data->this_stats );
+                        // __print_stats( it->local_data->this_stats, cltr->print_stats, "Processor", it->name, (void*)it );
                         __tally_stats( cltr->stats, it->local_data->this_stats );
                         it = &(*it)`next;
@@ -948 +957 @@
                 // this doesn't solve all problems but does solve many
                 // so it's probably good enough
+                disable_interrupts();
                 uint_fast32_t last_size = ready_mutate_lock();
 
@@ -955 +965 @@
                 // Unlock the RWlock
                 ready_mutate_unlock( last_size );
+                enable_interrupts();
         }