source: benchmark/io/http/worker.cfa @ bf7c7ea

#include "worker.hfa"

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

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

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

//=============================================================================================
// Generic connection handling
//=============================================================================================
static void handle_connection( connection & this, volatile int & fd, char * buffer, size_t len, io_future_t * f, unsigned long long & last ) {
        REQUEST:
        for() {
                bool closed;
                HttpCode code;
                const char * file;
                size_t name_size;

                // Read the http request
                if( options.log ) mutex(sout) sout | "=== Reading request ===";
                [code, closed, file, name_size] = http_read(fd, buffer, len, f);
                f = 0p;

                // 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 ) mutex(sout) sout | "=== Request for /plaintext ===";

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

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

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

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

                        if( options.log ) mutex(sout) 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 ) mutex(sout) sout | "=== Answer sent ===";
        }

        if (stats_thrd) {
                unsigned long long next = rdtscl();
                if(next > (last + 500000000)) {
                        if(try_lock(stats_thrd->stats.lock __cfaabi_dbg_ctx2)) {
                                push(this.stats.sendfile, stats_thrd->stats.send);
                                unlock(stats_thrd->stats.lock);
                                last = next;
                        }
                }
        }
}

//=============================================================================================
// Self Accepting Worker Thread
//=============================================================================================
void ?{}( AcceptWorker & this ) {
        ((thread&)this){ "Server Worker Thread", *options.clopts.instance, 64000 };
        options.clopts.thrd_cnt++;
        this.done = false;
}

void main( AcceptWorker & this ) {
        park();
        unsigned long long last = rdtscl();
        /* paranoid */ assert( this.conn.pipe[0] != -1 );
        /* paranoid */ assert( this.conn.pipe[1] != -1 );
        for() {
                if( options.log ) mutex(sout) sout | "=== Accepting connection ===";
                int fd = cfa_accept4( this.sockfd, this.[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 ) mutex(sout) sout | "=== New connection" | fd | "" | ", waiting for requests ===";
                size_t len = options.socket.buflen;
                char buffer[len];
                handle_connection( this.conn, fd, buffer, len, 0p, last );

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


//=============================================================================================
// Channel Worker Thread
//=============================================================================================
void ?{}( ChannelWorker & this ) {
        ((thread&)this){ "Server Worker Thread", *options.clopts.instance, 64000 };
        options.clopts.thrd_cnt++;
        this.done = false;
}

void main( ChannelWorker & this ) {
        park();
        unsigned long long last = rdtscl();
        /* paranoid */ assert( this.conn.pipe[0] != -1 );
        /* paranoid */ assert( this.conn.pipe[1] != -1 );
        for() {
                size_t len = options.socket.buflen;
                char buffer[len];
                PendingRead p;
                p.next = 0p;
                p.in.buf = (void*)buffer;
                p.in.len = len;
                push(*this.queue, &p);

                if( options.log ) mutex(sout) sout | "=== Waiting new connection ===";
                handle_connection( this.conn, p.out.fd, buffer, len, &p.f, last );

                if( options.log ) mutex(sout) sout | "=== Connection closed ===";
                if(this.done) break;
        }
}

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

void ?{}( Acceptor & this ) {
        ((thread&)this){ "Server Acceptor Thread", *options.clopts.instance, 64000 };
        options.clopts.thrd_cnt++;
        this.done = false;
}

#define ACCEPT_SPIN

void main( Acceptor & this ) {
        park();
        unsigned long long last = rdtscl();

#if defined(ACCEPT_SPIN)
        if( options.log ) sout | "=== Accepting connection ===";
        for() {
                int fd = accept4(this.sockfd, this.[addr, addrlen, flags]);
                if(fd < 0) {
                        if( errno == EWOULDBLOCK) {
                                this.stats.eagains++;
                                yield();
                                continue;
                        }
                        if( errno == ECONNABORTED ) break;
                        if( this.done && (errno == EINVAL || errno == EBADF) ) break;
                        abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
                }
                this.stats.accepts++;

                if(this.done) return;

                if( options.log ) sout | "=== New connection" | fd | "" | ", waiting for requests ===";

                if(fd) {
                        PendingRead * p = 0p;
                        for() {
                                if(this.done) return;
                                p = pop(*this.queue);
                                if(p) break;
                                yield();
                                this.stats.creates++;
                        };

                        p->out.fd = fd;
                        async_recv(p->f, p->out.fd, p->in.buf, p->in.len, 0, CFA_IO_LAZY);
                }

                if (stats_thrd) {
                        unsigned long long next = rdtscl();
                        if(next > (last + 500000000)) {
                                if(try_lock(stats_thrd->stats.lock)) {
                                        push(this.stats, stats_thrd->stats.accpt);
                                        unlock(stats_thrd->stats.lock);
                                        last = next;
                                }
                        }
                }

                if( options.log ) sout | "=== Accepting connection ===";
        }

#elif define(ACCEPT_MANY)
        const int nacc = 10;
        io_future_t results[nacc];

        for(i; nacc) {
                io_future_t & res = results[i];
                reset(res);
                /* paranoid */ assert(!available(res));
                if( options.log ) mutex(sout) sout | "=== Re-arming accept no" | i | " ===";
                async_accept4(res, this.sockfd, this.[addr, addrlen, flags], CFA_IO_LAZY);
        }

        for() {
                if (stats_thrd) {
                        unsigned long long next = rdtscl();
                        if(next > (last + 500000000)) {
                                if(try_lock(stats_thrd->stats.lock __cfaabi_dbg_ctx2)) {
                                        push(this.stats, stats_thrd->stats.accpt);
                                        unlock(stats_thrd->stats.lock);
                                        last = next;
                                }
                        }
                }

                for(i; nacc) {
                        io_future_t & res = results[i];
                        if(available(res)) {
                                if( options.log ) mutex(sout) sout | "=== Accept no " | i | "completed with result" | res.result | "===";
                                int fd = get_res(res);
                                reset(res);
                                this.stats.accepts++;
                                if(fd < 0) {
                                        if( errno == ECONNABORTED ) continue;
                                        if( this.done && (errno == EINVAL || errno == EBADF) ) continue;
                                        abort( "accept error: (%d) %s\n", (int)errno, strerror(errno) );
                                }
                                push_connection( this, fd );

                                /* paranoid */ assert(!available(res));
                                if( options.log ) mutex(sout) sout | "=== Re-arming accept no" | i | " ===";
                                async_accept4(res, this.sockfd, this.[addr, addrlen, flags], CFA_IO_LAZY);
                        }
                }
                if(this.done) return;

                if( options.log ) mutex(sout) sout | "=== Waiting for any accept ===";
                this.stats.eagains++;
                wait_any(results, nacc);

                if( options.log ) mutex(sout) {
                        sout | "=== Acceptor wake-up ===";
                        for(i; nacc) {
                                io_future_t & res = results[i];
                                sout | i | "available:" | available(res);
                        }
                }

        }

        for(i; nacc) {
                wait(results[i]);
        }
#else
#error no accept algorithm specified
#endif
}