time-protocol.h

#include <assert.h>
#include "util/time.h"

#ifdef XLAP
#  error "does not support XLAP right now"
#endif

#ifdef TCP
#
#  define PROTOCOL "tcp"
#  include <sys/types.h>
#  include <sys/socket.h>
#  include <netdb.h>
#  include <unistd.h>
#  include <netinet/in.h>
#  include <netinet/tcp.h>

typedef int socket_t;

static inline void _time() {
    long long       ns; // Milliseconds
    time_t          s;  // Seconds
    struct timespec spec;

    clock_gettime(CLOCK_REALTIME, &spec);

    s  = spec.tv_sec;
    ns = spec.tv_nsec;
    printf("T: %09ld.%09ld\n", s, ns);
}

static inline socket_t _open_sender(struct arguments *args)
{
    struct addrinfo hints = {
      .ai_socktype = SOCK_STREAM,
      .ai_family = AF_UNSPEC,
      .ai_flags = AI_ADDRCONFIG,
    };

    char port[16];
    snprintf(port, sizeof(port), "%d", (int) args->port);

    struct addrinfo *head;
    int fail = getaddrinfo(args->target, port, &hints, &head);
    assert(!fail);
    int sock;
    struct addrinfo *iter;
    int send_buffer_size = args->sndbuf;
    uint32_t send_buffer_size_len = sizeof(send_buffer_size);

    for (iter = head; iter != NULL; iter = iter->ai_next) {
        sock = socket(iter->ai_family, iter->ai_socktype, iter->ai_protocol);

        if(args->optimized) {
          // Disable Nagles algorithm.
          int flag = 1;
          setsockopt(sock, SOL_TCP, TCP_NODELAY, &flag, sizeof(flag));
          setsockopt(sock, SOL_SOCKET, SO_SNDBUF, &send_buffer_size, send_buffer_size_len);
          getsockopt(sock, SOL_SOCKET, SO_SNDBUF, &send_buffer_size, &send_buffer_size_len);
          printf("SO_SNDBUF: %d\n", send_buffer_size);
        }
        int connect_res = connect(sock, iter->ai_addr, iter->ai_addrlen);
        if (connect_res == 0) {
            break;
        } else {
            printf("CONNECT ERROR: %d, errno: %d\n", connect_res, errno);
        }
        close(sock);
    }
    assert(iter && "failed to connect via TCP");
    freeaddrinfo(head);
    return sock;
}

static inline socket_t _open_receiver(struct arguments *args)
{
    int ssock = socket(AF_INET, SOCK_STREAM, 0);
    assert(ssock >= 0);

    if(args->optimized) {
      int yes = 1;
      setsockopt(ssock, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes));
      setsockopt(ssock, SOL_TCP, TCP_QUICKACK, &yes, sizeof(yes));

      int recv_buffer_size = args->rcvbuf;
      uint32_t recv_buffer_size_len = sizeof(recv_buffer_size);
      setsockopt(ssock, SOL_SOCKET, SO_RCVBUF, &recv_buffer_size, recv_buffer_size_len);
      getsockopt(ssock, SOL_SOCKET, SO_RCVBUF, &recv_buffer_size, &recv_buffer_size_len);
      printf("SO_RCVBUF: %d\n", recv_buffer_size);
    }

    struct sockaddr_in address = {
        .sin_family = AF_INET,
        .sin_port = htons(args->port),
    };
    memset(address.sin_zero, '\0', sizeof(address.sin_zero));
    address.sin_addr.s_addr = htonl(INADDR_ANY);
    assert(0 == bind(ssock, (const struct sockaddr *) &address, sizeof(address)));

    assert(0 == listen(ssock, SOMAXCONN));

    int sock = accept(ssock, NULL, NULL);
    assert(sock >= 0);

    int res = close(ssock);
    assert(res >= 0);

    return sock;
}

static inline ssize_t _send(socket_t conn, const char *buf, size_t size)
{
    int written = 0;
    ssize_t n = 0;
    while (written < size) {
        n = send(conn, buf, size - written, 0);
        if (n < 0) {
            _time();
            printf("WRITE FAILED: %ld, errno: %d\n", n, errno);
            printf("%s\n", buf - written);
            return written;
        }
        written += n;
        buf += n;
    }
    return written;
}

static inline ssize_t _recv(socket_t conn, char *buf, size_t size)
{
    ssize_t received = 0;
    ssize_t n = 0;
    while (received < size && keepRunning) {
        fd_set rfds;
        struct timeval tv;
        int retval;
        FD_ZERO(&rfds);
        FD_SET(conn, &rfds);

        tv.tv_sec = 1;
        tv.tv_usec = 0;
        retval = select(conn + 1, &rfds, NULL, NULL, &tv);
        if (!FD_ISSET(conn, &rfds)) {
            continue;
        }
        if (!keepRunning) {
            printf("No longer running.\n");
            return -1;
        }

        if (retval == -1) {
            printf("Select failed: %d.\n", errno);
            return -1;
        } else if (retval > 0) {
            n = recv(conn, buf, size - received, 0);
            if (n < 0) {
                _time();
                printf("recv failed: %ld, errno: %d, received: %ld, size: %lu\n", n, errno, received, size);
                return -1;
            }
            if (n == 0) {
                break;
            }
            buf += n;
            received += n;
        } else {
            continue;
        }
    }
    return received;
}

static inline void _close(socket_t conn)
{
    char buffer[10];
    int res;
    shutdown(conn, SHUT_WR);
    for(;;) {
        res = read(conn, buffer, 1);
        if(res < 0) {
            perror("reading");
            exit(1);
        }
        if(!res)
            break;
    }
    res = close(conn);
    assert(res >= 0);
}

static inline long long int _send_peer_btl_pace(socket_t conn) {
    return 0;
}

static inline long long int _recv_peer_btl_pace(socket_t conn) {
    return 0;
}

static inline long long int _nw_pace(socket_t conn) {
    return 0;
}

static inline long long int _app_send_pace_internal(socket_t conn) {
    return 0;
}

static inline long long int _app_send_pace_dependent(socket_t conn) {
    return 0;
}

static inline long long int _app_send_pace_external(socket_t conn) {
    return 0;
}

static inline long long int _trans_transmit_pace_internal(socket_t conn) {
    return 0;
}

static inline long long int _trans_transmit_pace_dependent(socket_t conn) {
    return 0;
}

static inline long long int _trans_transmit_pace_external(socket_t conn) {
    return 0;
}

static inline long long int _trans_receive_pace_internal(socket_t conn) {
    return 0;
}

static inline long long int _trans_receive_pace_dependent(socket_t conn) {
    return 0;
}

static inline long long int _trans_receive_pace_external(socket_t conn) {
    return 0;
}

static inline long long int _app_deliver_pace_internal(socket_t conn) {
    return 0;
}

static inline long long int _app_deliver_pace_dependent(socket_t conn) {
    return 0;
}

static inline long long int _app_deliver_pace_external(socket_t conn) {
    return 0;
}


#else
#  define PROTOCOL "prrt"

typedef PrrtSocket *socket_t;

static inline socket_t _open_sender(struct arguments *args) {
    PrrtSocket *s = PrrtSocket_create(args->size, 100 * 1000);
    s->withTimestamp = true;
    assert(s != NULL && "Socket create failed.");

    if (args->thread_pinning) {
        PrrtSocket_enable_thread_pinning(s);
    }
    PrrtSocket_set_coding_parameters(s, 1, 1, 0, NULL);

    assert(PrrtSocket_bind(s, "0.0.0.0", args->local_port) && "bind failed");

    PrrtSocket_connect(s, args->target, args->port);
    return s;
}

static inline socket_t _open_receiver(struct arguments *args) {
    socket_t s = PrrtSocket_create(args->size, HALF_TIMESTAMP - 1);
    assert(s != NULL && "Could not create socket.");

    if (args->thread_pinning) {
        PrrtSocket_enable_thread_pinning(s);
    }

    assert(PrrtSocket_bind(s, "0.0.0.0", args->port) && "bind failed");
    return s;
}

static inline ssize_t _send(socket_t conn, const char *buf, size_t size) {
    PrrtSocket_send_sync(conn, (unsigned char *) buf, size);
    return size;
}

static inline ssize_t _recv(socket_t conn, char *buf, size_t size) {
    struct sockaddr_in addr;
    return PrrtSocket_receive_asap_wait(conn, buf, (struct sockaddr*) &addr);
}

static inline void _close(socket_t conn) {
    PrrtSocket_close(conn);
    free(conn);
}

static inline long long int _send_peer_btl_pace(socket_t conn) {
    return PrrtSocket_get_sock_opt(conn, "send_peer_btl_pace");
}

static inline long long int _recv_peer_btl_pace(socket_t conn) {
    return PrrtSocket_get_sock_opt(conn, "recv_peer_btl_pace");
}

static inline long long int _nw_pace(socket_t conn) {
    return PrrtSocket_get_sock_opt(conn, "nw_pace");
}

static inline long long int _app_send_pace_internal(socket_t conn) {
    return PrrtSocket_get_sock_opt(conn, "appSend_pace_internal");
}

static inline long long int _app_send_pace_dependent(socket_t conn) {
    return PrrtSocket_get_sock_opt(conn, "appSend_pace_dependent");
}

static inline long long int _app_send_pace_external(socket_t conn) {
    return PrrtSocket_get_sock_opt(conn, "appSend_pace_external");
}

static inline long long int _trans_transmit_pace_internal(socket_t conn) {
    return PrrtSocket_get_sock_opt(conn, "prrtTransmit_pace_internal");
}

static inline long long int _trans_transmit_pace_dependent(socket_t conn) {
    return PrrtSocket_get_sock_opt(conn, "prrtTransmit_pace_dependent");
}

static inline long long int _trans_transmit_pace_external(socket_t conn) {
    return PrrtSocket_get_sock_opt(conn, "prrtTransmit_pace_external");
}

static inline long long int _trans_receive_pace_internal(socket_t conn) {
    return PrrtSocket_get_sock_opt(conn, "prrtReceive_pace_internal");
}

static inline long long int _trans_receive_pace_dependent(socket_t conn) {
    return PrrtSocket_get_sock_opt(conn, "prrtReceive_pace_dependent");
}

static inline long long int _trans_receive_pace_external(socket_t conn) {
    return PrrtSocket_get_sock_opt(conn, "prrtReceive_pace_external");
}

static inline long long int _app_deliver_pace_internal(socket_t conn) {
    return PrrtSocket_get_sock_opt(conn, "appDeliver_pace_internal");
}

static inline long long int _app_deliver_pace_dependent(socket_t conn) {
    return PrrtSocket_get_sock_opt(conn, "appDeliver_pace_dependent");
}

static inline long long int _app_deliver_pace_external(socket_t conn) {
    return PrrtSocket_get_sock_opt(conn, "appDeliver_pace_external");
}

#endif

static inline socket_t _open(bool sender, struct arguments *args) {
    if (sender)
        return _open_sender(args);
    else
        return _open_receiver(args);
}