<commit_before>
#include <assert.h>
#include <unistd.h>
#include "event_queue.hpp"
#include "fsm.hpp"
#include "worker_pool.hpp"
#include "networking.hpp"

// Process commands received from the user
int process_command(event_queue_t *event_queue, event_t *event);

// This function returns the socket to clean connected state
void return_to_fsm_socket_connected(event_queue_t *event_queue, fsm_state_t *state) {
    state->state = fsm_state_t::fsm_socket_connected;
    event_queue->alloc.free((io_buffer_t*)state->buf);
    state->buf = NULL;
    state->nbuf = 0;
}

// This state represents a connected socket with no outstanding
// operations. Incoming events should be user commands received by the
// socket.
void fsm_socket_ready(event_queue_t *event_queue, event_t *event) {
    int res;
    size_t sz;
    fsm_state_t *state = (fsm_state_t*)event->state;

    // TODO: Are we checking for reading here? What about when socket
    // is ready to write?
    if(event->event_type == et_sock) {
        if(event->op == eo_rdwr || event->op == eo_read) {
            if(state->buf == NULL) {
                state->buf = (char*)event_queue->alloc.malloc<io_buffer_t>();
                state->nbuf = 0;
            }
            
            // TODO: we assume the command will fit comfortably into
            // IO_BUFFER_SIZE. We'll need to implement streaming later.
            sz = read(state->source,
                      state->buf + state->nbuf,
                      IO_BUFFER_SIZE - state->nbuf);
            check("Could not read from socket", sz == -1);
            if(sz > 0) {
                state->nbuf += sz;
                res = process_command(event_queue, event);
                if(res == -1) {
                    // Command wasn't processed correctly, send error
                    send_err_to_client(event_queue, state);
                }
                if(res == 0 || res == -1) {
                    // TODO: presumably, since we're using edge
                    // trigerred version of epoll, we should try to
                    // read from the socket again in this case, and
                    // only move into fsm_socket_send_incomplete state
                    // when we can't parse the message *and* read
                    // returns EAGAIN or EWOULDBLOCK. Actually, even
                    // if we parse the message successfully, we should
                    // keep reading until these two error messages
                    // (although no transition to incomplete state is
                    // necessary here). This should be done before
                    // production, least we experience weird behavior.
                    if(state->state != fsm_state_t::fsm_socket_send_incomplete) {
                        // Command is either completed or malformed, in any
                        // case get back to clean connected state
                        return_to_fsm_socket_connected(event_queue, state);
                    } else {
                        // We've sent a partial packet to the client, so
                        // we must remain in the partial send state. At
                        // this point there is nothing to do but wait
                        // until we can write to the socket.
                    }
                } else if(res == 1) {
                    state->state = fsm_state_t::fsm_socket_recv_incomplete;
                }
            } else {
                // No data left, destroy the connection
                queue_forget_resource(event_queue, state->source);
                fsm_destroy_state(state, event_queue);
                    
                // TODO: what if the fsm is not in a finished
                // state? What if we free it during an AIO
                // request, and the AIO request comes back? We
                // need an fsm_terminated flag for these cases.

                // TODO: what about application-level keepalive?
            }
        } else {
            // The kernel tells us we're ready to write even when we
            // didn't ask it.
        }
    } else {
        check("fsm_socket_ready: Invalid event", 1);
    }
}

// The socket is ready for sending more information and we were in the
// middle of an incomplete send request.
void fsm_socket_send_incomplete(event_queue_t *event_queue, event_t *event) {
    // TODO: incomplete send needs to be tested therally. It's not
    // clear how to get the kernel to artifically limit the send
    // buffer.
    if(event->event_type == et_sock) {
        if(event->op == eo_rdwr || event->op == eo_write) {
            fsm_state_t *state = (fsm_state_t*)event->state;
            send_msg_to_client(event_queue, state);
            if(state->state != fsm_state_t::fsm_socket_send_incomplete) {
                // We've successfully finished sending the message
                return_to_fsm_socket_connected(event_queue, state);
            } else {
                // We still didn't send the full message yet
            }
        } else {
            // TODO: we've received a message from the client while
            // we're in the process of sending him a message. For now
            // we're dropping the message received out of bound, but
            // what is the proper solution here? (note: same is true
            // above if op == eo_rdwr)
            printf("received out-of-bound msg from client (during a partial response)\n");
        }
    } else {<commit_msg>Modifying fsm read loop to behave as specified in epollet documentation (the implementation is also slightly more complex)<commit_after>
#include <assert.h>
#include <unistd.h>
#include "event_queue.hpp"
#include "fsm.hpp"
#include "worker_pool.hpp"
#include "networking.hpp"

// Process commands received from the user
int process_command(event_queue_t *event_queue, event_t *event);

// This function returns the socket to clean connected state
void return_to_fsm_socket_connected(event_queue_t *event_queue, fsm_state_t *state) {
    event_queue->alloc.free((io_buffer_t*)state->buf);
    fsm_init_state(state);
}

// This state represents a connected socket with no outstanding
// operations. Incoming events should be user commands received by the
// socket.
void fsm_socket_ready(event_queue_t *event_queue, event_t *event) {
    int res;
    size_t sz;
    fsm_state_t *state = (fsm_state_t*)event->state;

    // TODO: Are we checking for reading here? What about when socket
    // is ready to write?
    if(event->event_type == et_sock) {
        if(event->op == eo_rdwr || event->op == eo_read) {
            if(state->buf == NULL) {
                state->buf = (char*)event_queue->alloc.malloc<io_buffer_t>();
                state->nbuf = 0;
            }
            
            // TODO: we assume the command will fit comfortably into
            // IO_BUFFER_SIZE. We'll need to implement streaming later.

            do {
                sz = read(state->source,
                          state->buf + state->nbuf,
                          IO_BUFFER_SIZE - state->nbuf);
                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.
                        if(state->state != fsm_state_t::fsm_socket_recv_incomplete)
                            return_to_fsm_socket_connected(event_queue, state);
                        break;
                    } else {
                        check("Could not read from socket", sz == -1);
                    }
                } else if(sz > 0) {
                    state->nbuf += sz;
                    res = process_command(event_queue, event);
                    if(res == -1 || res == 0) {
                        if(res == -1) {
                            // Command wasn't processed correctly, send error
                            send_err_to_client(event_queue, state);
                        }
                        if(state->state != fsm_state_t::fsm_socket_send_incomplete) {
                            // Command is either completed or malformed, in any
                            // case get back to clean connected state
                            state->state = fsm_state_t::fsm_socket_connected;
                            state->nbuf = 0;
                            state->snbuf = 0;
                        } else {
                            // Wait for the socket to finish sending
                            break;
                        }
                    } else if(res == 1) {
                        state->state = fsm_state_t::fsm_socket_recv_incomplete;
                    }
                } else {
                    // Socket has been closed, destroy the connection
                    queue_forget_resource(event_queue, state->source);
                    fsm_destroy_state(state, event_queue);
                    break;
                    
                    // TODO: what if the fsm is not in a finished
                    // state? What if we free it during an AIO
                    // request, and the AIO request comes back? We
                    // need an fsm_terminated flag for these cases.

                    // TODO: what about application-level keepalive?
                }
            } while(1);
        } else {
            // The kernel tells us we're ready to write even when we
            // didn't ask it.
        }
    } else {
        check("fsm_socket_ready: Invalid event", 1);
    }
}

// The socket is ready for sending more information and we were in the
// middle of an incomplete send request.
void fsm_socket_send_incomplete(event_queue_t *event_queue, event_t *event) {
    // TODO: incomplete send needs to be tested therally. It's not
    // clear how to get the kernel to artifically limit the send
    // buffer.
    if(event->event_type == et_sock) {
        fsm_state_t *state = (fsm_state_t*)event->state;
        if(event->op == eo_rdwr || event->op == eo_write) {
            send_msg_to_client(event_queue, state);
            if(state->state != fsm_state_t::fsm_socket_send_incomplete) {
                // We've successfully finished sending the message
                return_to_fsm_socket_connected(event_queue, state);
            } else {
                // We still didn't send the full message yet
            }
        }
        if(state->state != fsm_state_t::fsm_socket_send_incomplete) {
            // We've finished sending completely, now see if there is
            // anything left to read from the old epoll notification
            fsm_socket_ready(event_queue, event);
        }
    } else {
{'fraction_non_alphanumeric': 0.08577232019204137, 'fraction_numerical': 0.0019388791432000739, 'mean_word_length': 1.574756358450202, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 65, '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': '5339403', 'n_tokens_mistral': 2921, 'n_tokens_neox': 2776, 'n_words': 901}