<commit_before>#include "conn_fsm.hpp"
#include <unistd.h>
#include <errno.h>
#include "utils.hpp"
#include "request_handler/memcached_handler.hpp"
#include <signal.h>
#include "worker_pool.hpp"

/*
~~~ A Brief History of the conn_fsm_t ~~~

Initially, the event queue would call conn_fsm_t::do_transition() and
then examine the result to figure out what to do.

When Tim abstracted the IO layer, then the events would come through
conn_fsm_t::on_net_conn_* instead; however, he kept do_transition()
because he didn't want to rewrite the whole conn_fsm_t.

So event_t and conn_fsm_t::result_t are actually more or less obsolete.
TODO: Rewrite this shit.
*/

void conn_fsm_t::init_state() {
    state = fsm_socket_connected;
    rbuf = NULL;
    sbuf = NULL;
    nrbuf = 0;
    corked = false;

    ext_rbuf = NULL;
    ext_nrbuf = 0;
    ext_size = 0;
    cb = NULL;
}

// This function returns the socket to clean connected state
void conn_fsm_t::return_to_socket_connected() {
    if(rbuf)
        delete (iobuf_t*)(rbuf);
    if(sbuf)
        delete (linked_buf_t*)(sbuf);
    init_state();
}

conn_fsm_t::result_t conn_fsm_t::fill_buf(void *buf, unsigned int *bytes_filled, unsigned int total_length) {
    // TODO: we assume the command will fit comfortably into
    // IO_BUFFER_SIZE. We'll need to implement streaming later.
    ssize_t sz = conn->read_nonblocking((byte *) buf + *bytes_filled, total_length - *bytes_filled);
    if(sz == -1) {
        if(errno == EAGAIN || errno == EWOULDBLOCK) {
            // The machine can't be in
            // fsm_socket_send_incomplete state here,
            // since we break out in these cases. So it's
            // safe to free the buffer.
            assert(state != fsm_socket_send_incomplete);
            //TODO Modify this so that we go into send_incomplete and try to empty our send buffer
            if(state != fsm_socket_recv_incomplete && *bytes_filled == 0)
                return_to_socket_connected();
            else
                state = fsm_socket_connected; //we're wating for a socket event
            //break;
        } else if (errno == ENETDOWN) {
            check("Enetdown wtf", sz == -1);
        } else {
            check("Could not read from socket", sz == -1);
        }
    } else if(sz > 0 || *bytes_filled > 0) {
        *bytes_filled += sz;
        if (state != fsm_socket_recv_incomplete)
            state = fsm_outstanding_data;
    } else {
        if (state == fsm_socket_recv_incomplete)
            return fsm_no_data_in_socket;
        else
            return fsm_quit_connection;
        // TODO: what about application-level keepalive?
    }

    return fsm_transition_ok;
}

// This state represents a connected socket with no outstanding
// operations. Incoming events should be user commands received by the
// socket.
conn_fsm_t::result_t conn_fsm_t::fill_rbuf() {
    if(rbuf == NULL) {
        rbuf = (char *) new iobuf_t();
        nrbuf = 0;
    }
    if(sbuf == NULL) {
        sbuf = new linked_buf_t();
    }
    return fill_buf(rbuf, &nrbuf, iobuf_t::size);
}

conn_fsm_t::result_t conn_fsm_t::fill_ext_rbuf() {
    assert(ext_rbuf);
    assert(nrbuf == 0);
    return fill_buf(ext_rbuf, &ext_nrbuf, ext_size);
}

conn_fsm_t::result_t conn_fsm_t::read_data(event_t *event) {
    // TODO: this is really silly; this notification should be done differently
    assert((conn_fsm_t *) event->state == this);

    if (ext_rbuf) {
        result_t res = fill_ext_rbuf();
        check_external_buf();
        return res;
    } else {
        return fill_rbuf();
    }
}

void conn_fsm_t::fill_external_buf(byte *external_buf, unsigned int size, data_transferred_callback *callback) {
    assert(!ext_rbuf && ext_nrbuf == 0 && ext_size == 0 && !cb);
    //assert(!ext_rbuf && ext_nrbuf == 0 && ext_size == 0);// && !cb);

    ext_rbuf = external_buf;
    ext_size = size;
    cb = callback;<commit_msg>Adds a todo about an inefficient part of the code.<commit_after>#include "conn_fsm.hpp"
#include <unistd.h>
#include <errno.h>
#include "utils.hpp"
#include "request_handler/memcached_handler.hpp"
#include <signal.h>
#include "worker_pool.hpp"

/*
~~~ A Brief History of the conn_fsm_t ~~~

Initially, the event queue would call conn_fsm_t::do_transition() and
then examine the result to figure out what to do.

When Tim abstracted the IO layer, then the events would come through
conn_fsm_t::on_net_conn_* instead; however, he kept do_transition()
because he didn't want to rewrite the whole conn_fsm_t.

So event_t and conn_fsm_t::result_t are actually more or less obsolete.
TODO: Rewrite this shit.
*/

void conn_fsm_t::init_state() {
    state = fsm_socket_connected;
    rbuf = NULL;
    sbuf = NULL;
    nrbuf = 0;
    corked = false;

    ext_rbuf = NULL;
    ext_nrbuf = 0;
    ext_size = 0;
    cb = NULL;
}

// This function returns the socket to clean connected state
void conn_fsm_t::return_to_socket_connected() {
    if(rbuf)
        delete (iobuf_t*)(rbuf);
    if(sbuf)
        delete (linked_buf_t*)(sbuf);
    init_state();
}

conn_fsm_t::result_t conn_fsm_t::fill_buf(void *buf, unsigned int *bytes_filled, unsigned int total_length) {
    // TODO: we assume the command will fit comfortably into
    // IO_BUFFER_SIZE. We'll need to implement streaming later.
    ssize_t sz = conn->read_nonblocking((byte *) buf + *bytes_filled, total_length - *bytes_filled);
    if(sz == -1) {
        if(errno == EAGAIN || errno == EWOULDBLOCK) {
            // The machine can't be in
            // fsm_socket_send_incomplete state here,
            // since we break out in these cases. So it's
            // safe to free the buffer.
            assert(state != fsm_socket_send_incomplete);
            //TODO Modify this so that we go into send_incomplete and try to empty our send buffer
            if(state != fsm_socket_recv_incomplete && *bytes_filled == 0)
                return_to_socket_connected();
            else
                state = fsm_socket_connected; //we're wating for a socket event
            //break;
        } else if (errno == ENETDOWN) {
            check("Enetdown wtf", sz == -1);
        } else {
            check("Could not read from socket", sz == -1);
        }
    } else if(sz > 0 || *bytes_filled > 0) {
        *bytes_filled += sz;
        if (state != fsm_socket_recv_incomplete)
            state = fsm_outstanding_data;
    } else {
        if (state == fsm_socket_recv_incomplete)
            return fsm_no_data_in_socket;
        else
            return fsm_quit_connection;
        // TODO: what about application-level keepalive?
    }

    return fsm_transition_ok;
}

// This state represents a connected socket with no outstanding
// operations. Incoming events should be user commands received by the
// socket.
conn_fsm_t::result_t conn_fsm_t::fill_rbuf() {
    if(rbuf == NULL) {
        rbuf = (char *) new iobuf_t();
        nrbuf = 0;
    }
    if(sbuf == NULL) {
        sbuf = new linked_buf_t(); //TODO This line is a source of inefficiency, we should only delete the sbuf if the connection sits idle for a while
    }
    return fill_buf(rbuf, &nrbuf, iobuf_t::size);
}

conn_fsm_t::result_t conn_fsm_t::fill_ext_rbuf() {
    assert(ext_rbuf);
    assert(nrbuf == 0);
    return fill_buf(ext_rbuf, &ext_nrbuf, ext_size);
}

conn_fsm_t::result_t conn_fsm_t::read_data(event_t *event) {
    // TODO: this is really silly; this notification should be done differently
    assert((conn_fsm_t *) event->state == this);

    if (ext_rbuf) {
        result_t res = fill_ext_rbuf();
        check_external_buf();
        return res;
    } else {
        return fill_rbuf();
    }
}

void conn_fsm_t::fill_external_buf(byte *external_buf, unsigned int size, data_transferred_callback *callback) {
    assert(!ext_rbuf && ext_nrbuf == 0 && ext_size == 0 && !cb);
    //assert(!ext_rbuf && ext_nrbuf == 0 && ext_size == 0);// && !cb);

    ext_rbuf = external_buf;
    ext_size = size;
    cb = callback;
{'fraction_non_alphanumeric': 0.10729506141890198, 'fraction_numerical': 0.0037603409375783405, 'mean_word_length': 2.804959465903672, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2830525', 'n_tokens_mistral': 2690, 'n_tokens_neox': 2612, 'n_words': 742}