src/vfer_control.c

Go to the documentation of this file.

/*
 * Copyright 2005, 2006, Internet2
 * Legal conditions are in file LICENSE
 * (MD5 = c434f2e53b8089d8b4d0172c7ce07360).
 */
/**
 *
 * @file   vfer_control.c
 * @author Ivan Beschastnikh
 * @brief  Implements flow controls, fragmentation, sending, receiving, and acking logic.
 *
 * This file implements the brains of VFER
 *
 * -     05/13/06        ivan            Integrated most of logic from Control_Accept into ControlT, added new state: CONN_ACCEPTING
 * -     05/11/06        ivan            Removed the server and connect threads and made ControlT thread handle listening and connecting sockets
 * -     05/04-06/06     ivan            Removed reader and writer threads and made ControlT killable. Many other changes, notably to CC.
 * -     02/19/06        ivan            Fixed shift by 32 bug in SendAllData
 * -     02/18/06        ivan            Factored out CC managing functionality into ccontrol.c
 * -     02/18/06        ivan            Factored out frame managing functionality to datagram.c and datagram.h, corrected relfun for accepted sockets bug
 * -     02/05/06        ivan            Corrected some minor CC bugs to get test_rcp and test_rcpd working again
 * -     02/04/06        ivan            Updated for new return codes of Writer and Reader (in writer.c and reader.c)
 * -     12/30/05        ivan            Changed to use byte accounting instead of packet accounting for cc window
 * -     12/29/05        ivan            Exponential ramp in window size & Flow control
 * -     12/29/05        ivan            Control_Close_Conn now acts correctly in all cases- supports nonblocking sockets and timeouts. Added ccack receipt action to purge old frames.
 *                                       added Control_ConnectT to act as thread \ nonthread func for nonblocking \ blocking vfer_connect calls from api.c ;
 *                                       fixed state transitions for disconnected and closed states
 * -     12/28/05        ivan            static function defs which modularize the scheduling ; data_timeout and more
 * -     10/30/05        ivan            renamed to control.c
 * -     10/15/05        ivan            added close_req and reset packet sending
 * -     09/01/05        ivan            modified to use new packet structures
 * -     08/05/05        ivan            changed C_Control_New_Conn to connect to reader and writer
 * -     07/28/05        ivan            Send,Recv added: fragmentation + hooks to reader,writer
 * -     07/27/05        ivan            ControlT and many list internal functions added
 * -     07/21/05        ivan            created
 */

#include "vfer_control.h"
#include "vfer_ccontrol.h"

static pthread_t ControlT_tid;

/* private function prototypes */
inline  static int              Control_Write           (packet* p, int fd);
        static int              Control_RecvAll         (vfer_sock* sock);
inline  static int              Control_SendCtl         (vfer_sock* sock, packet* p);
inline  static int              Control_Send_Ack        (vfer_sock* s);
        static int              Control_SendAllData     (vfer_sock* sock);
        static void             ControlT                (void* ptr);
        static void             Control_Remove_Socket   (vfer_sock* sock);
        static int              Unconnect_Rebind        (vfer_sock* sock);


/******************
 * Public Functions
 ******************/

/**
 * Returns the top connection on te accept queue for a socket.
 *
 * This function changes the state of ret_sock to CONN_ACCEPTING and
 * blocks in a cond_wait until ControlT threads goes through the
 * handshake, changes the state of ret_sock to CONN_CONNECTED (or
 * something else on error) and signals to this thread to wake up.
 *
 * @param sock a pointer to a socket structure
 * @param ret_sock a pointer to a socket structure that is 'connected' as a result of this call
 * @return
 *      - 0 on success
 *      - VFER_BADSOCK bad vfer_sock
 *      - VFER_WOULDBLOCK socket is non-blocking and there are no waiting connections
 *      - VFER_IMPL underlying socket error
 */
int Control_Accept(vfer_sock* sock, vfer_sock* ret_sock)
{
        int new_port;                   /* contains the port we will use for the rest of connection (not handshake) */
        struct sockaddr_in sa;
        char buf[INET_ADDRSTRLEN];      /* holds remote addr as string */
        socklen_t socklen;
        struct timespec timedwait_timeout; /* used as argument to pthread_cond_timedwait */
        struct timeval start_time;
        int ret;

        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Accept", "Entered with listening sock->fd[%d]", sock->fd);

        ret_sock->fd = socket(AF_INET, sock->type, 0);
        if (ret_sock->fd == -1) {
                perror("server_accept :: socket");
                return VFER_IMPL;
        }

        if (fcntl(ret_sock->fd, F_SETFD, O_NONBLOCK) == -1) {
                perror("server_accept :: fcntl");
                return VFER_IMPL;
        }

        /* we bind the local port, wait for a connection, and
           explicitly connect to the remote address */

        bzero((char*)(&sa), sizeof(sa));
        sa.sin_port             = htons(INADDR_ANY);
        sa.sin_addr.s_addr      = htonl(INADDR_ANY);
        sa.sin_family           = AF_INET;

        if (bind(ret_sock->fd, (SA*)(&sa), sizeof(sa)) < 0) {
                perror("server_accept: bind for the new socket failed");
                close(ret_sock->fd);
                return VFER_IMPL;
        }

        socklen = sizeof(sa);
        if (getsockname(ret_sock->fd, (SA*)(&sa), &socklen) == -1) {
                perror("server_accept: getsockname for new socket failed");
                close(ret_sock->fd);
                return VFER_IMPL;
        }

        new_port = ntohs(sa.sin_port);

        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Accept", "New port socket: %s.%d",
                    inet_ntop(AF_INET, &(sa.sin_addr), buf, INET_ADDRSTRLEN),
                    new_port);

        ret_sock->state = CONN_ACCEPTING;
        ret_sock->listening_sock = sock;
        ret_sock->last_response.tv_sec = 0;
        ret_sock->response = NULL;
        ret_sock->sa = sa;
        ret_sock->response_tries = MAX_RESPONSE_TRIES;

        pthread_mutex_unlock(&(ret_sock->mutex));
        Control_Register_Socket(ret_sock);
        pthread_mutex_lock(&(ret_sock->mutex));
        /***************/
        GET_TIME_OF_DAY(&(start_time));

        if (sock->accept_timeout != 0) {
                /* compute the timeout for cond_timedwait */
                timedwait_timeout.tv_sec = start_time.tv_sec + (sock->accept_timeout / 1000000); /* seconds */
                timedwait_timeout.tv_nsec = (start_time.tv_usec + sock->accept_timeout - (sock->accept_timeout / 1000000)) * 1000; /* nanoseconds */
                do {
                        ret = pthread_cond_timedwait(&(ret_sock->cond), &(ret_sock->mutex), &(timedwait_timeout));
                        /* pthread_cond_wait(&(ret_sock->cond), &(ret_sock->mutex)); */
                        if (ret == ETIMEDOUT) {
                                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Accept", "timedout while trying to accept a new connection");
                                ret_sock->err = VFER_TIMEOUT;
                                close(ret_sock->fd);
                                return VFER_TIMEOUT;
                        }
                } while (ret_sock->state == CONN_ACCEPTING);
        } else {
                do {
                        pthread_cond_wait(&(ret_sock->cond), &(ret_sock->mutex));
                } while (ret_sock->state == CONN_ACCEPTING);
        }

        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Accept", "reached non conn_accepting state[%d]", ret_sock->state);

        if (ret_sock->state != CONN_CONNECTED) {
                close(ret_sock->fd);
                return ret_sock->err;
        }
        return 0;
} /* Control_Accept() */


/**
 * This function changes the state of the socket to CONN_CONNECTING
 * and either return immediately if the socket is non blocking or
 * blocks waiting on a condition variable for the ControlT thread to
 * signal on it indicating that the socket is connected and is in
 * CONN_CONNECTED state. Follows CONNECT_TIMEOUT (defined in
 * globals.h) for timing out.

 * @param sock is a pointer to a socket structure on which to connect.
 * The socket must have its addr field set to the address to connect
 * to.
 * @return
 *      - 0  on connection succes
 *      - -1 on connection failure- system connect() call failed
 *      - -2 if the socket is nonblocking- the caller can find out about
 *           the success or failure of the connection via vfer_select()-
 *           NOTE: this is broken right now.
 */
int Control_Connect(vfer_sock* sock) {
        SA *serv_addr;
        int addrlen;
        struct timespec timedwait_timeout; /* used as argument to pthread_cond_timedwait */
        struct timeval start_time;
        int ret;

        serv_addr = (SA*)(&(sock->addr));
        addrlen = sizeof(sock->addr);

        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Connect", "Entered sock->fd[%d]", sock->fd);

        /* try to connect the udp socket to proceed with the handshake*/
        if (connect(sock->fd, serv_addr, addrlen) < 0)  {
                perror("Control_Connect :: connect");
                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Connect", "connect failed");
                sock->err = VFER_NOCONN;
                return -1;
        }
        memcpy(&sock->addr, serv_addr, addrlen);
        sock->connect_tries = 0;
        sock->request = NULL;
        sock->state = CONN_CONNECTING;
        bzero(&(sock->last_request), sizeof(struct timeval));
        pthread_mutex_unlock(&(sock->mutex));
        /*************/

        Control_Register_Socket(sock);

        if (SOCK_OPT_ISSET(sock, VFERSO_NONBLOCK)) {
                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Connect", "non blocking socket, connection in progress");
                return -2;
        }

        /*************/
        /* lock the socket's mutex prior to calling pthread_cond_wait */
        pthread_mutex_lock(&(sock->mutex));
        GET_TIME_OF_DAY(&(start_time));
        if (sock->connect_timeout != 0) {
                /* compute the timeout for cond_timedwait */
                timedwait_timeout.tv_sec = start_time.tv_sec + (sock->connect_timeout / 1000000); /* seconds */
                timedwait_timeout.tv_nsec = (start_time.tv_usec + sock->connect_timeout - (sock->connect_timeout / 1000000)) * 1000; /* nanoseconds */
                do {
                        // DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Connect", "waiting on sock->mutex until sock's state changes from CONN_CONNECTING");
                        pthread_cond_timedwait(&(sock->cond), &(sock->mutex), &(timedwait_timeout));
                        /* pthread_cond_wait(&(sock->cond), &(sock->mutex)); */
                        if (ret == ETIMEDOUT) {
                                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Connect", "timedout trying to connect");
                                sock->err = VFER_TIMEOUT;
                                return -1;
                        }
                } while (sock->state == CONN_CONNECTING);
        } else {
                do {
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Connect", "waiting on sock->mutex until sock's state changes from CONN_CONNECTING");
                        pthread_cond_wait(&(sock->cond), &(sock->mutex));
                } while (sock->state == CONN_CONNECTING);
        }

        if (sock->state != CONN_CONNECTED) {
                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Connect", "reached a non conn_connecting state[%d]", sock->state);
                return -1;
        }

        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Connect", "reached connected state");
        return 0;
} /* Control_Connect() */

/**
 * Sets the socket's state to DISCONNECTED and blocks (if blocking
 * socket) until ControlT thread set socket's state to DISCONNECTED or
 * returns if nonblocking socket Socket's mutex must be locked before
 * calling this function Socket's state must be CONN_CONNECTED before
 * calling this function, this function does not unlock the mutex of
 * the socket
 *
 * @param sock is a socket pointer
 * @return
 *      0  on success
 *      -2 if socket is nonblocking, will disconnect eventually
 */
int Control_Close(vfer_sock* sock) {
        struct timespec timedwait_timeout; /* used as argument to pthread_cond_timedwait */
        struct timeval start_time;
        int ret;

        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Close", "Entered sock->fd[%d]", sock->fd);
        sock->state = CONN_DISCONNECTING;
        if (SOCK_OPT_ISSET(sock, VFERSO_NONBLOCK)) {
                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Close", "close() : VFER_INPROGRESS, send_queue[%d]", sock->send_frames.num);
                return -2;
        }

        GET_TIME_OF_DAY(&(start_time));
        if (sock->close_timeout != 0) {
                /* compute the timeout for cond_timedwait */
                timedwait_timeout.tv_sec = start_time.tv_sec + (sock->close_timeout / 1000000); /* seconds */
                timedwait_timeout.tv_nsec = (start_time.tv_usec + sock->close_timeout - (sock->close_timeout / 1000000)) * 1000; /* nanoseconds */
                do {
                        pthread_cond_timedwait(&(sock->cond), &(sock->mutex), &(timedwait_timeout));
                        if (ret == ETIMEDOUT) {
                                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Close", "timedout trying to close");
                                sock->err = VFER_TIMEOUT;
                                return -1;
                        }
                } while (sock->state != CONN_DISCONNECTED);
        } else {
                do {
                        pthread_cond_wait(&(sock->cond), &(sock->mutex));
                } while (sock->state != CONN_DISCONNECTED);
        }


        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Close", "reached disconnected|local_closed|err state");
        return 0;
} /* Control_Close() */

/**
 * Sends data on a socket. This function is called by the client via
 * the api vfer_send call. The socket should be locked before the call
 * to this function. This function unlocks the socket before returning.
 *
 * @param sock          a socket pointer
 * @param buffer        a pointer to data to send
 * @param len           the length of data in this buffer to send
 * @return
 *      -1 on error
 *      size of the buffer sent on succcess
 */
size_t Control_Send(vfer_sock* sock, const void* buffer, int len) {
        struct timespec timedwait_timeout; /* used as argument to pthread_cond_timedwait */
        struct timeval start_time;
        int ret;

        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Send", "Entered with sock->fd[%d] buffer_len[%d]", sock->fd, len);
        if (len > MAX_SEND_WINDOW) {
                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Send", "buffer length is more than MAX_SEND_WINDOW");
                return -1;
        }

        if (sock->state != CONN_CONNECTED) {
                pthread_mutex_unlock(&(sock->mutex));
                SET_ERROR_RETURN (sock, VFER_INVAL, -1);
        }

        if ((sock->send_frames.length + len) > MAX_SEND_WINDOW) {
                /* return if this is a non blocking socket */
                if ((sock->opts & VFERSO_NONBLOCK) == 1) {
                        pthread_mutex_unlock(&(sock->mutex));
                        SET_ERROR_RETURN (sock, VFER_WOULDBLOCK, -1);
                }
                /* block until we have enough space in datagram list to enqueue the buffer */
                GET_TIME_OF_DAY(&(start_time));
                if (sock->snd_timeout != 0) {
                        /* compute the timeout for cond_timedwait */
                        timedwait_timeout.tv_sec = start_time.tv_sec + (sock->snd_timeout / 1000000); /* seconds */
                        timedwait_timeout.tv_nsec = (start_time.tv_usec + sock->snd_timeout - (sock->snd_timeout / 1000000)) * 1000; /* nanoseconds */
                        do {
                                ret = pthread_cond_timedwait(&(sock->cond), &(sock->mutex), &(timedwait_timeout));
                                /* pthread_cond_wait(&(sock->cond), &(sock->mutex)); */
                                if (ret == ETIMEDOUT) {
                                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Send", "timedout while sending a frame");
                                        pthread_mutex_unlock(&(sock->mutex));
                                        SET_ERROR_RETURN (sock, VFER_TIMEOUT, -1);
                                }

                        } while ((sock->send_frames.length + len) > MAX_SEND_WINDOW && sock->state == CONN_CONNECTED);
                } else {
                        do {
                                pthread_cond_wait(&(sock->cond), &(sock->mutex));
                        } while ((sock->send_frames.length + len) > MAX_SEND_WINDOW && sock->state == CONN_CONNECTED);
                }

                if (sock->state != CONN_CONNECTED) {
                        pthread_mutex_unlock(&(sock->mutex));
                        SET_ERROR_RETURN (sock, VFER_INVAL, -1);
                }
        }

        if (Datagram_Enqueue(&sock->send_frames, buffer, len, sock->l_frame_num) == -1) {
                pthread_mutex_unlock(&(sock->mutex));
                SET_ERROR_RETURN (sock, VFER_BADSOCK, -1);
        }
        sock->l_frame_num++;
        sock->stats.data_bytes_sender_sent += len;

        pthread_mutex_unlock(&(sock->mutex));
        /****************/
        return (size_t)(len);
} /* Control_Send() */

/**
 * Receive data on a socket. This is a function that is called by the
 * client via the api call vfer_recv. The socket's mutex should be
 * locked before calling this function. This function unlocks the
 * socket's mutex before returning. Socket's state must be
 * conn_connected/conn_disconnected/conn_disconnecting on entry
 *
 * @param sock          a socket pointer
 * @param buffer        a pointer to buffer to which to write received data
 * @param len           buffer size
 * @return
 *      - size of the buffer received on succcess
 *      - VFER_BADSOCK bad vfer_sock
 *      - VFER_TIMEOUT timed out while waiting to receive
 *      - VFER_UNCONN sock is not connected
 *      - VFER_INVAL invalid buf or len
 *      - VFER_WOULDBLOCK socket is non-blocking and requested operation would block
 */
int Control_Recv(vfer_sock* sock, void* buffer, int len) {
        int ret;
        struct timespec timedwait_timeout; /* used as argument to pthread_cond_timedwait */
        struct timeval start_time;

        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Recv", "Entered sock->fd[%d] ; buffer len[%d]", sock->fd, len);

        if (sock->recvd_frames.num == 0) {
                /* in order to receive more recvd frames, we need to be in connected state
                   note: to recv already recvd frames, this isn't true */

                if (sock->state != CONN_CONNECTED) {
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Recv", "Returning 0 as socket state doesn't allow receiving");
                        pthread_mutex_unlock(&(sock->mutex));
                        return VFER_UNCONN;
                }
                /* block or return error if nonblocking opt is set */
                if ((sock->opts & VFERSO_NONBLOCK) == 1) {
                        pthread_mutex_unlock(&(sock->mutex));
                        return VFER_WOULDBLOCK;
                }

                GET_TIME_OF_DAY(&(start_time));

                if (sock->rcv_timeout != 0) {
                        /* compute the timeout for cond_timedwait */
                        timedwait_timeout.tv_sec = start_time.tv_sec + (sock->rcv_timeout / 1000000); /* seconds */
                        timedwait_timeout.tv_nsec = (start_time.tv_usec + sock->rcv_timeout - (sock->rcv_timeout / 1000000)) * 1000; /* nanoseconds */
                        while (sock->recvd_frames.num == 0 && sock->state == CONN_CONNECTED) {
                                ret = pthread_cond_timedwait(&(sock->cond), &(sock->mutex), &(timedwait_timeout));
                                if (ret == ETIMEDOUT) {
                                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Recv", "timedout while receiving a new frame");
                                        pthread_mutex_unlock(&(sock->mutex));
                                        return VFER_TIMEOUT;
                                }
                        }
                } else {
                        while (sock->recvd_frames.num == 0 && sock->state == CONN_CONNECTED) {
                                ret = pthread_cond_wait(&(sock->cond), &(sock->mutex));
                        }
                }

                if (sock->state != CONN_CONNECTED) {
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Recv", "Returning 0 as socket state doesn't allow receiving");
                        pthread_mutex_unlock(&(sock->mutex));
                        return VFER_UNCONN;
                }
        }

        /* if we are here there is at least one frame in the recv_frames list */
        ret = Datagram_Dequeue(&sock->recvd_frames, buffer, len);
        if (ret == -1) {
                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Recv", "Can't return frame: buffer too small");
                pthread_mutex_unlock(&(sock->mutex));
                return VFER_INVAL;
        }
        pthread_mutex_unlock(&(sock->mutex));
        /****************/
        return ret;
} /* Control_Recv() */


/**
 * This is a helper function that makes the socket passed as argument
 * 'active' so that the ConnectT thread will pay attention to it. It
 * also maintains the sockets.max_fd and the sockets.fds vars which
 * are used as args to the select() call in ControlT thread
 *
 * @param sock socket pointer to socket we would like listed as active
 */
void Control_Register_Socket(vfer_sock* sock) {
        sigset_t set;

        /*************/
        pthread_mutex_lock(&(sockets.mutex));
        sockets.num_active++;

        FD_SET(sock->fd, &(sockets.fds));
        if (sockets.max_fd < sock->fd) {
                sockets.max_fd = sock->fd;
        }

        /* printf("register_socket new num_active[%d]\n", sockets.num_active); */
        if (sockets.num_active == 1) {
                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Register_Socket", "Starting ControlT thread, num_active = 1");
                if (pthread_create(&ControlT_tid, NULL, (void (*))ControlT, NULL) != 0) {
                        ERROR_PRINT("control.c", "Control_Register_Socket", "ERROR: Couldn't create ControlT thread");
                } else {
                        pthread_sigmask(SIG_BLOCK, NULL, &set);
                        sigaddset(&set, SIGINT);
                        pthread_sigmask(SIG_BLOCK, &set, NULL);
                        pthread_detach(ControlT_tid);
                }
        }
        pthread_mutex_unlock(&(sockets.mutex));
        /*************/
} /* Control_Register_Socket() */

/**
 * This function removes socket sock from the 'active' list of sockets
 * that are observed by the ControlT thread. This thread takes care to
 * maintain sockets.max_fd and sockets.fds vars
 *
 * @param sock is a socket pointer that will be removed from the 'active' list of sockets
 */
void Control_Unregister_Socket(vfer_sock* sock) {
        int i, j;

        /****************/
        pthread_mutex_lock(&(sockets.mutex));
        FD_CLR(sock->fd, &(sockets.fds));
        sockets.num_active--;
        
        /* close icmp socket */
        if(sock->icmp_fd != -1){
                close(sock->icmp_fd);
        }
        
        /* printf("unregister_socket new num_active[%d]\n", sockets.num_active); */
        i = sockets.max_fd;
        if (sock->fd == i) {
                /* recompute max_fd */
                sockets.max_fd = 0;
                i = j = 0;
                while (j < sockets.num_active) {
                        pthread_mutex_lock(&(sockets.s[i].mutex));
                        if (((sockets.s[i].state == CONN_DISCONNECTING) ||
                             (sockets.s[i].state == CONN_CONNECTED) ||
                             (sockets.s[i].state == CONN_CONNECTING) ||
                             (sockets.s[i].state == CONN_ACCEPTING) ||
                             (sockets.s[i].state == CONN_LISTENING)) &&

                            (sockets.max_fd < sockets.s[i].fd)) {
                                j++;
                                sockets.max_fd = sockets.s[i].fd;
                        }
                        pthread_mutex_unlock(&(sockets.s[i].mutex));
                        i++;
                }
        }
        pthread_mutex_unlock(&(sockets.mutex));
        /****************/
} /* Control_Unregister_Socket() */

/*************************************************************************
 * Private (Internal) functions
 *************************************************************************/

/**
 * Writes the iov to an fd and then deallocs the iov (different for
 * DATA, DATAACK iovs).
 *
 * @return
 *       0      on success
 *       errno  on error (from Packet_Write(), see packet.c)
 */
inline static int Control_Write(packet* p, int fd) {
        int ret;

        ret = Packet_Write(p, fd, NULL, 0);
        switch (p->type) {
        case DATA:
                /* we do not want to release the data portion of this packet */
                RELEASE_IOV(p->iov, 1);
                RELEASE(p, sizeof(packet));
                break;
         default:
                 RELEASE_PACK(p);
                 break;
        }

        if (ret != 0) {
                if (ret == EAGAIN) return 0; /* as if we were able to send it */
                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Write", "Packet_Write returned error[%d]", ret);
                return -1;
        }
        return 0;
} /* Control_Write() */

/**
 * Receives all the packets on a given socket
 * NOTE: socket's mutex must be locked before calling this function
 *
 * @param sock socket on which to receive packets
 * @return
 *      -1 on error
 *      0 if received no packets
 *      1 if received only data packets
 *      2 if received only control packets
 *      3 if received both data and control packets
 */
static int Control_RecvAll(vfer_sock* sock) {
        frame_link* frame_l;
        frame_link* frame_l2;
        frame_link* frame_prev;
        packet* p;
        packet* ack;
        uint32_t pframe_num;    /* packet's frame num- used when receiving a data packet */
        int i;
        int ret_val;
        int is_phantom;         /* used upon receiving a data packet */
        ccontrol_t* c;
        int n;
        interval_t *interval_ptr, *interval_ptr2;

        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "Entered with sock->fd[%d]", sock->fd);
        ret_val = 0;
        c = &(sock->ccontrol);
        
        while ((Packet_Read_ICMP(sock->icmp_fd, &(sock->pmtu)) >= 0) && ((n = Packet_Read(sock->fd, &p, NULL, NULL)) != EAGAIN)) {
                switch (n) {
                case -1:
                case -2:
                case -3:
                case -4:
                        /* malformed packet errors */
                        continue;
                }
                if (n != 0) {
                        /* stop processing this socket immediately */
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "Packet_Read returned error[%d]", n);
                        return -1;
                }

                /* if we are the client, change the sign of the delay measure set by packet.c
                   this also lets us compute the RTT in the same way for client\server later */
                if (SOCK_OPT_ISSET(sock, VFERSO_CLIENT_SOCK)) {
                        p->delay *= -1;
                }

                DEBUG_PACK(DEBUG_CTL, "control.c", "Control_RecvAll", p, "Received packet on sock->fd[%d]", sock->fd);

                /* note that we count packets & bytes that might be malformed and later ignored */
                sock->stats.packets_recvd++;
                /* the whole packet is stored in p->iov[0] */
                sock->stats.bytes_recvd += p->iov[0].iov_len;

                /* handle the packet depending on its type */
                switch (p->type) {
                case REQUEST:
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "Received REQUEST during wrong state ; ignoring");
                        break;
                case RESPONSE:
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "Received RESPONSE during wrong state ; ignoring");
                        /* this needs to do more than just send an ack back, particularly authenticate the response seq number
                           and zero out all the indicator bits so that all the data can be resent, also if we received a packet
                           such as ack or data after previous response implies this one is obviously a decoy and should be ignored */
                        ack = Packet_Form_CCAck(0, sock->request->seq + 1, p->seq, 0,
                                                p->u.res.init_frame_num, p->u.res.init_frame_num,
                                                0, p->delay, 0, 0, NULL);
                        if (Packet_Write(ack, sock->fd, NULL, 0) != 0) {
                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Packet_Write failed");
                                sock->err       = VFER_NOCONN;
                                RELEASE_PACK(ack);
                        }
                        break;
                case CC_ACK:
                        CC_Recvd_Ack(p, sock);

                        /* check if the newest_frame seen field is not the last_frame_num
                           for send_frames list, reset all frames >= newest_frame value bit vectors to 0 */

                        /* remove frames from the send queue that are no longer of interest (ie. all frames whose frame_num < last_frame_num in ccack) */
//                      printf("first[%d] oldest_acked[%d] send_frames_len1[%d] ",
//                             (sock->send_frames.first == NULL ? -27 : sock->send_frames.first->frame.frame_num), p->u.ccack.oldest_frame, sock->send_frames.length);

                        while (sock->send_frames.first != NULL &&
                               sock->send_frames.first->frame.frame_num < p->u.ccack.oldest_frame) {
                                frame_l = sock->send_frames.first;
                                /* DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "removing frame[%d] implicitly by oldest_frame[%d] in ack",
                                   frame_l->frame.frame_num, p->u.ccack.oldest_frame); */
                                Datagram_Remove (&(sock->send_frames), NULL, frame_l);
                                
                                /* PMTU: If probe in completed frame, increase MTU. */
                                if(sock->pmtu.probe_state == PROBE_SENT && sock->pmtu.frame_num == frame_l->frame.frame_num){
                                        c->mtu = *(sock->pmtu.probe_size);
                                        sock->pmtu.search_low = sock->pmtu.probe_size;
                                        if(sock->pmtu.search_low == sock->pmtu.search_high){
                                                /* PMTU found */
                                                sock->pmtu.probe_state = MTU_FOUND;
                                                struct timeval tv;
                                                GET_TIME_OF_DAY(&tv);
                                                sock->pmtu.next_probe_time = tv.tv_sec + PMTUD_TIME_INTERVAL;
                                                DEBUG_PMTUD_PRINT(DEBUG_PMTUD, "control.c", "Control_RecvAll()",
                                                                  "PMTU of size %d found [oldest frame]. Next probe to be begin at %lu.", c->mtu, (long int)sock->pmtu.next_probe_time);
                                        }else{
                                                sock->pmtu.probe_size++;
                                                sock->pmtu.probe_state = NO_PROBE_SENT;
                                                DEBUG_PMTUD_PRINT(DEBUG_PMTUD, "control.c", "Control_RecvAll()", "Increasing PMTU to size %d. [oldest frame]", c->mtu);
                                        }
                                }
                                
                                RELEASE(frame_l->frame.data, frame_l->frame.len);
                                RELEASE(frame_l, sizeof(frame_link));
                        }

//                      printf("send_frames_len2[%d]", sock->send_frames.length);

                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "Received CC_ACK ; newest_frame[%d] ; send_frames.first_frame_num[%d]",
                                    p->u.ccack.newest_frame, (sock->send_frames.first == NULL ? -0 : sock->send_frames.first->frame.frame_num));

                        /* CLEAR the unacked_data var ONLY if this ack acknowledges ALL of the past data we've sent
                           if it does not then do NOT reset it! */
                        if (p->u.ccack.last_seq == (sock->ccontrol.s_last_seq - 1)) {
//                              printf("clearing last_unacked_data (last_seq onack matches our last_seq)\n");
                                c->s_last_unacked_data.tv_sec = 0; /* clear the unacked data var */
                        }

                        /* update the frag_indicator for frames that are outside of the range seen by the received as implied by this ccack */
                        if (ret_val < 2) ret_val+=2;

                        if (p->u.ccack.interval_count == 0) {
                                break;
                        }
                        /* update the missing data ranges for the datagram specified by this ack */
                        /* for now don't care what the sequence number is, but later make sure that
                           if a previous ack with larger seq number for this datagram was received then
                           we don't pay attention to this one */
                        // printf("updating from receved missing ranges in ACK!\n");
                        frame_l = sock->send_frames.first;
                        while (frame_l != NULL) {
                                if (frame_l->frame.frame_num == p->u.ccack.intervals_frame_num && frame_l->frame.done == 0) {
                                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "Updating intervals for frame[%d]", p->u.ccack.intervals_frame_num);
                                        
                                        if (p->u.ccack.interval_count == 1 &&
                                            ntohl(p->u.ccack.raw_intervals[0]) == 0 &&
                                            ntohl(p->u.ccack.raw_intervals[1]) == 0) {
                                                /* a (0,0) interval implies that none of the frame has been received
                                                   and thus the endhost doesn't even know the size of the frame */
                                                p->u.ccack.raw_intervals[1] = htonl(frame_l->frame.len);
                                        }
                                        
                                        /* PMTU: If only missing interval and entirely consists of probe data, then stop searching upward */
                                        int net_interval_count = p->u.ccack.interval_count - frame_l->frame.missing_data.count;
                                        if(sock->pmtu.probe_state == PROBE_SENT && frame_l->frame.frame_num == sock->pmtu.frame_num && net_interval_count > 0){
                                                i = 0;
                                                while(i < p->u.ccack.interval_count){
                                                        int start = ntohl(p->u.ccack.raw_intervals[2*i]);
                                                        int end = ntohl(p->u.ccack.raw_intervals[(2*i) + 1]);
                                                        if(start == sock->pmtu.data_offset 
                                                                && end == (sock->pmtu.data_offset + *(sock->pmtu.probe_size))
                                                                && net_interval_count == 1){
                                                                        sock->pmtu.search_high = sock->pmtu.probe_size;
                                                                        sock->pmtu.probe_state = MTU_FOUND;
                                                                        struct timeval tv;
                                                                        GET_TIME_OF_DAY(&tv);
                                                                        sock->pmtu.next_probe_time = tv.tv_sec + PMTUD_TIME_INTERVAL;
                                                                        DEBUG_PMTUD_PRINT(DEBUG_PMTUD, "control.c", "Control_RecvAll()",
                                                                                          "PMTU of size %d found [bit vector indicated failure]. Next probe to be begin at %lu.",
                                                                                          c->mtu, (long int)sock->pmtu.next_probe_time);
                                                                        break;
                                                        }else if(start <= sock->pmtu.data_offset 
                                                                && end >= (sock->pmtu.data_offset + *(sock->pmtu.probe_size))){
                                                                        /* PMTU:If probe missing but other data also missing, then probe should be resent */
                                                                        DEBUG_PMTUD_PRINT(DEBUG_PMTUD, "control.c", "Control_RecvAll()", "Probe needs to be resent");
                                                                        sock->pmtu.probe_state = NO_PROBE_SENT;
                                                                        break;
                                                        }
                                                        i++;
                                                }
                                                
                                                /* if went through all intervals but probe not missing and state not changed, then probe must have been received. 
                                                   In this case we can increase the path MTU. NOTE: Most likely the CC_ACK oldest frame will handle this but in 
                                                   case this appears first, can increase in this manner. The probe state will prevent any conflicts */
                                                   if(sock->pmtu.probe_state == PROBE_SENT){
                                                                c->mtu = *(sock->pmtu.probe_size);
                                                                sock->pmtu.search_low = sock->pmtu.probe_size;
                                                                if(sock->pmtu.search_low == sock->pmtu.search_high){
                                                                        /* PMTU found */
                                                                        sock->pmtu.probe_state = MTU_FOUND;
                                                                        struct timeval tv;
                                                                        GET_TIME_OF_DAY(&tv);
                                                                        sock->pmtu.next_probe_time = tv.tv_sec + PMTUD_TIME_INTERVAL;
                                                                        DEBUG_PMTUD_PRINT(DEBUG_PMTUD, "control.c", "Control_RecvAll()",
                                                                                          "PMTU of size %d found [oldest frame]. Next probe to begin at %lu.", c->mtu, (long int)tv.tv_sec);
                                                                }else{
                                                                        sock->pmtu.probe_size++;
                                                                        sock->pmtu.probe_state = NO_PROBE_SENT;
                                                                        DEBUG_PMTUD_PRINT(DEBUG_PMTUD, "control.c", "Control_RecvAll()", "Increasing PMTU to size %d. [oldest frame]", c->mtu);
                                                                }
                                                   }
                                                   
                                        } /* End PMTU */

                                        /* update the intervals list based on the list that's carried in the ack packet */
                                        interval_ptr = frame_l->frame.missing_data.head;
                                        interval_ptr2 = interval_ptr;
                                        i = 0;
                                        /* interval_ptr is treated as the previous elem in this loop except at start */
                                        // printf("old_count[%d] vec_ack_count[%d]\n", frame_l->frame.missing_data.count, p->u.ccack.interval_count);
                                        while (i < frame_l->frame.missing_data.count && i < p->u.ccack.interval_count) {
                                                // printf("1 ");
                                                // printf("old[%d %d] ", interval_ptr->start, interval_ptr->end);
                                                interval_ptr->start = ntohl(p->u.ccack.raw_intervals[2*i]);
                                                interval_ptr->end = ntohl(p->u.ccack.raw_intervals[(2*i)+1]);
                                                // fflush(stdout);
                                                // printf("new[%d %d] ", interval_ptr->start, interval_ptr->end);
                                                // fflush(stdout);

                                                interval_ptr2 = interval_ptr;
                                                interval_ptr = interval_ptr->next;
                                                i++;
                                        }
                                        
                                        interval_ptr = interval_ptr2;
                                        /* interval_ptr points to the last updated interval in the missing_data intervals list */
                                        if (i == frame_l->frame.missing_data.count &&
                                            i < p->u.ccack.interval_count) {
                                                /* make new intervals at end of interval list */
                                                for (; i < p->u.ccack.interval_count; i++) {
                                                        // printf("2 ");
                                                        if (frame_l->frame.missing_data.head == NULL) {
                                                                /* missing_data is empty, therefore have to start from head of list */
                                                                // printf(". ");
                                                                ALLOC(interval_ptr, interval_t*, interval_t, 1);
                                                                frame_l->frame.missing_data.head = interval_ptr;
                                                                interval_ptr->prev = NULL;
                                                        } else {
                                                                /* missing_data is no empty, start from interval_ptr (where we left off) */
                                                                //printf(", ");
                                                                ALLOC((interval_ptr->next), interval_t*, interval_t, 1);
                                                                interval_ptr->next->prev = interval_ptr;
                                                                interval_ptr = interval_ptr->next;
                                                        }
                                                        /* interval_ptr here points to the newly created interval */
                                                        interval_ptr->next = NULL;
                                                        interval_ptr->start = ntohl(p->u.ccack.raw_intervals[2*i]);
                                                        interval_ptr->end = ntohl(p->u.ccack.raw_intervals[(2*i)+1]);
                                                        //printf("+new[%d %d] ", interval_ptr->start, interval_ptr->end);
                                                        // fflush(stdout);
                                                }
                                        } else if (i < frame_l->frame.missing_data.count &&
                                                   i == p->u.ccack.interval_count) {
                                                /* delete extra intervals at end of the missing_data list, start with interval after interval_ptr */
                                                interval_ptr = interval_ptr->next;
                                                if (interval_ptr != NULL) {
                                                        interval_ptr->prev->next = NULL;
                                                }
                                                for (; i < frame_l->frame.missing_data.count; i++) {
                                                        //printf("3 ");
                                                        //printf("-old[%d %d] ", interval_ptr->start, interval_ptr->end);
                                                        // fflush(stdout);
                                                        interval_ptr2 = interval_ptr->next;
                                                        RELEASE(interval_ptr, sizeof(interval_t));
                                                        interval_ptr = interval_ptr2;
                                                }
                                        }
                                        //printf("\n");
                                        frame_l->frame.missing_data.count = p->u.ccack.interval_count;
                                        //printf("new_count[%d]\n", frame_l->frame.missing_data.count);
                                        break;
                                }
                                frame_l = frame_l->next;
                        }
                        break;
                case CLOSE:
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "Received CLOSE");
                        GET_TIME_OF_DAY(&(sock->stats.tv_end));
                        sock->state = CONN_DISCONNECTED;
                        CC_Recvd_Close(p, sock);
                        if (ret_val < 2) ret_val+=2;
                        return ret_val;
                case DATA:
                        goto data_packet;
                default:
                        if (p->type == DATA_ACK) {
                                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "Received DATA_ACK ; ignoring for now");
                                break;
                        } else {
                                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "Received unknown packet type!");
                        }
                } /* switch (p->type) */
                RELEASE_PACK(p);
                continue;

        data_packet:

                /* if it's a data packet with sync option set, reply immediately with an ack */
                if ((p->opts & OPT_SYNC) == OPT_SYNC) {
                        /* printf("sending ack b/c data packet has OPT_SYNC\n"); */
                        // sock->ccontrol.syncing = 1;
                        Control_Send_Ack (sock);
                        // sock->ccontrol.syncing = 0;
                }


                pframe_num = p->u.data.frame_num;
//              DEBUG_PACK(1, "control.c", "Control_RecvAll", p, "DATA: ");

                /* data packet fields sanity checks */
                if (pframe_num < sock->recv_frames.first_frame_num) {
                        /* check if the frame referenced by packet is out of bounds (old frame) and ignore it */
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "DATA Packet with old frame number[%d] outside bounds: [%d,inf], ignoring",
                                    pframe_num, sock->recv_frames.first_frame_num);
                        break;
                } else if (p->u.data.frame_len == 0) {
                        /* check if the packet carries frame size */
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "DATA Packet has no frame_len ; ignoring");
                        RELEASE_PACK(p);
                        continue;
                } else if (p->u.data.frame_len > MAX_FRAME_SIZE) {
                        /* check that the frame doesn't exceed maximum frame size */
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "DATA Packet has too large a frame_len[%d] ; my max[%d]", p->u.data.frame_len, MAX_FRAME_SIZE);
                        RELEASE_PACK(p);
                        continue;
                /* } else if (p->u.data.data_len > MAX_FRAGMENT_SIZE) {
                        // check that the data size of the frame doesn't exceed max fragment size
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "DATA Packet of new frame has frag_size too large[%d] ; my max[%d]", p->u.data.data_len, MAX_FRAME_SIZE);
                        RELEASE_PACK(p);
                        continue; */
                }

                /* from here on, the packet's frame_num is greater or equal to first_frame_num */
                /* signal to cc that we have received a data packet */
                CC_Recvd_Data(p, sock);
                sock->stats.data_bytes_recvd+=p->u.data.data_len;

                /* update the last_frame_num if the frame number in packet is the newest(largest) seen so far */
                if (sock->recv_frames.last_frame_num < pframe_num) {
                        sock->recv_frames.last_frame_num = pframe_num;
                }

                /* add phanton frames to beginning of the queue if they are necessary */
                if (sock->recv_frames.num == 0) {
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "recv_frames is empty, creating phantoms [first[%d], %d]",
                                    sock->recv_frames.first_frame_num, pframe_num);

                        frame_prev = NULL;
                        /* this loop generates phantom frames and the packet's frame */
                        for (i = sock->recv_frames.first_frame_num; i <= pframe_num; i++) {
                                /* create the missing frame as a phantom (see datagram.c) */
                                frame_prev = Datagram_Insert_Phantom(&(sock->recv_frames), frame_prev, i);
                        }
                        is_phantom = 1;
                        frame_l = frame_prev;
                } else {

//                      printf("before creating phantoms and packet's frame:\n");
//                      Datagrams_Print(&(sock->recv_frames));

                        frame_l = sock->recv_frames.first;
                        /* this loop assumes consecutive frame numbers in frames in the recv_frames list */
                        while (frame_l != NULL) {
                                if (frame_l->frame.frame_num == pframe_num) {
//                                      DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "###: found frame");
                                        is_phantom = 0;
                                        break;  /* found frame, it's stored in frame_l */
                                } else if (frame_l->next == NULL) {
                                        if (frame_l->frame.frame_num >= pframe_num) {
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "ERR: Inconsistent recv_frame list");
                                        }
//                                      DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "###: creating phantoms and frame at end of list");
                                        /* need to create the missing frame and possibly phantom frames
                                           appearing between pframe_num and the last frame in the recvd list */
                                        frame_prev = frame_l;
                                        for (i = frame_l->frame.frame_num + 1; i <= pframe_num; i++) {
                                                frame_prev = Datagram_Insert_Phantom(&(sock->recv_frames), frame_prev, i);
                                        }
                                        /* remember the frame pointer to the packet's frame in frame_l */
                                        frame_l = frame_prev;
                                        is_phantom = 1;
                                        break;
                                } else if (frame_l->frame.frame_num < pframe_num && frame_l->next->frame.frame_num > pframe_num) {
                                        /* need to create the missing frame and possibly phantom frames
                                           appearing between frame_l's frame_num and pframe_num, and
                                           between pframe_num and frame_l->next's frame_num */
//                                      DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "###: creating phantoms and frame in middle of list");
                                        frame_prev = frame_l;
                                        for (i = frame_l->frame.frame_num + 1; i < frame_l->next->frame.frame_num; i++) {
                                                frame_prev = Datagram_Insert_Phantom(&(sock->recv_frames), frame_prev, i);
                                                if (i == pframe_num) {
                                                        /* remember the frame pointer to the packet's frame in frame_l */
                                                        frame_l = frame_prev;
                                                }
                                        }
                                        is_phantom = 1;
                                        break;
                                }
                                frame_l = frame_l->next;
                        }
                }

                /* frame_l should not be NULL here ; it should point
                 * to either the found frame for the packet or the
                 * newly created frame for the packet */

//              printf("after creating phantoms and packet's frame:\n");
//              Datagrams_Print(&(sock->recv_frames));

                if (is_phantom == 1) {
                        /* packet's frame has just been created as a phantom */
                        ALLOC(frame_l->frame.data, void*, char, p->u.data.frame_len);
                        frame_l->frame.len = p->u.data.frame_len;

                        ALLOC(frame_l->frame.missing_data.head, interval_t*, interval_t, 1);
                        frame_l->frame.missing_data.head->next = NULL;
                        frame_l->frame.missing_data.head->prev = NULL;
                        frame_l->frame.missing_data.head->end = frame_l->frame.len;
                        frame_l->frame.missing_data.head->start = 0;
                        frame_l->frame.missing_data.count = 1;
                        frame_l->frame.frame_num        = p->u.data.frame_num;
                        frame_l->frame.urgency          = 0;
                        frame_l->frame.done             = 0;

                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "New recv_frame allocated: id[%d] len[%d] ; recv_frames->num[%d]",
                                    frame_l->frame.frame_num, frame_l->frame.len, sock->recv_frames.num);
//                      Datagrams_Print(&(sock->recv_frames));
                } else {
                        /* packet's frame was found and was not just created as a phantom */
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "Found the frame in recv_frames");
                        /* check if this fragment is destined for a previously phantom created frame */
                        if (frame_l->frame.len == 0) {
                                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "Received first packet to a previously phantom frame[%d]", frame_l->frame.frame_num);
                                /* update the previously blank frame with the correct properties for the frame from the fragment */
                                frame_l->frame.len      = p->u.data.frame_len;
                                if (ALLOC(frame_l->frame.data, void*, char, p->u.data.frame_len) == 0) {
                                        frame_l->frame.len = 0;
                                        RELEASE_PACK(p);
                                        continue;
                                }
                                ALLOC(frame_l->frame.missing_data.head, interval_t*, interval_t, 1);
                                frame_l->frame.missing_data.head->next = NULL;
                                frame_l->frame.missing_data.head->prev = NULL;
                                frame_l->frame.missing_data.head->end = frame_l->frame.len;
                                frame_l->frame.missing_data.head->start = 0;
                                frame_l->frame.missing_data.count = 1;
                                /* urgency is reset in ControlT right after the first nack_vec is sent */
                        }
                }


                /* check if the frame referenced by packet needs the data interval carried by this packet */
                interval_ptr = frame_l->frame.missing_data.head;
                //printf("intervals: ");
                while (interval_ptr != NULL) {
                        //printf("{%d %d ", interval_ptr->start, interval_ptr->end);
                        if ((interval_ptr->start == p->u.data.data_offset) &&
                            (interval_ptr->end == (p->u.data.data_offset + p->u.data.data_len))) {
                                //printf("1");
                                /* delete the old interval the data for which has been received */
                                if (interval_ptr->prev != NULL) {
                                        interval_ptr->prev->next = interval_ptr->next;
                                        if (interval_ptr->next != NULL) {
                                                interval_ptr->next->prev = interval_ptr->prev;
                                        }
                                } else { /* this is the first interval */
                                        frame_l->frame.missing_data.head = interval_ptr->next;
                                        if (interval_ptr->next != NULL) {
                                                interval_ptr->next->prev = NULL;
                                        }
                                }
                                RELEASE(interval_ptr, sizeof(interval_t));
                                frame_l->frame.missing_data.count--;
                                break;
                        } else if ((interval_ptr->start < p->u.data.data_offset) &&
                                   (interval_ptr->end == (p->u.data.data_offset + p->u.data.data_len))) {
                                //printf("2");
                                /* update the old interval with a new endpoint */
                                interval_ptr->end = p->u.data.data_offset;
                                break;
                        } else if ((interval_ptr->start == p->u.data.data_offset) &&
                                   (interval_ptr->end > (p->u.data.data_offset + p->u.data.data_len))) {
                                //printf("3");
                                /* update the old interval with a new start point */
                                interval_ptr->start = interval_ptr->start + p->u.data.data_len;
                                break;
                        } else if ((interval_ptr->start < p->u.data.data_offset) &&
                                   (interval_ptr->end > (p->u.data.data_offset + p->u.data.data_len))) {
                                //printf("4");
                                /* update the old interval with a new endpoint and create a new interval for the remaining section at the end */
                                ALLOC(interval_ptr2, interval_t*, interval_t, 1);
                                interval_ptr2->next = interval_ptr->next;
                                interval_ptr2->prev = interval_ptr;
                                interval_ptr->next = interval_ptr2;
                                interval_ptr2->start = p->u.data.data_offset + p->u.data.data_len;
                                interval_ptr2->end = interval_ptr->end;
                                interval_ptr->end = p->u.data.data_offset;
                                frame_l->frame.missing_data.count++;
                                break;
                        }
                        //printf("} ");
                        interval_ptr = interval_ptr->next;
                }
                //printf("\n");

                if (interval_ptr == NULL) {
                        /* this means we couldn't find a suitable
                           interval for the data [ie. no interval
                           exists that include (data_offset,
                           data_offset+data_len)] or this data has
                           been already received so we ignore this
                           packet */
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "DATA Packet with offset[%d] len[%d] ignored",
                                    p->u.data.data_offset, p->u.data.data_len);
                        RELEASE_PACK(p);
                        continue;
                }

                /* copy fragment :( */
                memcpy(frame_l->frame.data + p->u.data.data_offset,
                       p->u.data.data,
                       p->u.data.data_len);

                sock->recv_frames.length += p->u.data.data_len;

                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "Added data to frame[%d] at offset[%d] with frame_len[%d] ",
                            frame_l->frame.frame_num, p->u.data.data_offset, frame_l->frame.len);

                /* update the urgency for all the receiving frames for this socket:
                   -> increment for all frames < this frame
                   -> decrement for all frames >= this frame
                */

                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "Updating urgencies");
                frame_l2 = sock->recv_frames.first;
                while (frame_l2 != frame_l) {
                        if (frame_l2->frame.urgency >= 0 && frame_l2->frame.done == 0 &&
                            (frame_l2->frame.urgency < URGENCY_THRESHOLD)) {
                                frame_l2->frame.urgency++;
//                              printf("incrementing frame[%d] urgency to [%d]\n", frame_l2->frame.frame_num, frame_l2->frame.urgency);
//                              DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "frame[%d].urgency: [%d]", frame_l2->frame.frame_num, frame_l2->frame.urgency);
                        }
                        frame_l2 = frame_l2->next;
                }

                frame_l2 = frame_l;
                while (frame_l2 != NULL) {
                        if (frame_l2->frame.urgency > 0 && frame_l2->frame.done == 0) {
//                              printf("resetting frame[%d] urgency to 0\n", frame_l2->frame.frame_num);
                                frame_l2->frame.urgency = 0;
                        }
//                      DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "frame[%d].urgency: [%d]", frame_l2->frame.frame_num, frame_l2->frame.urgency);
                        frame_l2 = frame_l2->next;
                }

                /* check if the front frame is ready to be delivered (moved to recvd queue)
                   but do not move it if it's a phantom frame- a blank
                   or if it is incomplete */
                if (frame_l->frame.len > 0 && (frame_l->frame.missing_data.count == 0)) {
                        /* no missing data intervals in the frame */
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "frame[%d] len[%d] DONE", frame_l->frame.frame_num, frame_l->frame.len);
                        frame_l->frame.done = 1;
                        if (sock->recv_frames.first == frame_l) {
                                frame_l2 = frame_l;
                                while (frame_l2 != NULL) {
                                        if (frame_l2->frame.len == 0) break; /* break because we reached a phantom frame */

                                        /* if frame completed and in
                                         * line to be delivered to the
                                         * app - move it to the
                                         * recevied queue where we
                                         * keep completed frames and
                                         * loop through the rest of
                                         * the recv_frames list and
                                         * move the frames that are
                                         * completed and right after
                                         * this frame to the recvd
                                         * queue (note: the next statement
                                         * is evaluated twice but it's
                                         * better than twice the
                                         * amount of code inside loop
                                         * body) */
                                        frame_l = frame_l2;
                                        frame_l2 = frame_l->next;

                                        if (frame_l->frame.missing_data.count == 0) {
                                                /* recv_frames */
                                                sock->recv_frames.first = frame_l->next;
                                                if (sock->recv_frames.last == frame_l) {
                                                        sock->recv_frames.last = NULL;
                                                        sock->recv_frames.last_frame_num = frame_l->frame.frame_num + 1;
                                                }

                                                sock->recv_frames.first_frame_num = frame_l->frame.frame_num + 1;
                                                sock->recv_frames.num--;

                                                /* recvd_frames */
                                                if (sock->recvd_frames.num == 0) {
                                                        sock->recvd_frames.first = frame_l;
                                                } else {
                                                        sock->recvd_frames.last->next = frame_l;
                                                }

                                                sock->recvd_frames.last = frame_l;
                                                frame_l->next = NULL;

                                                sock->recvd_frames.num++;
                                                sock->recvd_frames.length += frame_l->frame.len;
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_RecvAll", "Frame[%d] moved into received queue; recv_frames->num[%d] ; recvd_frames->num[%d]",
                                                            frame_l->frame.frame_num, sock->recv_frames.num, sock->recvd_frames.num);
                                                /* Datagrams_Print(&(sock->recvd_frames)); */

                                        } else {
                                                break;
                                        }
                                        /* frame_l2 holds the next element in the recv_frames queue */
                                } /* while (frame_l2 != NULL) */
                        } /* if sock->recv_frames.first == frame_l */
                } /* if frame.len > 0 && missing_data.count == 0 */

                if (ret_val == 0 || ret_val == 2) ret_val+=1;
                /* release the received packet */
                RELEASE_PACK(p);
        }
        return ret_val;
} /* Control_RecvAll() */

/**
 * Sends all the control packets that are in the control packet queue
 * maintained by control.c for each socket
 * NOTE: socket's mutex must be locked before calling this function
 *
 * @param sock vfer_sock pointer
 * @param p control packet to send on the socket
 * @return
 *      - 0 on success
 *      - -1 on error
 */
inline static int Control_SendCtl (vfer_sock* sock, packet* p) {
        /* control packets are not congestion controlled because they do not have a data payload */
        Packet_Set_Seq(p,  sock->l_seq);
        DEBUG_PACK(DEBUG_CTL, "control.c", "ControlSendCtl", p, "-");

        sock->stats.bytes_sent+=p->iov[0].iov_len;
        sock->stats.packets_sent++;

        /* Control_Write releases the packet mem */
        if (Control_Write(p, sock->fd) != 0) {
                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_SendAllCtl", "Control_Write failed; sock->fd[%d] ; CONN_ERR", sock->fd);
                sock->state = CONN_ERR;
                return -1;
        }
        CC_Sent_CtlPack(p, sock);
        sock->l_seq++;
        return 0;
} /* Control_SendCtl() */

/**
 * Sends a congestion control ack packet on socket by creating it and
 * then enqueueing it into the control packet queue.
 *
 * CC is byte accounted (for data carrying packets only) so CC always
 * allows for sending CCAcks
 *
 * @param s socket on which to send the ccack
 * @return 0 on success
 * @return -1 on error
 */
inline static int Control_Send_Ack (vfer_sock* s) {
        packet* p;
        ccontrol_t* c;
        struct timeval tv;
        intervals_t* missing;
        
        c = &(s->ccontrol);
        GET_TIME_OF_DAY(&tv);

        missing = NULL;
        if (s->recv_frames.first != NULL &&
            s->recv_frames.first->frame.missing_data.count != 0) {
                // printf("setting missing_data to ~NULL b/c first->frame.missing_data.count = %d\n", s->recv_frames.first->frame.missing_data.count);
                missing = &(s->recv_frames.first->frame.missing_data);
        }
        
        p = Packet_Form_CCAck(0, 0, c->r_last_seq - 1, c->r_packets_lost,
                              s->recv_frames.first_frame_num, s->recv_frames.last_frame_num,
                              Tree_Max(&(c->rev_path_curr_delayhist), tv) - Tree_Min(&(c->rev_path_base_delayhist), tv),
                              Tree_Max(&(c->rev_path_curr_delayhist), tv), c->r_bytes_total, s->recv_frames.first_frame_num, missing);
        
        if (Control_SendCtl(s, p) != 0) {
                return -1;
        }
        c->r_first_seq = c->r_last_seq;
        return 0;
} /* Control_Send_Ack() */

/**
 * Sends all data packets that need to be sent on a socket
 * @param sock socket link pointer to send data on
 * @return
 *      0 on success no packets sent
 *      1 on success some packets sent
 *      -1 on sending error
 */
static int Control_SendAllData (vfer_sock* sock) {
        /* note that if we are in CONN_CLOSING state, we can still send data packets ; we just can't accept any to send (Control_Send)*/
        frame_link* frame_l;
        frame* frame;
        interval_t* interval_ptr;
        interval_t* interval_ptr2;
        uint32_t opts;
        uint32_t data_size;
        uint32_t frame_len;
        int packets_sent;
        packet* pack;
        int ret;
        ccontrol_t* c;

        c = &(sock->ccontrol);
        packets_sent = 0;
        frame_l = sock->send_frames.first;
        while (frame_l != NULL) {

                /* itterate through existing missing data intervals and send data in chunks of current MTU size
                   upon every interval send, remove the interval from the list */
                frame = &(frame_l->frame);
                interval_ptr = frame->missing_data.head;
                while (interval_ptr != NULL) {
                        /* check if there is space in the bin for this socket for another packet */

                        /* TODO: check if any packets arrived while we are
                         * sending data in case we need to adjust the
                         * window to be smaller and thereby avoid
                         * congestion collapse */

                        opts = 0;
                        frame_len = frame->len; /* for now include frame length on all data packets, not just on the initial packet */
                        while ((interval_ptr->end - interval_ptr->start) != 0) {
                                if(sock->pmtu.probe_state == NO_PROBE_SENT && (interval_ptr->end - interval_ptr->start) >= *(sock->pmtu.probe_size)){
                                        /* PMTU: Send probe if none outstanding and enough data in frame left to fill probe */
                                        data_size = *(sock->pmtu.probe_size);
                                        sock->pmtu.frame_num = frame->frame_num;
                                        sock->pmtu.data_offset = interval_ptr->start;
                                }else if ((interval_ptr->end - interval_ptr->start) >= c->mtu) {
                                        data_size = c->mtu;
                                } else {
                                        /* send the end piece in the interval that might be less than c->mtu */
                                        data_size = interval_ptr->end - interval_ptr->start;
                                }

                                /* check with CC control whether we can send another packet */
                                if (CC_Can_Send_Data(data_size, sock) == 0) {
                                        // printf("sendalldata : CC said we can't send more data; exiting sendalldata function\n");
                                        return packets_sent;
                                }

                                /* printf("sending frame[%d] offset[%d] data_size[%d]\n", frame->frame_num, interval_ptr->start, data_size); */
                                
                                if (c->syncing == 1) {
                                        /* printf("SETTING OPT_SYNC\n"); */
                                        opts = opts | OPT_SYNC;
                                        c->syncing = 0;
                                }

                                if ((pack = Packet_Form_Data(opts,
                                                             sock->l_seq,
                                                             frame->frame_num,
                                                             interval_ptr->start,
                                                             frame_len,
                                                             (frame->data + interval_ptr->start),
                                                             data_size)) == NULL) {
                                        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_SendAllData", "Couldn't form data packet for frame[%d] sock->fd[%d]", frame->frame_num, sock->fd);
                                } else {
                                        DEBUG_PACK(DEBUG_CTL, "control.c", "Control_SendAllData", pack, "Sending DATA packet on skt->fd[%d]", sock->fd);
                                        /* note we assume that a data packets is composed of 2 buffers, this is a bad dependency */
                                        ret = Control_Write(pack, sock->fd);
                                        
                                        if(ret != 0 && sock->pmtu.probe_state == NO_PROBE_SENT && *(sock->pmtu.probe_size) == data_size){
                                                /* PMTU: send failed because packet bigger than local MTU. */
                                                //clear error
                                                int err_val;
                                                socklen_t err_val_len = 4;
                                                if (getsockopt(sock->fd, SOL_SOCKET, SO_ERROR, &err_val, &err_val_len) == -1)   perror("getsockopt");
                                                
                                                sock->pmtu.search_high = sock->pmtu.probe_size;
                                                sock->pmtu.probe_state = MTU_FOUND;
                                                struct timeval tv;
                                                GET_TIME_OF_DAY(&tv);
                                                sock->pmtu.next_probe_time = tv.tv_sec + PMTUD_TIME_INTERVAL;
                                                
                                                DEBUG_PMTUD_PRINT(DEBUG_PMTUD, "control.c", "Control_SendAllData",
                                                                  "PMTU of size %d found [Probe write failed]. Next probe to be begin at %lu.", c->mtu, (long int)sock->pmtu.next_probe_time);
                                                continue;
                                        }else if(sock->pmtu.probe_state == NO_PROBE_SENT && *(sock->pmtu.probe_size) == data_size){
                                                /* PMTU: Successful write so set probe as sent */
                                                sock->pmtu.probe_state = PROBE_SENT;
                                                DEBUG_PMTUD_PRINT(DEBUG_PMTUD, "control.c", "Control_SendAllData",
                                                                  "Probe of size %d sent frame_num[%d] data_offset[%d]", data_size, sock->pmtu.frame_num, sock->pmtu.data_offset);
                                        }else if (ret != 0) {
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_SendAllData", "Control_Write failed; skt->fd[%d] ; CONN_ERR", sock->fd);
                                                sock->state = CONN_ERR;
                                                return -1;
                                        }

                                        /* indicates that this packet was sent */
                                        /* update the stats */
                                        sock->stats.bytes_sent+=pack->iov[0].iov_len;
                                        sock->stats.bytes_sent+=pack->iov[1].iov_len;
                                        sock->stats.data_bytes_sent+=pack->iov[1].iov_len;
                                        sock->stats.packets_sent++;

                                        CC_Sent_Data(pack, frame_l, sock); /* Let CC control update the CC information for this socket */
                                        packets_sent = 1; /* set the return value to indicate that we managed to send some data packet */
                                        sock->l_seq++;
                                }
                                /* update the interval with the fact that we've sent a piece from the front of size data_size */
                                interval_ptr->start += data_size;
                        } /* interval_ptr->end - interval_ptr->start != 0 */

                        /* delete the missing data interval that was just sent in pieces of current MTU */
                        if (interval_ptr->prev != NULL) {
                                interval_ptr->prev->next = interval_ptr->next;
                                if (interval_ptr->next != NULL) {
                                        interval_ptr->next->prev = interval_ptr->prev;
                                }
                        } else { /* this is the first interval */
                                frame_l->frame.missing_data.head = interval_ptr->next;
                                if (interval_ptr->next != NULL) {
                                        interval_ptr->next->prev = NULL;
                                }
                        }
                        frame_l->frame.missing_data.count--;
                        interval_ptr2 = interval_ptr->next;
                        RELEASE(interval_ptr, sizeof(interval_t));
                        interval_ptr = interval_ptr2;
                } /* interval_ptr != NULL */
                frame_l = frame_l->next;
        } /* while (frame_l != NULL) */
        return packets_sent;
} /* Control_SendAllData() */

/**
 * Main Control thread that selects on incoming fds, receives all
 * packets, manages timers for congestion control and does a bunch of
 * other logic things for the protocol. This thread goes through the
 * socket linked list and performs the necessary operations to send &
 * recv frames to and from the api land and all of this is marshalled
 * based on the state of each of the sockets.
 *
 * @param
 *      ptr is not used for anything right now
 */
static void ControlT(void* ptr) {
        int int_optval;
        socklen_t int_optlen;
        struct timeval tv, tv_const;
        // frame_link* frame_prev;
        // frame_link* frame_l;
        // frame_link* frame_l2;
        // context_t context;
        packet* p;
        uint32_t rtt;
        ccontrol_t *c;
        fd_set fds;
        int max_fd, num_active, i, j, ret;
        vfer_sock* sock;
        int val;
        int num_packets;
        /* server vars */
        struct sockaddr_in addr;
        packet* response;
        packet* ack;
        socklen_t addr_len;
        char buf[INET_ADDRSTRLEN];
        int err;
        interval_t *interval_ptr;

        tv_const.tv_sec = 0;
        tv_const.tv_usec = CONTROLT_SLEEPTIME;

        /* set fds, and max_fd for first itteration */
        pthread_mutex_lock(&(sockets.mutex));
        fds = sockets.fds;
        max_fd = sockets.max_fd;
        num_active = sockets.num_active;
        pthread_mutex_unlock(&(sockets.mutex));

        DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Entered with num_active[%d]", num_active);

        while (1) {
        select:
                tv = tv_const;
                /* TODO: maintain a timeout set to smallest rtt of all sockets minus some epsilon */

                if (select(max_fd + 1, &fds, 0, 0, &tv) == -1) {
                        if (errno == EINTR) {
                                /* interrupted select call */
                                pthread_mutex_lock(&(sockets.mutex));
                                fds = sockets.fds;
                                max_fd = sockets.max_fd;
                                num_active = sockets.num_active;
                                pthread_mutex_unlock(&(sockets.mutex));
                                goto select; /* retry */
                        }
                }

                i = j = 0;
                while (j < num_active && i < MAX_SOCKETS) {
                        pthread_mutex_lock(&(sockets.s[i].mutex));
                        c = &(sockets.s[i].ccontrol);
                        sock = &(sockets.s[i]);
                        i++;


                        switch (sock->state) {
                        case CONN_DISCONNECTING:
                        case CONN_ERR:
                                if (sock->send_frames.num == 0) {
                                        j++;
                                        p = Packet_Form_Close(0, 0);
                                        Control_SendCtl(sock, p);
                                        goto disconnect_socket;

                                }
                                /* note that we don't break but continue to CONN_CONNECTED case
                                   because there are datagrams left that are unset
                                   note: j will be incremented in CONN_CONNECTED case */

                        case CONN_CONNECTED:
                                j++;

                                if (! FD_ISSET(sock->fd, &fds)) {
                                        /* nothing to read on this socket */
                                        goto skip_recvall;
                                }

                                ret = Control_RecvAll(sock);
                                if (ret < 0 || sock->state == CONN_DISCONNECTED) {
                                        goto disconnect_socket;
                                }

                                if (ret != 0) {
                                        GET_TIME_OF_DAY(&(tv));
                                        if (ret == 1) { /* received only data packets */
                                                c->r_last_data  = tv;
                                        } else if (ret == 2) {  /* received only control packets */
                                                c->r_last_ccack = tv;
                                        } else if (ret == 3) {  /* received both data and control packets */
                                                c->r_last_ccack = tv;
                                                c->r_last_data  = tv;
                                        }

                                        // note, c->s_last_unacked_data is cleared in Control_RecvAll when CC Ack is received that acks last sent date
                                        if ((ret == 1 || ret == 3) &&
                                            (c->r_unacked_bytes >= UNACKED_THRESH * c->mtu)) {
                                                /* received data, send an ack if unacked data bytes is above threshold */
                                                if (Control_Send_Ack(sock) != 0) {
                                                        goto disconnect_socket;
                                                }
                                                c->r_unacked_bytes = 0;
                                                c->s_last_ack = tv;
                                        }
                                }

                        skip_recvall:

                                GET_TIME_OF_DAY(&(tv));

                                /* RTT returns rtt in microseconds */
                                rtt = RTT(sock, tv);
                                if (rtt == 0) {
                                        printf("RTT EMPTY\n");
                                        rtt = 5000; /* 5ms */
                                }

                                if (TIMEVAL_US_DIFF(c->s_rtt_start, tv) >= rtt) {
                                        c->s_rtt_start = tv;
                                        c->s_cwnd_sent = 0;
                                }

                                /* reset the recorded timestamp of last loss event if set and older than rtt */
                                if (c->r_last_loss.tv_sec != 0 &&
                                    TIMEVAL_US_DIFF(c->r_last_loss, tv) >= rtt) {
                                        c->r_last_loss.tv_sec = 0;
                                }

                                /* send an ack if we have unacked bytes to ack and
                                   time since we sent last ack is >= rtt/2 */
                                if (c->r_unacked_bytes != 0 &&
                                    TIMEVAL_US_DIFF(c->s_last_ack, tv) >= rtt/2.0) {
                                        if (Control_Send_Ack(sock) != 0) {
                                                goto disconnect_socket;
                                        }
                                        c->r_unacked_bytes = 0;
                                        c->s_last_ack = tv;
                                }

                                /* reset s_last_vec after X*RTT indicating that we can send another vec ack now */
                                // if (c->s_last_vec.tv_sec != 0 &&
                                // TIMEVAL_US_DIFF(c->s_last_vec, tv) >= (2*rtt)) {
                                        // printf("resetting s_last_vec from [%d,%d] and tv[%d,%d] and rtt[%d]\n", c->s_last_vec.tv_sec, c->s_last_vec.tv_usec, tv.tv_sec, tv.tv_usec, rtt);
                                // c->s_last_vec.tv_sec = 0;
                                // }

                                // if (c->s_last_vec.tv_sec == 1) {
                                        /* check urgency values and generate nack vecs as necessary */
                                        /* frame_l = sock->recv_frames.first;
                                        frame_prev = NULL;
                                        while (frame_l != NULL) {
                                                if (frame_l->frame.done == 1) {
                                                        frame_prev = frame_l;
                                                        frame_l = frame_l->next;
                                                        continue;
                                                }
                                        */
                                                /* if no data in 2 round trip times then we send a vec ack for first unfinished frame */
                                                // if (frame_l->frame.urgency == URGENCY_THRESHOLD || frame_l->frame.urgency == -1) {
                                                        /* send a vec ack for the first occuring threshold frame */
                                                        /* evaluate the reliability function to determine if user still interested in the datagram */
                                                        /* context.f = frame_l->frame;
                                                        sock->relfun(NULL);
                                                        context.stats = &(sock->stats);
                                                        context.fstats = &(frame_l->frame.stats); */
                                                        // if (sock->relfun(&context) == 0) {
                                                                /* discard frame */
                                                        /* Datagram_Remove (&(sock->recv_frames), frame_prev, frame_l);
                                                                frame_l2 = frame_l->next;
                                                                RELEASE(frame_l->frame.data, frame_l->frame.len);
                                                                RELEASE(frame_l, sizeof(frame_link));
                                                                frame_l = frame_l2;
                                                                continue; */ /* continue the loop with frame_l set to next frame in list and prev frame unchanged */
                                                        // }

                                                        // printf("frame[%d] reached THRESHOLD, sending vec ack\n", frame_l->frame.frame_num);

                                                        /* start the negative counter in the urgency for this frame */
                                                        // p = Packet_Form_VecAck(0, 0, frame_l->frame.frame_num, &(frame_l->frame.missing_data)); /* seq filled in by Control_SendAllCtl() */
                                        /* if (Control_SendCtl(sock, p) != 0) {
                                                                goto disconnect_socket;
                                                                } */

                                                        /* empty frame created initially with urgency of -1 ;
                                                           set it's urgency to THRESHOLD so as to make it behave as a normal frame in the future */
                                                        /* frame_l->frame.urgency = URGENCY_THRESHOLD;
                                                        c->s_last_vec = tv;
                                                }
                                                frame_prev = frame_l;
                                                frame_l = frame_l->next;
                                                }
                                                } */
                        
                                // printf("checking last_unacked_data[%d %d] with [%d %d]\n", c->s_last_unacked_data.tv_sec, c->s_last_unacked_data.tv_usec, tv.tv_sec, tv.tv_usec);
                                if (c->s_last_unacked_data.tv_sec != 0 &&
                                    (TIMEVAL_US_DIFF(c->s_last_unacked_data, tv) >= 5*rtt)) {
                                        /* consider this a timeout, ie. previous data unacked so we must resend it */
                                        /* printf("expired waiting for ack with last_unacked_data[%d,%d] and tv[%d,%d] rtt[%d]\n", c->s_last_unacked_data.tv_sec, c->s_last_unacked_data.tv_usec, tv.tv_sec, tv.tv_usec, rtt); */
                                        c->s_last_unacked_data.tv_sec = 0; /* reset the timestamp (it will be set once the repeat data pack leaves */

                                        interval_ptr = c->s_last_data_frame->frame.missing_data.head;
                                        if (interval_ptr == NULL) {
                                                ALLOC(interval_ptr, interval_t*, interval_t, 1);
                                                interval_ptr->next = NULL;
                                                interval_ptr->prev = NULL;
                                                interval_ptr->start = c->s_last_data_offset;
                                                interval_ptr->end = c->s_last_data_offset + c->mtu; /* note: this depends on the current MTU */
                                                c->s_last_data_frame->frame.missing_data.head = interval_ptr;
                                                c->s_last_data_frame->frame.missing_data.count = 1;
                                        }
                                        c->syncing = 1;
                                }

                                /* send all data packets on every socket */

                                if (sock->send_frames.length != 0 && CC_Can_Send_Data(c->mtu, sock) != 0) {
                                        /* send data packets */
                                        // printf("trying to sendall data\n");
                                        ret = Control_SendAllData(sock);
                                        switch(ret) {
                                        case 1:
                                                /* if  did send some data packets remember the time of this data transmission */
                                                GET_TIME_OF_DAY(&(c->s_last_unacked_data));
                                                break;
                                        case -1:
                                                goto disconnect_socket;
                                        }
                                }

                                /* alert any threads waiting to dequeue recvd frames or to send frames */
                                if (sock->recvd_frames.num != 0) {
                                        if (sock->select_active == 1 && (sock->select_mark & VFER_READABLE) == VFER_READABLE) {
                                                sock->select_result |= VFER_READABLE;
                                                pthread_mutex_unlock(&(sock->mutex));
                                                pthread_mutex_lock(&(sockets.select_mutex));
                                                pthread_cond_signal(&(sockets.select_cond));
                                        } else {
                                                pthread_mutex_unlock(&(sock->mutex));
                                        }
                                        pthread_cond_broadcast(&(sock->cond));
                                } else if ((sock->send_frames.length + c->mtu) <= MAX_SEND_WINDOW) {
                                        if (sock->select_active == 1 && (sock->select_mark & VFER_WRITABLE) == VFER_WRITABLE) {
                                                sock->select_result |= VFER_WRITABLE;
                                                pthread_mutex_unlock(&(sock->mutex));
                                                pthread_mutex_lock(&(sockets.select_mutex));
                                                pthread_cond_signal(&(sockets.select_cond));
                                        } else {
                                                pthread_mutex_unlock(&(sock->mutex));
                                        }
                                        pthread_cond_broadcast(&(sock->cond));
                                } else {
                                        pthread_mutex_unlock(&(sock->mutex));
                                }
                                continue;

                        case CONN_LISTENING:

                                j++;
                                if (! FD_ISSET(sock->fd, &fds)) {
                                        /* nothing to read on this socket */
                                        break;
                                }

                                addr_len = sizeof(SA);
                                if ((err = Packet_Read(sock->fd, &sock->request, (SA*)(&addr), &addr_len)) != 0) {
                                        DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Packet_Read returned error[%d]", err);
                                        break;
                                }

                                if (sock->request->type != REQUEST) {
                                        RELEASE_PACK(sock->request);
                                        DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Packet is not REQUEST, ignoring");
                                        break;
                                }
                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "received request: [%s : %d]",
                                            inet_ntop(AF_INET, &(addr.sin_addr), buf, INET_ADDRSTRLEN), ntohs(addr.sin_port));

                                if (Accept_Queue_Enqueue (sock->accept_queue, sock->fd, *((SA*)(&addr)), sock->request) == -1) {
                                        /* have to drop the connection some kind of queue error */
                                        RELEASE_PACK(sock->request);
                                        DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "couldn't enqueue new connection");
                                        break;
                                }

                                pthread_mutex_unlock(&(sock->mutex));
                                pthread_cond_signal(&(sock->cond));
                                continue;

                        case CONN_CONNECTING:

                                j++;

                                /* set up the addr structure to unconnect the socket if necessary */
                                bzero((char*)(&addr), sizeof(addr));
                                addr.sin_family = AF_UNSPEC;
                                addr_len = sizeof(addr);

                                // printf("conn_conencting state fd[%d] max_fd[%d]\n", sock->fd, max_fd);
                                if (FD_ISSET(sock->fd, &(fds))) {
                                        ret = Packet_Read(sock->fd, &response, NULL, NULL);
                                        if (ret != 0) {
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_ConnectT", "Packet_Read failed[%d]", ret);
                                                switch (ret) {
                                                case EAGAIN:
                                                        /* try handshake again */
                                                        break;
                                                case ECONNREFUSED: /* perhaps a syscall error of some sort */
                                                        sock->err = VFER_REFUSED;
                                                        goto could_not_connect;
                                                default:
                                                        sock->err = VFER_NOCONN;
                                                        goto could_not_connect;
                                                }
                                                RELEASE_PACK(sock->request);
                                                break;
                                        }

                                        /* since we are the client, change the sign of the delay measure set by packet.c */
                                        response->delay *= -1;
                                        DEBUG_PACK(DEBUG_CTL, "control.c", "Control_ConnectT",
                                                   response, "Received [2/3 handshake: Response] on sock->fd[%d]", sock->fd);
                                        
                                        if (response->type != RESPONSE) {
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_ConnectT",
                                                            "Received wrong packet type[%d] for [2/3 handshake] ; expected RESPONSE[%d]",
                                                            response->type, RESPONSE);
                                                RELEASE_PACK(response);
                                                goto could_not_connect;
                                        }

                                        /* unconnect the socket - we have to (re)connect to the port the server sent us) */
                                        /* This does not work with the VSL code-
                                           connect(sock->fd, (SA*) (&addr), addr_len); */
                                        getsockopt(sock->fd, SOL_SOCKET, SO_ERROR, &int_optval, &int_optlen); /* reset error if any */

                                        /* set up the new addr structure we will connect to */
                                        sock->addr.sin_port = htons(response->u.res.connection_port);
                                        sock->addr_len = sizeof(sock->addr);

                                        // printf("received response port: %d\n", response->u.res.connection_port);
                                        // printf("received response host: %s\n", inet_ntop(AF_INET, &(sock->addr.sin_addr), buf, INET_ADDRSTRLEN));
                                        
                                        /* open a new connection on the port specified by the server */
                                        if (connect(sock->fd, (SA*) &sock->addr, sock->addr_len) < 0 ) {
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_ConnectT", "connecting to port specified in response packet failed, %s",
                                                            strerror(errno));
                                                sock->err = VFER_NOCONN;
                                                RELEASE_PACK(response);
                                                goto could_not_connect;
                                        }
                                        
                                        /* fill in the ack packet structure */
                                        ack = Packet_Form_CCAck(0, sock->request->seq + 1, response->seq, 0,
                                                                response->u.res.init_frame_num, response->u.res.init_frame_num,
                                                                0, response->delay, 0, 0, NULL);
                                        
                                        DEBUG_PACK(DEBUG_CTL, "control.c", "ControlT", ack,
                                                   "Sending  [3/3 handshake: Ack] on sock->fd[%d]", sock->fd);

                                        /* send back an ack packet without specifying address (udp socket already connected) */
                                        if (Packet_Write(ack, sock->fd, NULL, 0) != 0) {
                                                /* note: Packet_Write will check to see if system connect() generated an error on this socket
                                                   but depending on latency the error might not be caught until later */
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Packet_Write failed");
                                                sock->err       = VFER_NOCONN;
                                                RELEASE_PACK(response);
                                                RELEASE_PACK(ack);
                                                goto could_not_connect;
                                        }

                                        /* set up all the socket parameters stored in the
                                         * request\response\ack packets here and in control.c */
                                        sock->opts |= VFERSO_CLIENT_SOCK;
                                        sock->l_seq = ack->seq + 1;

                                        pthread_mutex_unlock(&(sock->mutex));
                                        Control_Add_Socket(sock, sock->request, response, ack, 1);
                                        pthread_cond_signal(&(sock->cond));

                                        RELEASE_PACK(sock->request);
                                        RELEASE_PACK(response);
                                        RELEASE_PACK(ack);
                                        /* mutex unlocked, continue to next socket */
                                        continue;
                                } /* FD_ISSET */

                                /* socket not readable (or just made CONN_CONNECTING and therefore not in the reading set */
                                /* check to make sure we are not over max_connect_tries */
                                if (sock->connect_tries == MAX_REQUEST_TRIES) {
                                        DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Reached MAX_REQUEST_TRIES[%d]", MAX_REQUEST_TRIES);
                                        goto could_not_connect;
                                }

                                /* make sure we timeout before each request based on the REQUEST_RETRY_INTERVAL */
                                GET_TIME_OF_DAY(&(tv));
                                if ((unsigned int)(TIMEVAL_US_DIFF(sock->last_request, tv)) < (unsigned int)(REQUEST_RETRY_INTERVAL)) {
                                        break;
                                }

                                sock->last_request = tv;

                                sock->connect_tries++;
                                /* fill in the request packet */
                                if (sock->request == NULL) {
                                        sock->request = Packet_Form_Request (0, Packet_Gen_SeqNum(), MAX_FRAME_SIZE,
                                                                             sock->udp_recv_buf_size, sock->udp_send_buf_size, Packet_Gen_FrameNum());
                                }

                                DEBUG_PACK(DEBUG_CTL, "control.c", "ControlT", sock->request,
                                           "Sending  [1/3 handshake: Request] on sock->fd[%d]", sock->fd);

                                /* send request packet (because the socket is connected we do not specify addr/addrlen) */
                                ret = Packet_Write(sock->request, sock->fd, NULL, 0);
                                if (ret != 0) {
                                        /* test the returned pending error on the socket */
                                        switch (ret) {
                                        case ECONNREFUSED:
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "connection refused on Packet_Write");
                                                sock->err = VFER_REFUSED;
                                                break;
                                        default:
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Packet_Write failed with error[%d]", ret);
                                                sock->err = VFER_NOCONN;
                                        }
                                        RELEASE_PACK(sock->request);
                                        goto could_not_connect;
                                }
                                break;

                        could_not_connect:

                                connect(sock->fd, (SA*) &addr, addr_len);       /* unconnect the socket */
                                sock->state = CONN_BOUND;               /* reset connection state to bound */
                                getsockopt(sock->fd, SOL_SOCKET, SO_ERROR, &int_optval, &int_optlen); /* reset error if any */

                                if (sock->select_active == 1 && (sock->select_mark & VFER_EXCEPTION) == VFER_EXCEPTION) {
                                        sock->select_result |= VFER_EXCEPTION;
                                        pthread_mutex_unlock(&(sock->mutex));
                                        pthread_mutex_lock(&(sockets.select_mutex));
                                        pthread_cond_signal(&(sockets.select_cond));
                                } else {
                                        pthread_mutex_unlock(&(sock->mutex));
                                }
                                pthread_cond_signal(&(sock->cond));
                                break;

                        case CONN_ACCEPTING:

                                j++;

                                if (sock->listening_sock->state != CONN_LISTENING) {
                                        /* race condition here- if listning socket is closed before a connection is fully accepted, vfer_accept() will fail */
                                        sock->err = VFER_IMPL;
                                        goto disconnect_socket;
                                }

                                if (FD_ISSET(sock->fd, &(fds))) {
                                        /* if socket readable try to receive cc_ack packet on connection fd */
                                        sock->addr_len = sizeof(SA);
                                        /* printf("conn_accepting readable socket\n"); */
                                        if ((err = Packet_Read(sock->fd, &ack, (SA*) &(sock->addr), &(sock->addr_len))) != 0
                                            || ack->type != CC_ACK) {
                                                if (err != 0) {
                                                        DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Try[%d] Packet_Read returned error[%d]", sock->response_tries, err);
                                                } else {
                                                        /* printf("Try[%d] Received a non CC_ACK packet as [3/3 handshake] on sock->fd[%d]", sock->response_tries, sock->fd); */
                                                        DEBUG_PACK(DEBUG_CTL, "control.c", "ControlT", ack,
                                                                   "Try[%d] Received a non CC_ACK packet as [3/3 handshake] on sock->fd[%d]", sock->response_tries, sock->fd);
                                                        RELEASE_PACK(ack);
                                                }
                                                if (Unconnect_Rebind(sock) == -1) {
                                                        /* couldn't rebind socket */
                                                        close(sock->fd);
                                                        RELEASE_PACK(sock->qc->request);
                                                        RELEASE(sock->qc, sizeof(queued_conn));
                                                        RELEASE_PACK(sock->response);
                                                        sock->err = VFER_IMPL;
                                                        goto disconnect_socket;
                                                }
                                                /* try sending the response packet again */
                                                goto try_response;
                                        }

                                        /* compare the host addresses on the received packet and previously received request packet */
                                        if (memcmp(&(sock->qc->sa), &(sock->addr), sizeof(sock->qc->sa)) != 0) {
                                                RELEASE_PACK(ack);
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT",
                                                            "Try[%d] Mismatched addresses of remote host ignoring this conn request", sock->response_tries);
                                                if (Unconnect_Rebind(sock) == -1) {
                                                        /* couldn't rebind socket */
                                                        close(sock->fd);
                                                        RELEASE_PACK(sock->qc->request);
                                                        RELEASE(sock->qc, sizeof(queued_conn));
                                                        RELEASE_PACK(sock->response);
                                                        sock->err = VFER_IMPL;
                                                        goto disconnect_socket;
                                                }
                                                /* try sending the response packet again */
                                                goto try_response;
                                        }

                                        DEBUG_PACK(DEBUG_CTL, "control.c", "ControlT", ack, "Try[%d] Received [3/3 handshake] on sock->fd[%d]", sock->response_tries, sock->fd);
                                        err = 0;
                                        if (ack->seq != sock->qc->request->seq + 1) {
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Try[%d] bad seq number on cc_ack[%u] should be[%u]", sock->response_tries, ack->seq,
                                                            (sock->qc->request->seq + 1));
                                                err = -1;
                                        } else if (ack->u.ccack.last_seq != sock->response->seq) {
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Try[%d] bad last_seq_recv number on cc_ack", sock->response_tries);
                                                err = -1;
                                        } else if (ack->u.ccack.packets_lost != 0) {
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Try[%d] bad packets lost on cc_ack", sock->response_tries);
                                                err = -1;
                                        } else if (ack->u.ccack.oldest_frame != sock->response->u.res.init_frame_num) {
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Try[%d] bad oldest_frame on cc_ack", sock->response_tries);
                                                err = -1;
                                        }

                                        if (err == -1) { /* bad cc_ack */
                                                RELEASE_PACK(ack);
                                                if (Unconnect_Rebind(sock) == -1) {
                                                        /* couldn't rebind socket */
                                                        close(sock->fd);
                                                        RELEASE_PACK(sock->qc->request);
                                                        RELEASE(sock->qc, sizeof(queued_conn));
                                                        RELEASE_PACK(sock->response);
                                                        sock->err = VFER_IMPL;
                                                        goto disconnect_socket;
                                                }
                                                /* try sending the response packet again */
                                                goto try_response;
                                        }

                                        /* fill in the parameters in the sock structure */
                                        sock->r_seq = ack->seq + 1; /* the others are filled in control.c from original vfer_accept() call */

                                        /* connected ! */
                                        sock->type              = SOCK_DGRAM;
                                        sock->opts              = 0;
                                        sock->err               = 0;
                                        sock->select_mark       = 0;
                                        sock->select_result     = 0;
                                        sock->select_active     = 0;

                                        /* address structure copying */
                                        bzero(&addr, sock->addr_len);
                                        memcpy(&addr, &sock->addr, sock->addr_len);

                                        /* signal to ccontrol that we have a new connection to
                                           monitor for incoming and outgoing packets */
                                        pthread_mutex_unlock(&(sock->mutex));
                                        Control_Add_Socket(sock, sock->qc->request, sock->response, ack, 0);
                                        pthread_cond_signal(&(sock->cond));

                                        RELEASE_PACK(sock->qc->request);
                                        RELEASE(sock->qc, sizeof(queued_conn));
                                        RELEASE_PACK(sock->response);
                                        RELEASE_PACK(ack);
                                        continue;
                                }

                                /* response packet sending logic, sends a response if either
                                   1. This is the first time we enter the loop
                                   2. We have waited longer than HANDSHAKE_ACK_WAIT_INTERVAL microsecs since sending a previous response
                                */

                        try_response:
                                GET_TIME_OF_DAY(&(tv));
                                if (sock->last_response.tv_sec != 0 &&
                                    (unsigned int)(TIMEVAL_US_DIFF(sock->last_response, tv)) < (unsigned int)(HANDSHAKE_ACK_WAIT_INTERVAL)) {
                                        break;
                                }

                                if (sock->response_tries == MAX_RESPONSE_TRIES) {
                                try_next_conn:
                                        /* grab the top connection waiting to be accepted  */
                                        sock->qc = Accept_Queue_Dequeue (sock->listening_sock->accept_queue);
                                        if (sock->qc == NULL) {
                                                /* server socket is non-blocking => don't wait for new connections */
                                                if (SOCK_OPT_ISSET(sock, VFERSO_NONBLOCK)) {
                                                        sock->err = VFER_WOULDBLOCK;
                                                        goto disconnect_socket;
                                                }
                                                /* server blocking sockets => wait indefinitly for a connection */
                                                // DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "blocking with empty accepted queue");
                                                break;
                                        }

                                        DEBUG_PACK(DEBUG_CTL, "control.c", "ControlT", sock->qc->request,
                                                   "Dequeued [1/3 handshake] from acceptQ backlog for server_sock->fd[%d]", sock->fd);

                                        /* connect the return socket(already bound to new port) to the host we dequeued from the accept queue */
                                        if (connect(sock->fd, &(sock->qc->sa), sizeof(sock->qc->sa)) != 0) {
                                                perror("ControlT[connect]");
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "connect on new socket failed");
                                                close(sock->fd);
                                                RELEASE_PACK(sock->qc->request);
                                                RELEASE(sock->qc, sizeof(queued_conn));
                                                sock->err = VFER_IMPL;
                                                goto disconnect_socket;
                                        }


                                        /* try to set the receive buffer sizes */
                                        num_packets = DEFAULT_RECV_BUF_MAXPACKETS;
                                        val = CALC_UDP_RECV_BUF_SIZE(num_packets);
                                        do {
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Trying to set SO_RCVBUF to %d for accepting socket", val);
                                                ret = setsockopt(sock->fd, SOL_SOCKET, SO_RCVBUF, &val, sizeof(val));
                                                if (ret != 0) {
                                                        if (errno == ENOBUFS && num_packets > 100) {
                                                                num_packets -= 100;
                                                                val = CALC_UDP_RECV_BUF_SIZE(num_packets);
                                                        } else {
                                                                perror("ControlT :: setsockopt");
                                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Could not set RCVBUF size to [%d]", val);
                                                                close(sock->fd);
                                                                RELEASE_PACK(sock->qc->request);
                                                                RELEASE(sock->qc, sizeof(queued_conn));
                                                                sock->err = VFER_IMPL;
                                                                goto disconnect_socket;
                                                        }
                                                }
                                        } while (ret != 0);
                                        sock->udp_recv_buf_size = val;

                                        /* try to set the send buffer size */
                                        num_packets = DEFAULT_SEND_BUF_MAXPACKETS;
                                        val = CALC_UDP_SEND_BUF_SIZE(num_packets);
                                        do {
                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Trying to set SO_SNDBUF to %d for accepting socket", val);
                                                ret = setsockopt(sock->fd, SOL_SOCKET, SO_SNDBUF, &val, sizeof(val));
                                                if (ret != 0) {
                                                        if (errno == ENOBUFS && num_packets > 100) {
                                                                num_packets -= 100;
                                                                val = CALC_UDP_SEND_BUF_SIZE(num_packets);
                                                        } else {
                                                                perror("ControlT :: setsockopt");
                                                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Could not set SNDBUF size to [%d]", val);
                                                                close(sock->fd);
                                                                RELEASE_PACK(sock->qc->request);
                                                                RELEASE(sock->qc, sizeof(queued_conn));
                                                                sock->err = VFER_IMPL;
                                                                goto disconnect_socket;
                                                        }
                                                }
                                        } while (ret != 0);
                                        sock->udp_send_buf_size = val;

                                        /* form response */
                                        if (sock->response != NULL) {
                                                /* sock->response is NULL on first itteration */
                                                RELEASE_PACK(sock->response);
                                        }

                                        sock->response = Packet_Form_Response(0, Packet_Gen_SeqNum(), Packet_Gen_FrameNum(), ntohs(sock->sa.sin_port),
                                                                              MAX_FRAME_SIZE, sock->udp_recv_buf_size, sock->udp_send_buf_size, sock->qc->request->delay);
                                        sock->response_tries = 0;
                                }

                                sock->last_response = tv;
                                sock->response_tries++;
                                DEBUG_PACK(DEBUG_CTL, "control.c", "ControlT", sock->response, "Try[%d] Sending [2/3 handshake] on sock->fd[%d]", sock->response_tries, sock->fd);

                                /* memcpy(&addr, &(sock->qc->sa), sizeof(sock->qc->sa));
                                DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "sending response to [%s : %d]",
                                inet_ntop(AF_INET, &(addr.sin_addr), buf, INET_ADDRSTRLEN), ntohs(addr.sin_port)); */

                                /* send response packet on handshake fd */
                                if ((ret = Packet_Write(sock->response, sock->listening_sock->fd, &(sock->qc->sa), sizeof(sock->qc->sa))) != 0) {
                                        /* ignore this connection, move on */
                                        DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "Try[%d] Packet_Write failed with error[%d]", sock->response_tries, ret);
                                        /* unconnect & rebind the socket */
                                        if (Unconnect_Rebind(sock) == -1) {
                                                /* couldn't rebind socket */
                                                close(sock->fd);
                                                RELEASE_PACK(sock->qc->request);
                                                RELEASE(sock->qc, sizeof(queued_conn));
                                                RELEASE_PACK(sock->response);
                                                sock->err = VFER_IMPL;
                                                goto disconnect_socket;
                                        }
                                        /* try the next queued connection */
                                        RELEASE_PACK(sock->response);
                                        sock->response = NULL;  /* this is crucial for the RELEASE_PACK above */
                                        sock->response_tries = MAX_RESPONSE_TRIES;
                                        goto try_next_conn;
                                }
                                break;

                        case CONN_DISCONNECTED:
                        case CONN_ACQUIRED:
                        case CONN_BOUND:
                                /* not an active socket */
                                pthread_mutex_unlock(&(sock->mutex));
                                continue;
                        } /* switch */

                        pthread_mutex_unlock(&(sock->mutex));
                        continue;

                disconnect_socket:
                        GET_TIME_OF_DAY(&(sock->stats.tv_end));
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "fd[%d] CONN_DISCONNECTED", sock->fd);
                        sock->state = CONN_DISCONNECTED;
                        if (sock->select_active == 1 && (sock->select_mark & VFER_EXCEPTION) == VFER_EXCEPTION) {
                                sock->select_result |= VFER_EXCEPTION;
                                pthread_mutex_unlock(&(sock->mutex));
                                pthread_mutex_lock(&(sockets.select_mutex));
                                pthread_cond_signal(&(sockets.select_cond));
                        } else {
                                pthread_mutex_unlock(&(sock->mutex));
                        }
                        Control_Remove_Socket(sock);
                        pthread_cond_signal(&(sock->cond));

                } /* while (i < num_active) */

                /****************/
                pthread_mutex_lock(&(sockets.mutex));
                if (sockets.num_active == 0) {
                        /* exits the control thread if there aren't any active sockets left */
                        DEBUG_PRINT(DEBUG_CTL, "control.c", "ControlT", "reached 0 active sockets, exiting control thread");
                        sockets.max_fd = 0;
                        pthread_mutex_unlock(&(sockets.mutex));
                        i = 0;
                        pthread_exit(&i);
                }
                /* set fds, and max_fd for next itteration */
                fds = sockets.fds;
                max_fd = sockets.max_fd;
                num_active = sockets.num_active;
                pthread_mutex_unlock(&(sockets.mutex));
                /****************/
        } /* while (1) */
} /* ControlT() */


/**
 * Sets up the socket structure with default parameters based on the
 * initial handshake packets (request/response/ack). It sets the state
 * of the socket to CONN_CONNECTED.
 *
 * @param sock socket pointer to a vfer_sock
 * @param request request packet corresponding to the new connection
 * @param response packet corresponding to the new connection
 * @param ack ack corresponding to the new connection
 * @param client specifies whether this connection was initiated with a request (client == 1) OR
 *        replied to and accepted with a response (client != 1)
 */
void Control_Add_Socket(vfer_sock* sock, packet* request, packet* response, packet* ack, char client) {
        ccontrol_t *c;
        struct timeval tv;
        int32_t t;
        int common_mtus[] = COMMON_MTU_LIST;
        int j;
        
        c = &(sock->ccontrol);

        /* sock structure specific settings */
        c->mtu = INIT_MTU;
        sock->relfun = rf_always_reliable;      /* full reliability by default for all new connections */
        
        /* PMTU: Initialize path mtu values for socket */
        if(sock->is_df_set != 1){
                /* ivan: we know we can set icmp_fd to -1 here, but where else does it get set for accepted sockets if PMTUD is done without ICMP? */
                sock->icmp_fd = -1;
                sock->pmtu.probe_state = NO_PMTUD;
                DEBUG_PMTUD_PRINT(DEBUG_PMTUD, "control.c", "Control_Add_Socket", "Not performing PMTUD");
        }else{
                sock->pmtu.probe_state = NO_PROBE_SENT;
                DEBUG_PMTUD_PRINT(DEBUG_PMTUD, "control.c", "Control_Add_Socket", "Initial PMTUD values set.");
                for(j = 0; j <= COMMON_MTU_MAX_INDEX; j++){
                        sock->pmtu.common_mtus[j] = common_mtus[j];
                }
                sock->pmtu.search_low = sock->pmtu.common_mtus;
                sock->pmtu.search_high = sock->pmtu.common_mtus + COMMON_MTU_MAX_INDEX;
                sock->pmtu.probe_size = sock->pmtu.search_low + 1;
        }
        
        CLR_SOCKET_STATS(&(sock->stats));

        /* record the current timestamp as a connection start */
        GET_TIME_OF_DAY(&(sock->stats.tv_start));

        /* client/server specific settings */
        if (client == 1) {
                /* sock->l_seq is set in vfer_connect() ; uses ack packet */
                // printf("CLIENT delay curr_min[%d] base_min[%d]\n", Tree_Min(&(sock->ccontrol.rev_path_curr_delayhist), sock->stats.tv_start),
                // Tree_Min(&(sock->ccontrol.rev_path_base_delayhist), sock->stats.tv_start));

                sock->send_frames.first_frame_num = sock->send_frames.last_frame_num = request->u.req.init_frame_num;
                sock->recv_frames.first_frame_num = sock->recv_frames.last_frame_num = response->u.res.init_frame_num;
                sock->r_opts            = response->opts;
                sock->l_frame_num       = request->u.req.init_frame_num;
                sock->r_seq             = response->seq + 1;
                sock->r_frame_num       = response->u.res.init_frame_num;
                sock->r_recv_buf_size   = response->u.res.recv_buf_size;
                sock->r_send_buf_size   = response->u.res.send_buf_size;
                sock->r_max_frame_size  = response->u.res.max_frame_size;
                sock->stats.max_rtt     = abs(response->u.res.request_delay - response->delay);
                c->delay                = response->u.res.request_delay;
                if (sock->stats.max_rtt == 0) {
                        printf("RTT EMPTY\n");
                        sock->stats.max_rtt = 5000; /* 5ms */
                }

                /* printf("max_rtt [%d]\n", sock->stats.max_rtt); */
                /* set the base delay window, TREE_INIT defined in
                   globals.h t is in microseconds just like max_rtt */
                t = sock->stats.max_rtt * 100;
                TREE_INIT(&(c->rev_path_base_delayhist), t, tv);

                /* set the current delay window */
                t = sock->stats.max_rtt * 10;
                TREE_INIT(&(c->rev_path_curr_delayhist), t, tv);

                /* set the initial reverse path delay values from the response packet */
                Tree_Insert(&(sock->ccontrol.rev_path_base_delayhist), response->delay, sock->stats.tv_start);
                Tree_Insert(&(sock->ccontrol.rev_path_curr_delayhist), response->delay, sock->stats.tv_start);
                //sock->delay = response->delay;
        } else {
                /* sock->r_seq is set in Control_Accept() ; uses ack packet from client */
                // printf("Server delay curr_min[%d] base_min[%d]\n", Tree_Min(&(sock->ccontrol.rev_path_curr_delayhist), sock->stats.tv_start),
                // Tree_Min(&(sock->ccontrol.rev_path_base_delayhist), sock->stats.tv_start));

                sock->send_frames.first_frame_num = sock->send_frames.last_frame_num = response->u.res.init_frame_num;
                sock->recv_frames.first_frame_num = sock->recv_frames.last_frame_num = request->u.req.init_frame_num;
                sock->r_opts            = request->opts;
                sock->r_max_frame_size  = request->u.req.max_frame_size;
                sock->r_recv_buf_size   = request->u.req.recv_buf_size;
                sock->r_send_buf_size   = request->u.req.send_buf_size;
                sock->r_frame_num       = request->u.req.init_frame_num;
                sock->l_seq             = response->seq + 1;
                sock->l_frame_num       = response->u.res.init_frame_num;
                sock->stats.max_rtt     = abs(ack->u.ccack.delay - ack->delay);
                c->delay                = ack->u.ccack.delay;
                if (sock->stats.max_rtt == 0) {
                        printf("RTT EMPTY\n");
                        sock->stats.max_rtt = 1;
                }

                /* set the base delay window */
                t = sock->stats.max_rtt * 100;
                TREE_INIT(&(c->rev_path_base_delayhist), t, tv);

                /* set the current delay window */
                t = sock->stats.max_rtt * 10;
                TREE_INIT(&(c->rev_path_curr_delayhist), t, tv);

                /* set the initial reverse path delay values from the response and ack packets */
                Tree_Insert(&(sock->ccontrol.rev_path_base_delayhist), request->delay, sock->stats.tv_start);
                Tree_Insert(&(sock->ccontrol.rev_path_curr_delayhist), request->delay, sock->stats.tv_start);
                
                Tree_Insert(&(sock->ccontrol.rev_path_base_delayhist), ack->delay, sock->stats.tv_start);
                Tree_Insert(&(sock->ccontrol.rev_path_curr_delayhist), ack->delay, sock->stats.tv_start);

                //sock->delay = ack->delay;
                
        }

        /* stats structure settings (max_rtt set above in the conditional blocks) */
        sock->stats.avg_rtt                     = sock->stats.max_rtt;
        sock->stats.min_rtt                     = sock->stats.max_rtt;
        sock->ccontrol.rtt                      = sock->stats.max_rtt;
        sock->ccontrol.syncing                  = 0;
        sock->stats.rtt_measurements            = 1;
        sock->stats.data_bytes_sender_sent      = 0;

        // printf("FIRST ESTIMATED RTT: %f\n", sock->stats.avg_rtt);

        /* ccontrol structure settings */
        if (client == 1) {
                c->s_first_seq  = request->seq + 2; /* handshake ack packet seq number */
                c->s_last_seq   = request->seq + 2;
        } else {
                c->s_first_seq  = response->seq + 1;
                c->s_last_seq   = response->seq + 1;
        }
        /* set up max and current window size (in bytes) based on remote recv buf size and local UDP send buf size  */
        if (sock->udp_send_buf_size < sock->r_recv_buf_size) {
                c->s_maxw = (int)(sock->udp_send_buf_size);
        } else {
                c->s_maxw = (int)(sock->r_recv_buf_size);
        }

        c->s_cwnd               = INIT_MTU * 4;
        c->s_cwnd_sent          = 0;
        c->s_inflight           = 0;
        c->delay_delta          = 0;
        c->r_packets_lost       = 0;
        c->last_packets_lost    = 0;
        c->r_bytes_total        = 0;
        c->last_bytes_total     = 0;

        if (client == 1) {
                c->r_first_seq  = response->seq+1;
                c->r_last_seq   = response->seq+1;
        } else {
                c->r_first_seq  = request->seq+2;
                c->r_last_seq   = request->seq+2;
        }
        GET_TIME_OF_DAY(&(tv));
        c->r_last_ccack = tv;
        c->r_last_data  = tv;
        c->s_rtt_start  = tv;
        c->s_last_ack   = tv;
        c->s_last_unacked_data.tv_sec   = 0;
//      c->s_last_vec.tv_sec            = 0;
        c->r_last_loss.tv_sec           = 0;
        c->s_need_ccack = 0;

        DEBUG_CC(DEBUG_CTL, "control.c", "Add_Socket", sock, "CC_Init");

        /* sock structure settings */
        sock->send_frames.first = sock->send_frames.last        = NULL;
        sock->recv_frames.first = sock->recv_frames.last        = NULL;
        sock->recvd_frames.first        = sock->recvd_frames.last       = NULL;
        sock->send_frames.num           = 0;
        sock->recv_frames.num           = 0;
        sock->recvd_frames.num          = 0;
        sock->recvd_frames.length       = 0;
        sock->recv_frames.length        = 0;
        sock->send_frames.length        = 0;

        /*************/
        pthread_mutex_lock(&(sock->mutex));
        sock->state = CONN_CONNECTED;
        pthread_mutex_unlock(&(sock->mutex));
        /*************/

        pthread_mutex_lock(&(sockets.mutex));
        FD_SET(sock->fd, &(sockets.fds));
        if (sockets.max_fd < sock->fd) {
                sockets.max_fd = sock->fd;
        }
        pthread_mutex_unlock(&(sockets.mutex));
} /* Add_Socket() */

/**
 * Removes a socket from the sockets linked list Control, a thread
 * safe function
 *
 * @param sock is a pointer to an vfer_sock
 */
static void Control_Remove_Socket(vfer_sock* sock) {

        DEBUG_PRINT(DEBUG_CTL, "control.c", "Control_Remove_Socket", "Entered with sock->fd[%d]", sock->fd);

        Control_Unregister_Socket(sock);

        /****************/
        pthread_mutex_lock(&(sock->mutex));
        /* free up all the alloced send frames */
        Datagrams_Clear(&(sock->send_frames));
        /* free up all the alloced recv frames */
        Datagrams_Clear(&(sock->recv_frames));

        /* NOTE: we do NOT clear the recvd_frames list! */
        pthread_mutex_unlock(&(sock->mutex));
        /****************/
        return;
} /* Remove_Socket() */

/**
 * Unconnects an vfer_socket and rebinds it
 *
 * Internal use only, this function is used only by Control_Accept()
 *
 * @param sock a pointer to a socket structure
 * @return
 *      0 on success
 *      -1 on error
 */
static int Unconnect_Rebind(vfer_sock* sock) {
        /* for sake of portability, we'll disconnect the socket using AF_UNSPEC
           & also rebind it so that we know it has an ip and port associated with it */
        socklen_t socklen;

        bzero((char*)(&sock->sa), sizeof(sock->sa));
        sock->sa.sin_family = AF_UNSPEC;
        /* per R.Stevens: various platforms respond
         * differently, so we ignore the result of
         * this call */
        if (connect(sock->fd, (SA*)(&sock->sa), sizeof(sock->sa)) == -1) {
                perror("unconnect_rebind[unconnect]");
        }

        /* rebind socket as a wildcard */
        bzero((char*)(&sock->sa), sizeof(sock->sa));
        sock->sa.sin_port               = htons(INADDR_ANY);
        sock->sa.sin_addr.s_addr        = htonl(INADDR_ANY);
        sock->sa.sin_family             = AF_INET;
        if (bind(sock->fd, (SA*)(&sock->sa), sizeof(sock->sa)) < 0) {
                perror("unconnect_rebind[bind for the new socket]");
                return -1;
        }

        socklen = sizeof(sock->sa);
        if (getsockname(sock->fd, (SA*)&(sock->sa), &socklen) == -1) {
                perror("unconnect_rebind[getsockname for new socket]");
                return -1;
        }

        return 0;
} /* Unconnect_Rebind() */

---