source: benchmark/io/http/worker.cfa@ 6e2b04e

#include "worker.hfa"

#include <errno.h>
#include <stdio.h>
#include <string.h>
#include <unistd.h>

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

#include "options.hfa"
#include "protocol.hfa"
#include "filecache.hfa"

//=============================================================================================
// Worker Thread
//=============================================================================================
void ?{}( Worker & this ) {
        size_t cli = rand() % options.clopts.cltr_cnt;
        ((thread&)this){ "Server Worker Thread", *options.clopts.instance[cli], 64000 };
        options.clopts.thrd_cnt[cli]++;
        this.pipe[0] = -1;
        this.pipe[1] = -1;
        this.done = false;

        this.stats.sendfile.calls = 0;
        this.stats.sendfile.tries = 0;
        this.stats.sendfile.header = 0;
        this.stats.sendfile.splcin = 0;
        this.stats.sendfile.splcot = 0;
        for(i; zipf_cnts) {
                this.stats.sendfile.avgrd[i].calls = 0;
                this.stats.sendfile.avgrd[i].bytes = 0;
        }
}

extern "C" {
extern int accept4(int sockfd, struct sockaddr *addr, socklen_t *addrlen, int flags);
}

void main( Worker & this ) {
        park();
        /* paranoid */ assert( this.pipe[0] != -1 );
        /* paranoid */ assert( this.pipe[1] != -1 );

        const bool reuse = options.socket.manyreuse;

        CONNECTION:
        for() {
                if( options.log ) sout | "=== Accepting connection ===";
                int fd = cfa_accept4( this.[sockfd, addr, addrlen, flags], CFA_IO_LAZY );
                if(fd < 0) {
                        if( errno == ECONNABORTED ) break;
                        if( this.done && (errno == EINVAL || errno == EBADF) ) break;
                        abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
                }
                if(this.done) break;

                if( options.log ) sout | "=== New connection" | fd | "" | ", waiting for requests ===";
                REQUEST:
                for() {
                        bool closed;
                        HttpCode code;
                        const char * file;
                        size_t name_size;

                        // Read the http request
                        size_t len = options.socket.buflen;
                        char buffer[len];
                        if( options.log ) sout | "=== Reading request ===";
                        [code, closed, file, name_size] = http_read(fd, buffer, len);

                        // if we are done, break out of the loop
                        if( closed ) break REQUEST;

                        // If this wasn't a request retrun 400
                        if( code != OK200 ) {
                                sout | "=== Invalid Request :" | code_val(code) | "===";
                                answer_error(fd, code);
                                continue REQUEST;
                        }

                        if(0 == strncmp(file, "plaintext", min(name_size, sizeof("plaintext") ))) {
                                if( options.log ) sout | "=== Request for /plaintext ===";

                                int ret = answer_plaintext(fd);
                                if( ret == -ECONNRESET ) break REQUEST;

                                if( options.log ) sout | "=== Answer sent ===";
                                continue REQUEST;
                        }

                        if(0 == strncmp(file, "ping", min(name_size, sizeof("ping") ))) {
                                if( options.log ) sout | "=== Request for /ping ===";

                                // Send the header
                                int ret = answer_empty(fd);
                                if( ret == -ECONNRESET ) break REQUEST;

                                if( options.log ) sout | "=== Answer sent ===";
                                continue REQUEST;
                        }

                        if( options.log ) {
                                sout | "=== Request for file " | nonl;
                                write(sout, file, name_size);
                                sout | " ===";
                        }

                        if( !options.file_cache.path ) {
                                if( options.log ) {
                                        sout | "=== File Not Found (" | nonl;
                                        write(sout, file, name_size);
                                        sout | ") ===";
                                }
                                answer_error(fd, E405);
                                continue REQUEST;
                        }

                        // Get the fd from the file cache
                        int ans_fd;
                        size_t count;
                        [ans_fd, count] = get_file( file, name_size );

                        // If we can't find the file, return 404
                        if( ans_fd < 0 ) {
                                if( options.log ) {
                                        sout | "=== File Not Found (" | nonl;
                                        write(sout, file, name_size);
                                        sout | ") ===";
                                }
                                answer_error(fd, E404);
                                continue REQUEST;
                        }

                        // Send the desired file
                        int ret = answer_sendfile( this.pipe, fd, ans_fd, count, this.stats.sendfile );
                        if( ret == -ECONNRESET ) break REQUEST;

                        if( options.log ) sout | "=== Answer sent ===";
                }

                if( options.log ) sout | "=== Connection closed ===";
                continue CONNECTION;
        }
}