sender_utils.c

#include "sender_utils.h"


sender_state_t *state_new(bool fec_enabled)
{
    sender_state_t *state = (sender_state_t *) calloc(1, sizeof(sender_state_t));
    if (state == NULL)
    {
        return NULL;
    }

    state->r_window_size = 1; // defined in the report
    state->s_window_size = WINDOW_SIZE;
    state->head = 0;
    state->tail = 0;
    for (uint8_t i = 0; i < WINDOW_SIZE; i++)
    {
        state->buffer[i] = NULL;
    }
    for (uint16_t i = 0; i < SEQNUM_RANGE; i++)
    {
        state->map_seqnum_to_buffer_place[i] = OUT_OFF_WINDOW;
    }
    state->last_pkt_sent = RANDOM_PKT; // default
    state->fec_enabled = fec_enabled;
    if (fec_enabled)
    {
        state->FEC = pkt_new();
        state->FEC_nbr = 0;
    }
    state->stats = calloc(sizeof(transfer_stats_t), 1);
    return state;
}

void state_del(sender_state_t *state)
{
    // To be sure, we free the pkt that might not have been freed (in case of an error or timeout)
    for (uint8_t i = 0; i < WINDOW_SIZE; i++)
    {
        if (state->buffer[i] != NULL)
        {
            pkt_del(state->buffer[i]);
        }
    }
    if (state->fec_enabled)
    {
        pkt_del(state->FEC);
    }
    free(state->stats);
    free(state);
}

bool can_send(sender_state_t *state)
{
    if (state->last_pkt_sent == RANDOM_PKT)
    {
        return (state->r_window_size > 0) && (state->s_window_size > 0);
    }
    else if (state->last_pkt_sent == LAST_DATA_PKT)
    {
        // I want to be sure all the data pkt has been received before letting know the receiver
        // it was the end of the file (so that I can set a timer for timeout)
        return state->s_window_size == MAX_WINDOW_SIZE;
    }
    // Case: we're in FEC mode, the closing pkt has been sent but we will still send a last FEC if possible
    else if ((state->fec_enabled) && (state->last_pkt_sent == CLOSING_PKT) && (state->FEC_nbr > 0))
    {
        return true;
    }
    // Case last FEC has been sended
    else
    {
        return false;
    }
}

/** Cautious this function is used for sending and sending back pkt ! */
int send_pkt(sender_state_t *state, pkt_t *pkt, uint8_t position, int socket_fd)
{
    DEBUG("Sending the pkt with seqnum: %d", pkt_get_seqnum(pkt));
    char packet_to_be_sent[PKT_MAX_LEN];
    size_t len = PKT_MAX_LEN;
    pkt_status_code pkt_status = pkt_encode(pkt, packet_to_be_sent, &len);
    if (pkt_status != PKT_OK) 
    {
        ERROR("pkt_encode failed with status: %d", pkt_status);
        return -1;
    }
    
    ssize_t sent = send(socket_fd, packet_to_be_sent, len, 0);
    if (sent == -1) 
    {
        ERROR("The sending (using the function send from <sys/socket.h>) of the pkt failed");
        return -1;
    }
    
    // Insert the packet in the buffer
    state->buffer[position] = pkt;
    struct timeval time;
    gettimeofday(&time, NULL);
    state->timers[position] = time;
    state->stats->data_sent++;
    return 0;
}

int handle_returning_ack_nack(sender_state_t *state, int socket_fd)
{
    char buffer[ACK_OR_NACK_HEADER_SIZE]; // Only 10 bytes
    size_t read_bytes = read(socket_fd, buffer, ACK_OR_NACK_HEADER_SIZE);
    if (read_bytes == (size_t) -1) return -1;

    pkt_t *pkt = pkt_new();
    pkt_status_code pkt_status = pkt_decode(buffer, read_bytes, pkt);
    if (pkt_status != PKT_OK)
    {
        state->stats->packet_ignored++;
        pkt_del(pkt);
        DEBUG("Decode function on a received pkt failed with status: %d so we discard the pkt", pkt_status);
        return 0;
    }
    uint8_t seqnum    = pkt_get_seqnum(pkt);
    ptypes_t pkt_type = pkt_get_type(pkt);
    uint8_t r_window  = pkt_get_window(pkt);
    pkt_del(pkt);

    // Handling NACK:
    if (pkt_type == PTYPE_NACK) 
    {
        state->stats->nack_received++;
        uint8_t place = state->map_seqnum_to_buffer_place[seqnum];
        if (place == OUT_OFF_WINDOW)
        {
            DEBUG("The NACK with the seqnum: %d is out of the sended window so it has been discarded", seqnum);
        }
        else
        {
            state->stats->packet_retransmitted++;
            DEBUG("The NACK with the seqnum: %d has been received, so the pkt will be sent back", seqnum);
            state->r_window_size = r_window - 1; // -1 Because the receiver doesn't count yet the sended pkt
            pkt_t *n_pkt = state->buffer[place];
            if (send_pkt(state, n_pkt, place, socket_fd) == -1) return -1;
        }
    }
    // Handling ACK:
    else
    {
        state->stats->ack_received++;
        DEBUG("The ACK with the seqnum: %d has been received", seqnum);
        uint8_t seqnum_nack = seqnum;
        uint8_t seqnum_ack  = seqnum == 0 ? MAX_SEQNUM_SIZE : seqnum - 1;
        
        uint8_t place_last_ack  = state->map_seqnum_to_buffer_place[seqnum_ack];
        uint8_t place_last_nack = state->map_seqnum_to_buffer_place[seqnum_nack];
        
        // Nothing need to be acknowledged
        if (place_last_ack == OUT_OFF_WINDOW)
        {
            // Checking if it's in my window:
            if (place_last_nack == OUT_OFF_WINDOW)
            {
                DEBUG("The ACK with the seqnum: %d has been discarded because not in my window", seqnum_nack);
                return 0;
            }
            else
            {
                state->stats->packet_retransmitted++;
                DEBUG("The receiver is asking AGAIN the seqnum: %d so the sender sends it back", seqnum_nack);
                state->r_window_size = r_window - 1; // -1 Because the receiver doesn't count yet the sended pkt
                pkt_t *n_pkt = state->buffer[place_last_nack];
                if (send_pkt(state, n_pkt, place_last_nack, socket_fd) == -1) return -1;
                return 0;
            }
        }
        // Some pkt need to be acknowledged
        else
        {
            state->r_window_size = r_window;

            uint8_t upper_bound = (place_last_ack + 1) % WINDOW_SIZE; // The oldest seqnum sended
            DEBUG("The sender is cumulatively acknowledging [%d : %d[ (place in the buffer) | [%d, %d[ (seqnum)", 
                    state->tail, upper_bound, pkt_get_seqnum(state->buffer[state->tail]), seqnum_nack);
            // A do while is necessary here in case we want to make a full revolution (ack from 1 to 1 not included for example)
            do 
            {
                struct timeval time;
                gettimeofday(&time, NULL);
                unsigned long long int delta_time = time_milliseconds(&time) - time_milliseconds(&state->timers_first_send[state->tail]);
                if (state->stats->min_rtt == 0 || state->stats->min_rtt > delta_time)
                {
                    state->stats->min_rtt = delta_time;
                }
                if (state->stats->max_rtt < delta_time)
                {
                    state->stats->max_rtt = delta_time;
                }

                pkt_t *n_pkt = state->buffer[state->tail];
                seqnum = pkt_get_seqnum(n_pkt);
                state->map_seqnum_to_buffer_place[seqnum] = OUT_OFF_WINDOW;
                pkt_del(n_pkt);

                state->buffer[state->tail] = NULL;
                
                state->s_window_size++;
                state->tail = (state->tail + 1) % WINDOW_SIZE;
            } while (state->tail != upper_bound);

            // Send back the asked ACK if there is one to send back
            if (place_last_nack != OUT_OFF_WINDOW)
            {
                state->stats->packet_retransmitted++;
                pkt_t *n_pkt = state->buffer[place_last_nack];
                if (send_pkt(state, n_pkt, place_last_nack, socket_fd) == -1) return -1;
            }
        }
    }
    return 0;
}

int checking_timer(sender_state_t *state, int socket_fd)
{
    struct timeval time;
    // Checking if the slot contains a sended packet
    if (state->buffer[state->tail] != NULL)
    {
        gettimeofday(&time, NULL);
        // When the timer is over, we send the packet back
        if ((time_milliseconds(&time) - time_milliseconds(&state->timers[state->tail])) >= TIMER_LIMIT)
        {
            state->stats->packet_retransmitted++;
            pkt_t *pkt = state->buffer[state->tail];
            DEBUG("The pkt with seqnum: %d has timeout", pkt_get_seqnum(pkt));
            if (send_pkt(state, pkt, state->tail, socket_fd) == -1) return -1;
        }
    }
    return 0;
}

bool is_it_EOF(int sending_fd)
{
    // Checking if we're at the end of the file
    off_t curr_position = lseek(sending_fd, 0, SEEK_CUR);
    off_t end_position  = lseek(sending_fd, 0, SEEK_END);
    // put the reader cursor back
    lseek(sending_fd, curr_position, SEEK_SET);
    return (bool) (curr_position == end_position);
}

void construct_FEC(sender_state_t *state, pkt_t *pkt)
{
    if (state->FEC_nbr == 0)
    {
        pkt_set_seqnum(state->FEC, pkt_get_seqnum(pkt));
    }

    uint16_t length = pkt_get_length(state->FEC) ^ pkt_get_length(pkt);
    pkt_set_length(state->FEC, length);

    uint8_t payload[MAX_PAYLOAD_SIZE];
    uint8_t *p1 = (uint8_t *) pkt_get_payload(state->FEC);
    uint8_t *p2 = (uint8_t *) pkt_get_payload(pkt);
    for (int i = 0; i < MAX_PAYLOAD_SIZE; i++)
    {
        payload[i] = p1[i] ^ p2[i];
    }
    pkt_set_payload(state->FEC, (const char *) payload, MAX_PAYLOAD_SIZE);

    state->FEC_nbr++;
}

int send_FEC(sender_state_t *state, int socket_fd)
{
    if (state->last_pkt_sent == CLOSING_PKT)
    {
        DEBUG("Sending LAST FEC pkt with seqnum: %d", pkt_get_seqnum(state->FEC));
    }
    else 
    {
        DEBUG("Sending FEC pkt with seqnum: %d", pkt_get_seqnum(state->FEC));
    }
    
    char packet_to_be_sent[PKT_MAX_LEN];
    size_t len = PKT_MAX_LEN;
    pkt_status_code pkt_status = pkt_encode(state->FEC, packet_to_be_sent, &len);
    if (pkt_status != PKT_OK) 
    {
        ERROR("pkt_encode failed with status: %d", pkt_status);
        return -1;
    }
    
    ssize_t sent = send(socket_fd, packet_to_be_sent, len, 0);
    if (sent == -1) 
    {
        ERROR("The sending (using the function send from <sys/socket.h>) of the pkt failed");
        return -1;
    }
    memset(state->FEC, 0, sizeof(pkt_t));
    state->FEC_nbr = 0;
    state->stats->fec_sent++;
    return 0;
}

int read_and_send(sender_state_t *state, int sending_fd, int socket_fd)
{
    // Checking whether I need to send a PTYPE_FEC or PTYPE_DATA
    if ((state->fec_enabled) && ((state->FEC_nbr == 4) || ((state->last_pkt_sent == CLOSING_PKT) && (state->FEC_nbr > 0))))
    {
        return send_FEC(state, socket_fd);
    }
    else
    {
        state->s_window_size--;
        state->r_window_size--;

        pkt_t *pkt = pkt_new(); 
        ssize_t nbr_byte_read;
        bool is_EOF = is_it_EOF(sending_fd);

        // The file has already been read but we need to send an empty DATA pkt with length = 0
        if (is_EOF) 
        {
            nbr_byte_read = 0;
            state->last_pkt_sent = CLOSING_PKT;
            DEBUG("The CLOSING pkt is being sent !");
        }
        else 
        {
            nbr_byte_read = read(sending_fd, pkt->payload, MAX_PAYLOAD_SIZE);
        }

        pkt_set_type(pkt, PTYPE_DATA);
        pkt_set_tr(pkt, 0);
        pkt_set_window(pkt, state->s_window_size);
        pkt_set_length(pkt, nbr_byte_read);
        pkt_set_seqnum(pkt, state->next_seqnum);
        // Sending a specific timestamp to let the receiver know, it is the second to last pkt
        is_EOF = is_it_EOF(sending_fd);
        if (state->last_pkt_sent == RANDOM_PKT && is_EOF)
        {
            pkt_set_timestamp(pkt, SECOND_TO_LAST_PKT);
            state->last_pkt_sent = LAST_DATA_PKT;
            DEBUG("The LAST PTYPE_DATA is being sent !");
        }
        else
        {
            pkt_set_timestamp(pkt, 0);
        }
        
        // We set the TR to 0 in order to calculcate the CRC on the header
        char modified_header[8];
        memcpy((void *) &modified_header, (void *) &pkt->header, 8);
        modified_header[0] = modified_header[0] & TR_SETTER_TO_ZERO;
        uint32_t crc1 = htonl(calculate_crc(modified_header, 8));
        pkt_set_crc1(pkt, crc1);

        uint32_t crc2 = htonl(calculate_crc((char *) pkt->payload, (uint32_t) pkt_get_length(pkt)));
        pkt_set_crc2(pkt, crc2);

        if (send_pkt(state, pkt, state->head, socket_fd) == -1) return -1;
        if (state->fec_enabled)
        {
            construct_FEC(state, pkt);
        }
        struct timeval timer;
        gettimeofday(&timer, NULL);
        state->timers_first_send[state->head] = timer;

        state->map_seqnum_to_buffer_place[pkt_get_seqnum(pkt)] = state->head;
        state->head = (state->head + 1) % WINDOW_SIZE;

        // Careful we need to convert to uint16_t to avoid overflow
        state->next_seqnum = (uint8_t) (((uint16_t) state->next_seqnum) + 1) % SEQNUM_RANGE;
        return 0;
    }
}