/*
 * WinSock implementation of Inter-process Communication API.
 *
 * @copyright 2008, Research In Motion Ltd.
 *
 * @version $Id: //depot/main/dsp/ecall/c/testlab/ipc_sock.c#5 $
 */

#include <winsock2.h>
#include <ws2tcpip.h>
#include <stdio.h>
#include <stddef.h>
#include "ipc.h"
#include "msg.h"
#include "error.h"

/* IVS, PSAP, UL, DL => 4 connections in CONTROL */
#define MAX_IPC_CONN 4

typedef enum {
    IpcDirInbound,  /* listen for incoming connections (server) */
    IpcDirOutbound  /* establish outgoing connection (client) */
} IpcDir;

typedef struct _IpcConn {
    SOCKET  sock;
    IpcDir  dir;
} IpcConn;

static int nIpcConn;                    /* number of open connections */
static IpcConn ipcConn[MAX_IPC_CONN];   /* list of connections */


void IpcInit()
{
    WSADATA wsaData;
    Assert(
        WSAStartup(MAKEWORD(2, 2), &wsaData) == 0, 
        "IpcInit: WSAStartup failed");
    nIpcConn = 0;
}

void IpcFinal(void)
{
    IpcStop();
    WSACleanup();
}

/* Create a socket for either listen or connect */
static IpcConn *ipcOpen(const char *host, const char *port, IpcDir dir)
{
    struct addrinfo hints;
    struct addrinfo *res = NULL;
    SOCKET sock;
    int flag;
    IpcConn * conn = NULL;

    Assert(nIpcConn < MAX_IPC_CONN, "ipcOpen: Too many connections");

    ZeroMemory(&hints, sizeof(hints));
    hints.ai_family = AF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;
    hints.ai_protocol = IPPROTO_TCP;
    if(dir == IpcDirInbound) {
        hints.ai_flags = AI_PASSIVE;
    }
    if(getaddrinfo(host, port, &hints, &res) != 0) {
        goto cleanup;
    }

    sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
    if(sock == INVALID_SOCKET) {
        goto cleanup;
    }
    if(dir == IpcDirInbound) {
        if(bind(sock, res->ai_addr, (int)res->ai_addrlen) == SOCKET_ERROR) {
            goto cleanup;
        }
        /* We don't expect multiple connections */
        if(listen(sock, 1) == SOCKET_ERROR) {
            goto cleanup;
        }
    }
    else {
        if(connect(sock, res->ai_addr, (int)res->ai_addrlen) == SOCKET_ERROR) {
            goto cleanup;
        }
    }

    /* 
     * We care about latency since we are sending many relatively small
     * messages and waiting for replies. Performance depends upon the
     * round-trip time. So it makes sense to disable the Nagle algorithm. 
     */
    flag = 1;
    Assert(
        setsockopt(
            sock,
            IPPROTO_TCP,
            TCP_NODELAY,
            (char *)&flag,
            sizeof(flag)) == 0,
        "ipcOpen: setsockopt failed");

    /* Socket is ready for use */
    conn = &ipcConn[nIpcConn];
    conn->sock = sock;
    conn->dir = dir;
    nIpcConn += 1;

cleanup:
    if(res != NULL) freeaddrinfo(res);
    if(conn == NULL && sock != INVALID_SOCKET) closesocket(sock);
    return conn;
}

IpcConn *IpcListen(const char *host, const char *port)
{
    return ipcOpen(host, port, IpcDirInbound);
}

IpcConn *IpcConnect(const char *host, const char *port)
{
    return ipcOpen(host, port, IpcDirOutbound);
}

void IpcRegisterTask(IpcConn *conn, BOOL (*task)(void))
{
/* Do nothing */
}

void IpcStart(void)
{
    int i;
    for(i = 0; i < nIpcConn; ++i) {
        /* Accept a connection from any sockets that are inbound */
        if(ipcConn[i].dir == IpcDirInbound) {
            SOCKET sock;
            sock = accept(ipcConn[i].sock, NULL, NULL);
            Assert(sock != INVALID_SOCKET, "IpcStart: accept failed");

            /* Close the listen socket: we only ever accept 1 connection */
            closesocket(ipcConn[i].sock);
            /* From now on, use the client socket */
            ipcConn[i].sock = sock;
        }
    }
}

void IpcStop(void)
{
    int i;
    for(i = 0; i < nIpcConn; ++i) {
        shutdown(ipcConn[i].sock, SD_BOTH);
        closesocket(ipcConn[i].sock);
    }
    nIpcConn = 0;
}

/* Blocking "all or nothing" receive */
static int recvall(SOCKET sock, char *buf, int len, int flags)
{
    int remain = len;
    while(remain > 0) {
        int n;
        n = recv(sock, buf, remain, flags);
        if(n < 1) {
            return n;
        }
        remain -= n;
        buf += n;
    }
    return len;
}

IpcConn *IpcRecv(IpcConn *conn, Msg *msg)
{
    int nBytes;

    if(conn == NULL || msg == NULL) return NULL;

    /* Expect either a complete message or a closed socket */
    nBytes = recvall(conn->sock, &msg->id, sizeof(msg->id), 0);
    if(nBytes != sizeof(msg->id)) return NULL;
    Assert(msg->id < MsgIdCount, "IpcRecv: invalid message id");
    nBytes = recvall(
        conn->sock, 
        (char *)&msg->u, 
        MsgDataSize[msg->id], 
        0);
    if(nBytes != MsgDataSize[msg->id]) {
        Abort("IpcRecv: incomplete message received");
    }

    return conn;
}

void IpcSend(IpcConn *conn, const Msg *msg)
{
    int nBytes;

    if(msg == NULL || conn == NULL) return;
    Assert(msg->id < MsgIdCount, "IpcSend: invalid message id");

    nBytes = send(conn->sock, &msg->id, sizeof(msg->id), 0);
    Assert(nBytes == sizeof(msg->id), "IpcSend: send failed");
    nBytes = send(conn->sock, (char *)&msg->u, MsgDataSize[msg->id], 0);
    Assert(nBytes == MsgDataSize[msg->id], "IpcSend: send failed");
}

#ifdef __TEST__

#include <time.h>

/* Fill in msg with id and random payload data based on seed */
static void fillMsg(Msg *msg, MsgId id, Ord8 seed)
{
    int i;
    Ord8 *d = (Ord8 *)&msg->u;

    msg->id = id;
    for(i = 0; i < MsgDataSize[id]; ++i) {
        d[i] = seed;
        seed = ~((seed >> 1) | (seed << 7));
    }
}

/* Check that msg has correct id and payload data */
static void checkMsg(Msg *msg, MsgId id, Ord8 seed)
{
    int i;
    Ord8 *d = (Ord8 *)&msg->u;

    Assert(msg->id == id, "checkMsg: id mismatch");
    for(i = 0; i < MsgDataSize[id]; ++i) {
        Assert(d[i] == seed, "checkMsg: payload mismatch");
        seed = ~((seed >> 1) | (seed << 7));
    }
}

static void client(const char *host)
{
    Msg msg;
    IpcConn *rx;

    if(IpcConnect(host, "10000") == NULL) Abort("IpcConnect failed");
    if(IpcConnect(host, "10001") == NULL) Abort("IpcConnect failed");
    if(IpcConnect(host, "10002") == NULL) Abort("IpcConnect failed");
    if(IpcConnect(host, "10003") == NULL) Abort("IpcConnect failed");

    /* Loopback */
    IpcStart();
    while(1) {
        rx = IpcRecv(&msg);
        if(!rx) break;
        IpcSend(rx, &msg);
    }
    IpcStop();
}

static void server(const char *host)
{
    IpcConn *conn[MAX_IPC_CONN];
    Msg txMsg;
    Msg rxMsg;
    Ord8 seed;
    int i;
    int j;
    int k;
    int startMs;
    int stopMs;

    for(i = 0; i < MAX_IPC_CONN; ++i) {
        char port[6];
        sprintf(port, "%d", 10000 + i);

        conn[i] = IpcListen(host, port);
        if(conn[i] == NULL) {
            Abort("IpcListen failed");
        }
    }

    IpcStart();
    
    printf("Testing all message types sequentially...\n");

    for(i = 0; i < 10; ++i) {
        for(j = 0; j < MAX_IPC_CONN; ++j) {
            for(k = 0; k < MsgIdCount; ++k) {
                fillMsg(&txMsg, k, j + k);
                IpcSend(conn[j], &txMsg);
                if(!IpcRecv(&rxMsg)) {
                    Abort("Connection closed");
                }
                checkMsg(&rxMsg, k, j + k);
            }
        }
    }

    printf("Testing all message types concurrently...\n");

    for(i = 0; i < 10; ++i) {
        int rxTable[MAX_IPC_CONN][MsgIdCount];
        int rxRemain;

        for(j = 0; j < MAX_IPC_CONN; ++j) {
            for(k = 0; k < MsgIdCount; ++k) {
                fillMsg(&txMsg, k, i + j + k);
                IpcSend(conn[j], &txMsg);
            }
        }

        ZeroMemory(rxTable, sizeof(rxTable));
        for(rxRemain = MAX_IPC_CONN * MsgIdCount; rxRemain > 0;) {
            IpcConn *rxConn = IpcRecv(&rxMsg);
            if(rxConn == NULL) Abort("Connection closed");
            for(j = 0; j < MAX_IPC_CONN; ++j) {
                if(rxConn == conn[j]) {
                    k = rxMsg.id;
                    if(rxTable[j][k]) Abort("Duplicate message received");
                    checkMsg(&rxMsg, k, i + j + k);
                    rxTable[j][k] = 1;
                    rxRemain--;
                    break;
                }
            }
        }
    }

    printf("Measuring MsgPcmData overhead\n");

    startMs = (clock() * 1000) / CLOCKS_PER_SEC;
    for(i = 0; i < 10000; ++i) {
        int rxTable[MAX_IPC_CONN];
        int rxRemain;

        for(j = 0; j < MAX_IPC_CONN; ++j) {
            fillMsg(&txMsg, MsgPcmData, i + j);
            IpcSend(conn[j], &txMsg);
        }
        
        ZeroMemory(rxTable, sizeof(rxTable));
        for(rxRemain = MAX_IPC_CONN; rxRemain > 0;) {
            IpcConn *rxConn = IpcRecv(&rxMsg);
            if(rxConn == NULL) Abort("Connection closed");
            for(j = 0; j < MAX_IPC_CONN; ++j) {
                if(rxConn == conn[j]) {
                    if(rxTable[j]) Abort("Duplicate message received");
                    checkMsg(&rxMsg, MsgPcmData, i + j);
                    rxTable[j] = 1;
                    rxRemain--;
                    break;
                }
            }
        }
    }
    stopMs = (clock() * 1000) / CLOCKS_PER_SEC;
    printf("Overhead is %fms/frame\n", (float)(stopMs - startMs) / 10000);

    IpcStop();
}

int main(int argc, char *argv[])
{
    LogInit(stderr);

    if(argc < 3) {
        Abort("You must specify [c|s] [host]");
    }

    IpcInit();

    if(argv[1][0] == 'c') {
        client(argv[2]);
    }
    else if(argv[1][0] == 's') {
        server(argv[2]);
    }
    else {
        Abort("Invalid option");
    }

    IpcFinal();

    return 0;
}

#endif
