source: benchmark/io/http/protocol.cfa @ 6528d75

#include "protocol.hfa"

#define _GNU_SOURCE
extern "C" {
        #include <fcntl.h>
}

#include <fstream.hfa>
#include <iofwd.hfa>

#include <assert.h>
// #include <stdio.h> // Don't use stdio.h, too slow to compile
extern "C" {
      int snprintf ( char * s, size_t n, const char * format, ... );
        // #include <linux/io_uring.h>
}
#include <string.h>
#include <errno.h>

#include "options.hfa"

#define PLAINTEXT_1WRITE
#define PLAINTEXT_NOCOPY

struct https_msg_str {
        char msg[512];
        size_t len;
};

const https_msg_str * volatile http_msgs[KNOWN_CODES] = { 0 };

_Static_assert( KNOWN_CODES == (sizeof(http_msgs ) / sizeof(http_msgs [0])));

const int http_codes[KNOWN_CODES] = {
        200,
        200,
        400,
        404,
        405,
        408,
        413,
        414,
};

_Static_assert( KNOWN_CODES == (sizeof(http_codes) / sizeof(http_codes[0])));

int code_val(HttpCode code) {
        return http_codes[code];
}

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 == EAGAIN || errno == EWOULDBLOCK) return -EAGAIN;

                        abort( "'answer error' error: (%d) %s\n", (int)errno, strerror(errno) );
                }

                // update it/len
                it  += ret;
                len -= ret;
        }
        return 0;
}

int answer_error( int fd, HttpCode code ) {
        /* paranoid */ assert( code < KNOWN_CODES && code != OK200 );
        int idx = (int)code;
        return answer( fd, http_msgs[idx]->msg, http_msgs[idx]->len );
}

int answer_header( int fd, size_t size ) {
        char buffer[512];
        char * it = buffer;
        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 answer( fd, buffer, len );
}

#if defined(PLAINTEXT_NOCOPY)
int answer_plaintext( int fd ) {
        return answer(fd, http_msgs[OK200_PlainText]->msg, http_msgs[OK200_PlainText]->len + 1); // +1 cause snprintf doesn't count nullterminator
}
#elif defined(PLAINTEXT_1WRITE)
int answer_plaintext( int fd ) {
        char text[] = "Hello, World!\n";
        char buffer[512 + sizeof(text)];
        char * it = buffer;
        memcpy(it, http_msgs[OK200]->msg, http_msgs[OK200]->len);
        it += http_msgs[OK200]->len;
        int len = http_msgs[OK200]->len;
        int r = snprintf(it, 512 - len, "%d \n\n", sizeof(text));
        it += r;
        len += r;
        memcpy(it, text, sizeof(text));
        return answer(fd, buffer, len + sizeof(text));
}
#else
int answer_plaintext( int fd ) {
        char text[] = "Hello, World!\n";
        int ret = answer_header(fd, sizeof(text));
        if( ret < 0 ) return ret;
        return answer(fd, text, sizeof(text));
}
#endif

int answer_empty( int fd ) {
        return answer_header(fd, 0);
}


[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 ) {
        unsigned sflags = SPLICE_F_MOVE; // | SPLICE_F_MORE;
        off_t offset = 0;
        ssize_t ret;
        SPLICE1: while(count > 0) {
                ret = cfa_splice(ans_fd, &offset, pipe[1], 0p, count, sflags, CFA_IO_LAZY);
                if( ret < 0 ) {
                        if( errno != EAGAIN && errno != EWOULDBLOCK) continue SPLICE1;
                        if( errno == ECONNRESET ) return -ECONNRESET;
                        if( errno == EPIPE ) return -EPIPE;
                        abort( "splice [0] error: (%d) %s\n", (int)errno, strerror(errno) );
                }

                count -= ret;
                offset += ret;
                size_t in_pipe = ret;
                SPLICE2: while(in_pipe > 0) {
                        ret = cfa_splice(pipe[0], 0p, fd, 0p, in_pipe, sflags, CFA_IO_LAZY);
                        if( ret < 0 ) {
                                if( errno != EAGAIN && errno != EWOULDBLOCK) continue SPLICE2;
                                if( errno == ECONNRESET ) return -ECONNRESET;
                                if( errno == EPIPE ) return -EPIPE;
                                abort( "splice [1] error: (%d) %s\n", (int)errno, strerror(errno) );
                        }
                        in_pipe -= ret;
                }

        }
        return count;
}

//=============================================================================================

#include <clock.hfa>
#include <time.hfa>
#include <thread.hfa>

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",
        "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",
};

struct date_buffer {
        https_msg_str strs[KNOWN_CODES];
};

thread DateFormater {
        int idx;
        date_buffer buffers[2];
};

void ?{}( DateFormater & this ) {
        ((thread&)this){ "Server Date Thread", *options.clopts.instance[0] };
        this.idx = 0;
        memset( &this.buffers[0], 0, sizeof(this.buffers[0]) );
        memset( &this.buffers[1], 0, sizeof(this.buffers[1]) );
}

void main(DateFormater & this) {
        LOOP: for() {
                waitfor( ^?{} : this) {
                        break LOOP;
                }
                or else {}


                char buff[100];
                Time now = timeNsec();
                strftime( buff, 100, "%a, %d %b %Y %H:%M:%S %Z", now );
                sout | "Updated date to '" | buff | "'";

                for(i; KNOWN_CODES) {
                        size_t len = snprintf( this.buffers[this.idx].strs[i].msg, 512, original_http_msgs[i], buff );
                        this.buffers[this.idx].strs[i].len = len;
                }

                for(i; KNOWN_CODES) {
                        https_msg_str * next = &this.buffers[this.idx].strs[i];
                        __atomic_exchange_n((https_msg_str * volatile *)&http_msgs[i], next, __ATOMIC_SEQ_CST);
                }
                this.idx = (this.idx + 1) % 2;

                sout | "Date thread sleeping";

                sleep(1`s);
        }
}

//=============================================================================================
DateFormater * the_date_formatter;

void init_protocol(void) {
        the_date_formatter = alloc();
        (*the_date_formatter){};
}

void deinit_protocol(void) {
        ^(*the_date_formatter){};
        free( the_date_formatter );
}