comm.c

Go to the documentation of this file.

#include <config.h>

#include <stdio.h>
#include <unistd.h>
#include <string.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <netdb.h>
#include <fcntl.h>
#include <sys/ioctl.h>
#include <errno.h>

#define DEBUG_LEVEL 1
#include "debug.h"
#include <cartcomm/comm.h>
#include <cartcomm/cframe.h>


int cartcomm_connect(char *host, int port) {
    int sockfd, result;
    struct hostent *server_he;
    struct sockaddr_in server_addr;
    int true_flag = 1;
    struct timeval timeout;

    server_he = gethostbyname(host);
    if(server_he == NULL) {
        DEBUG1("Error looking up host!");
        return -1;
    }

    sockfd = socket(AF_INET, SOCK_STREAM, 0);
    if(sockfd < 0) {
        DEBUG1("Error opening socket");
        return -1;
    }

    // Control the maximum waiting on a socket
    timeout.tv_sec = 0;
    timeout.tv_usec = 2 * COMM_PERIOD_US;
    result = setsockopt(sockfd, SOL_SOCKET, SO_RCVTIMEO, (char *) &timeout,
                        sizeof(timeout));
    result += setsockopt(sockfd, SOL_SOCKET, SO_SNDTIMEO, (char *) &timeout,
                        sizeof(timeout));
    if(result < 0) {
        DEBUG1("Couldn't set socket timeouts!");
        close(sockfd);
        return -1;
    }

    bzero(&server_addr, sizeof(server_addr));
    server_addr.sin_family = AF_INET;
    bcopy(server_he->h_addr, &server_addr.sin_addr.s_addr, server_he->h_length);
    server_addr.sin_port = htons(port);

    result = connect(sockfd, (struct sockaddr *) &server_addr,
                     sizeof(server_addr));
    if(result < 0) {
        DEBUG2("Error connecting to server");
        close(sockfd);
        return -1;
    }

    // Request there be no delaying of coalescing of TCP packets
    result = setsockopt(sockfd, IPPROTO_TCP, TCP_NODELAY, (char *) &true_flag,
                        sizeof(true_flag));
    if(result < 0) {
        DEBUG1("Couldn't set TCP_NODELAY!");
        close(sockfd);
        return -1;
    }

    return sockfd;
}


void cartcomm_disconnect(int sockfd) {
    close(sockfd);
}


int cartcomm_synchronize(int sockfd) {
    int retries = MAX_RETRIES;
    char buf[4096];
    int size;
    cframe_t cf;
    int result;

    bzero(buf, sizeof(buf));
    bzero(&cf, sizeof(cframe_t));

    ioctl(sockfd, FIONREAD, &size);
    DEBUG2("bytes in buffer = %d", size);

    // Empty the socket buffer
    while(size > 0) {
        recv(sockfd, buf, sizeof(buf), MSG_DONTWAIT);
        ioctl(sockfd, FIONREAD, &size);
        DEBUG2("bytes in buffer = %d", size);
    }

    for(;;) {
        DEBUG1("Waiting for connection to be ready..");

        result = cartcomm_read_msg(sockfd, buf, sizeof(buf));
        if(result < 0) {
            DEBUG1("read_msg failed!");
            continue;
        }
        cartcomm_send_msg(sockfd, buf);

        cartcomm_decode_msg(&cf, buf);
        if(strcmp(cf.status, "ready") == 0)
            break;

        retries--;
        if(retries <= 0)
            return -1;
    }

    return 0;
}


int cartcomm_connect_sync(char *host, int port) {
    int sockfd;
    int result;

    sockfd = cartcomm_connect(host, port);
    if(sockfd < 0) {
        DEBUG2("Couldn't connect to server!");
        return sockfd;
    }

    result = cartcomm_synchronize(sockfd);
    if(result < 0) {
        DEBUG2("Couldn't synchronize to server!\n");
        cartcomm_disconnect(sockfd);
        return result;
    }

    return sockfd;
}


int cartcomm_read_msg(int sockfd, char *buf, int bufsize) {
    int i;
    int size;

start_over:
    // Block for recieving the first byte of the message
    size = recv(sockfd, buf, 1, MSG_PEEK | MSG_WAITALL);
    if(size <= 0) {
        DEBUG1("recv failed!");
        return -1;
    }

    // Nonblocking to find the end of the message
    for(;;) {
        size = recv(sockfd, buf, MAX_MSG_SIZE, MSG_PEEK | MSG_DONTWAIT);
        if(size == 0)
            goto start_over;
        if(size < 0) {
            DEBUG1("recv failed (%s)", strerror(errno));
            continue;
        }

        if(buf[0] != '{') {
            for(i = 0; i < size; i++) {
                if(buf[i] == '{') {
                    // Throw away junk data
                    DEBUG1("discarded a partial message");
                    recv(sockfd, buf, i, 0);
                    break;
                }
            }
            continue;
        }

        for(i = 1; i < size; i++) {
            if(buf[i] == '{') {
                // Throw away junk data
                DEBUG1("discarded a missed message");
                recv(sockfd, buf, i, 0);
                break;
            }
        }

        for(i = 0; i < size; i++) {
            if(buf[i] == '}') {
                // Remove message from socket
                recv(sockfd, buf, i + 1, 0);
                buf[i + 1] = 0;
                return 0;
            }
        }
    }

    return 0;
}


int cartcomm_send_msg(int sockfd, char *buf) {
    int retries = MAX_RETRIES;
    int size = -1;

    while(retries > 0 && size < 0) {
        size = send(sockfd, buf, strlen(buf), 0);
        if(size < strlen(buf)) {
            retries--;
            continue;
        }
    }

    if(retries > 0)
        return 0;
    else {
        DEBUG1("too many retries");
        return -1;
    }
}

---