/* Copyright (C) 2014 InfiniDB, Inc.

   This program is free software; you can redistribute it and/or
   modify it under the terms of the GNU General Public License
   as published by the Free Software Foundation; version 2 of
   the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
   MA 02110-1301, USA. */

// $Id: impl.cpp 3495 2013-01-21 14:09:51Z rdempsey $

/* This code is based on udpcast-20090830. Most of the source code in that release
   contains no copyright or licensing notices at all. The exception is fec.c, which
   is not used here. The udpcast website, http://udpcast.linux.lu/, implies that
   the source is covered under GPL. */

#include <net/if.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/ioctl.h>
#include <cerrno>
#include <pthread.h>
using namespace std;

#include "bytestream.h"
using namespace messageqcpp;

#include "udp-sender.h"
#include "udpc-protoc.h"
#include "util.h"
#include "mc_fifo.h"

#include "impl.h"

struct  participantsDb
{
    int nrParticipants;

    struct clientDesc
    {
        struct sockaddr_in addr;
        int used;
        int capabilities;
        unsigned int rcvbuf;
    } clientTable[MAX_CLIENTS];
};
struct produconsum
{
    unsigned int size;
    volatile unsigned int produced;
    unsigned int consumed;
    volatile int atEnd;
    pthread_mutex_t mutex;
    volatile int consumerIsWaiting;
    pthread_cond_t cond;
    const char* name;
};

struct stats
{
    int fd;
    struct timeval lastPrinted;
    long statPeriod;
    int printUncompressedPos;
};

struct sender_stats
{
    FILE* log;
    unsigned long long totalBytes;
    unsigned long long retransmissions;
    int clNo;
    unsigned long periodBytes;
    struct timeval periodStart;
    long bwPeriod;
    struct stats s;
};

struct receiver_stats
{
    struct timeval tv_start;
    int bytesOrig;
    long long totalBytes;
    int timerStarted;
    struct stats s;
};

#define SLICEMAGIC 0x41424344

typedef struct slice
{
    int base; /* base address of slice in buffer */
    int sliceNo;
    int bytes; /* bytes in slice */
    int nextBlock; /* index of next buffer to be transmitted */
    enum slice_state
    {
        SLICE_FREE, /* free slice, and in the queue of free slices */
        SLICE_NEW, /* newly allocated. FEC calculation and first
		    * transmission */
        SLICE_XMITTED, /* transmitted */
        SLICE_ACKED, /* acknowledged (if applicable) */
        SLICE_PRE_FREE, /* no longer used, but not returned to queue */
        SLICE_RECEIVING,
        SLICE_DONE,
    };
    volatile enum slice_state state;
    char rxmitMap[MAX_SLICE_SIZE / BITS_PER_CHAR];
    /* blocks to be retransmitted */

    char isXmittedMap[MAX_SLICE_SIZE / BITS_PER_CHAR];
    /* blocks which have already been retransmitted during this round*/

    int rxmitId; /* used to distinguish among several retransmission
		  * requests, so that we can easily discard answers to "old"
		  * requests */

    /* This structure is used to keep track of clients who answered, and
     * to make the reqack message
     */
    struct reqackBm
    {
        struct reqack ra;
        char readySet[MAX_CLIENTS / BITS_PER_CHAR]; /* who is already ok? */
    } sl_reqack;

    char answeredSet[MAX_CLIENTS / BITS_PER_CHAR]; /* who answered at all? */

    int nrReady; /* number of participants who are ready */
    int nrAnswered; /* number of participants who answered; */
    int needRxmit; /* does this need retransmission? */
    int lastGoodBlock; /* last good block of slice (i.e. last block having not
			* needed retransmission */

    int lastReqack; /* last req ack sent (debug) */
#ifdef BB_FEATURE_UDPCAST_FEC
    unsigned char* fec_data;
#endif
    int magic;
    int blocksTransferred; /* blocks transferred during this slice */
    int dataBlocksTransferred; /* data blocks transferred during this slice */
    struct retransmit retransmit;
    int freePos; /* where the next data part will be stored to */
    int bytesKnown; /* is number of bytes known yet? */
    short missing_data_blocks[MAX_FEC_INTERLEAVE];
}* slice_t;

#define QUEUE_SIZE 256

struct returnChannel
{
    pthread_t thread; /* message receiving thread */
    int rcvSock; /* socket on which we receive the messages */
    produconsum_t incoming; /* where to enqueue incoming messages */
    produconsum_t freeSpace; /* free space */
    struct
    {
        int clNo; /* client number */
        union message msg; /* its message */
    } q[QUEUE_SIZE];
    struct net_config* config;
    participantsDb_t participantsDb;
};
#define NR_SLICES 2

typedef struct senderState
{
    struct returnChannel rc;
    struct fifo* fifo;

    struct net_config* config;
    sender_stats_t stats;
    int socket;

    struct slice slices[NR_SLICES];

    produconsum_t free_slices_pc;

    unsigned char* fec_data;
    pthread_t fec_thread;
    produconsum_t fec_data_pc;
}* sender_state_t;

struct clientState
{
    struct fifo* fifo;
    struct client_config* client_config;
    struct net_config* net_config;
    union serverDataMsg Msg;

    struct msghdr data_hdr;

    /* pre-prepared messages */
    struct iovec data_iov[2];

    struct slice* currentSlice;
    int currentSliceNo;
    receiver_stats_t stats;

    produconsum_t free_slices_pc;
    struct slice slices[NR_SLICES];

    /* Completely received slices */
    int receivedPtr;
    int receivedSliceNo;

#ifdef BB_FEATURE_UDPCAST_FEC
    int use_fec; /* do we use forward error correction ? */
#endif
    produconsum_t fec_data_pc;
    struct slice* fec_slices[NR_SLICES];
    pthread_t fec_thread;

    /* A reservoir of free blocks for FEC */
    produconsum_t freeBlocks_pc;
    unsigned char** blockAddresses; /* adresses of blocks in local queue */

    unsigned char** localBlockAddresses;
    /* local blocks: freed FEC blocks after we
     * have received the corresponding data */
    int localPos;

    unsigned char* blockData;
    unsigned char* nextBlock;

    int endReached; /* end of transmission reached:
		       0: transmission in progress
		       2: network transmission _and_ FEC
		          processing finished
		    */

    int netEndReached; /* In case of a FEC transmission; network
			* transmission finished. This is needed to avoid
			* a race condition, where the receiver thread would
			* already prepare to wait for more data, at the same
			* time that the FEC would set endReached. To avoid
			* this, we do a select without timeout before
			* receiving the last few packets, so that if the
			* race condition strikes, we have a way to protect
			* against
			*/

    int selectedFd;

    int promptPrinted;  /* Has "Press any key..." prompt already been printed */

#ifdef BB_FEATURE_UDPCAST_FEC
    fec_code_t fec_code;
#endif
};

#define S_UCAST socks[0]
#define S_BCAST socks[1]
#define S_MCAST_CTRL socks[2]
#define S_MCAST_DATA socks[3]

#define SSEND(x) SEND(client_config->S_UCAST, x, client_config->serverAddr)

/**
 * Receiver will passively listen to sender. Works best if sender runs
 * in async mode
 */
#define FLAG_PASSIVE 0x0010


/**
 * Do not write file synchronously
 */
#define FLAG_NOSYNC 0x0040

/*
 * Don't ask for keyboard input on receiver end.
 */
#define FLAG_NOKBD 0x0080

/**
 * Do write file synchronously
 */
#define FLAG_SYNC 0x0100

namespace
{
int udpc_isFullDuplex(int s, const char* ifname)
{

#ifdef ETHTOOL_GLINK
    struct ifreq ifr;
    struct ethtool_cmd ecmd;

    ecmd.cmd = ETHTOOL_GSET;

    strncpy(ifr.ifr_name, ifname, sizeof(ifr.ifr_name) - 1);
    ifr.ifr_data = (char*) &ecmd;

    if (ioctl(s, SIOCETHTOOL, &ifr) == -1)
    {
        /* Operation not supported */
        return -1;
    }
    else
    {
        return ecmd.duplex;
    }

#else
    return -1;
#endif
}

#define getSinAddr(addr) (((struct sockaddr_in *) addr)->sin_addr)

int udpc_ipIsZero(struct sockaddr_in* ip)
{
    return getSinAddr(ip).s_addr == 0;
}

int hasLink(int s, const char* ifname)
{

#ifdef ETHTOOL_GLINK
    struct ifreq ifr;
    struct ethtool_value edata;

    edata.cmd = ETHTOOL_GLINK;

    strncpy(ifr.ifr_name, ifname, sizeof(ifr.ifr_name) - 1);
    ifr.ifr_data = (char*) &edata;

    if (ioctl(s, SIOCETHTOOL, &ifr) == -1)
    {
        /* Operation not supported */
        return -1;
    }
    else
    {
        return edata.data;
    }

#else
    return -1;
#endif
}

#define INET_ATON(a,i) inet_aton(a,i)

int udpc_doSend(int s, void* message, size_t len, struct sockaddr_in* to)
{
    /*    flprintf("sent: %08x %d\n", *(int*) message, len);*/
#ifdef LOSSTEST
    loseSendPacket();
#endif
    return sendto(s, message, len, 0, (struct sockaddr*) to, sizeof(*to));
}

void udpc_copyToMessage(unsigned char* dst, struct sockaddr_in* src)
{
    memcpy(dst, (char*) & ((struct sockaddr_in*)src)->sin_addr,
           sizeof(struct in_addr));
}

void udpc_sendHello(struct net_config* net_config, int sock,
                    int streaming)
{
//cerr << "sending hello..." << endl;
    struct hello hello;

    /* send hello message */
    if (streaming)
        hello.opCode = htons(CMD_HELLO_STREAMING);
    else
        hello.opCode = htons(CMD_HELLO);

    hello.reserved = 0;
    hello.capabilities = htonl(net_config->capabilities);
    udpc_copyToMessage(hello.mcastAddr, &net_config->dataMcastAddr);
    hello.blockSize = htons(net_config->blockSize);
    //TODO: FIXME
    //rgWaitAll(net_config, sock, net_config->controlMcastAddr.sin_addr.s_addr, sizeof(hello));
    BCAST_CONTROL(sock, hello);
}

char* udpc_getIpString(struct sockaddr_in* addr, char* buffer)
{
    long iaddr = htonl(getSinAddr(addr).s_addr);
    sprintf(buffer, "%ld.%ld.%ld.%ld",
            (iaddr >> 24) & 0xff,
            (iaddr >> 16) & 0xff,
            (iaddr >>  8) & 0xff,
            iaddr & 0xff);
    return buffer;
}

net_if_t* udpc_getNetIf(const char* wanted)
{
#ifndef __MINGW32__
    struct ifreq ibuf[100];
    struct ifreq* ifrp, *ifend, *chosen;
    struct ifconf ifc;
    int s;
#else /* __MINGW32__ */
    int i;

    int etherNo = -1;
    int wantedEtherNo = -2; /* Wanted ethernet interface */

    MIB_IPADDRTABLE* iptab = NULL;
    MIB_IFTABLE* iftab = NULL;

    MIB_IPADDRROW* iprow, *chosen = NULL;
    MIB_IFROW* chosenIf = NULL;
    WORD wVersionRequested; /* Version of Winsock to load */
    WSADATA wsaData; /* Winsock implementation details */
    ULONG a;

    int r;
#endif /* __MINGW32__ */

    int lastGoodness = 0;
    struct in_addr wantedAddress;
    int isAddress = 0;
    int wantedLen = 0;
    net_if_t* net_if;

    if (wanted == NULL)
    {
        wanted = getenv("IFNAME");
    }

    if (wanted && INET_ATON(wanted, &wantedAddress))
        isAddress = 1;
    else
        wantedAddress.s_addr = 0;

    if (wanted)
        wantedLen = strlen(wanted);

    net_if = MALLOC(net_if_t);
    //TODO: FIXME
    //if(net_if == NULL)
    //    udpc_fatal(1, "Out of memory error");

#ifndef __MINGW32__

    s = socket(PF_INET, SOCK_DGRAM, 0);

    if (s < 0)
    {
        perror("make socket");
        exit(1);
    }

    ifc.ifc_len = sizeof(ibuf);
    ifc.ifc_buf = (caddr_t) ibuf;

    if (ioctl(s, SIOCGIFCONF, (char*)&ifc) < 0 ||
            ifc.ifc_len < (signed int)sizeof(struct ifreq))
    {
        perror("udpcast: SIOCGIFCONF: ");
        exit(1);
    }

    ifend = (struct ifreq*)((char*)ibuf + ifc.ifc_len);
    chosen = NULL;

    for (ifrp = ibuf ; ifrp < ifend;
#ifdef IFREQ_SIZE
            ifrp = IFREQ_SIZE(*ifrp) + (char*)ifrp
#else
            ifrp++
#endif
        )
    {
        unsigned long iaddr = getSinAddr(&ifrp->ifr_addr).s_addr;
        int goodness;

        if (ifrp->ifr_addr.sa_family != PF_INET)
            continue;

        if (wanted)
        {
            if (isAddress && iaddr == wantedAddress.s_addr)
            {
                goodness = 8;
            }
            else if (strcmp(wanted, ifrp->ifr_name) == 0)
            {
                /* perfect match on interface name */
                goodness = 12;
            }
            else if (wanted != NULL &&
                     strncmp(wanted, ifrp->ifr_name, wantedLen) == 0)
            {
                /* prefix match on interface name */
                goodness = 7;
            }
            else
            {
                /* no match, try next */
                continue;
            }
        }
        else
        {
            if (iaddr == 0)
            {
                /* disregard interfaces whose address is zero */
                goodness = 1;
            }
            else if (iaddr == htonl(0x7f000001))
            {
                /* disregard localhost type devices */
                goodness = 2;
            }
            else if (strcmp("eth0", ifrp->ifr_name) == 0 ||
                     strcmp("en0",  ifrp->ifr_name) == 0)
            {
                /* prefer first ethernet interface */
                goodness = 6;
            }
            else if (strncmp("eth0:", ifrp->ifr_name, 5) == 0)
            {
                /* second choice: any secondary addresses of first ethernet */
                goodness = 5;
            }
            else if (strncmp("eth", ifrp->ifr_name, 3) == 0 ||
                     strncmp("en",  ifrp->ifr_name, 2) == 0)
            {
                /* and, if not available, any other ethernet device */
                goodness = 4;
            }
            else
            {
                goodness = 3;
            }
        }

        if (hasLink(s, ifrp->ifr_name))
            /* Good or unknown link status privileged over known
             * disconnected */
            goodness += 3;

        /* If all else is the same, prefer interfaces that
         * have broadcast */
        goodness = goodness * 2;

        if (goodness >= lastGoodness)
        {
            /* Privilege broadcast-enabled interfaces */
            if (ioctl(s,  SIOCGIFBRDADDR, ifrp) < 0)
            {
                //TODO: FIXME
                //udpc_fatal(-1, "Error getting broadcast address for %s: %s", ifrp->ifr_name, strerror(errno));
            }

            if (getSinAddr(&ifrp->ifr_ifru.ifru_broadaddr).s_addr)
                goodness++;
        }

        if (goodness > lastGoodness)
        {
            chosen = ifrp;
            lastGoodness = goodness;
            net_if->addr.s_addr = iaddr;
        }
    }


    if (!chosen)
    {
        fprintf(stderr, "No suitable network interface found\n");
        fprintf(stderr, "The following interfaces are available:\n");

        for (ifrp = ibuf ; ifrp < ifend;
#ifdef IFREQ_SIZE
                ifrp = IFREQ_SIZE(*ifrp) + (char*)ifrp
#else
                ifrp++
#endif
            )
        {
            char buffer[16];

            if (ifrp->ifr_addr.sa_family != PF_INET)
                continue;

            fprintf(stderr, "\t%s\t%s\n",
                    ifrp->ifr_name,
                    udpc_getIpString((struct sockaddr_in*)&ifrp->ifr_addr, buffer));
        }

        exit(1);
    }

    net_if->name = strdup(chosen->ifr_name);

#ifdef HAVE_STRUCT_IP_MREQN_IMR_IFINDEX

    /* Index for multicast subscriptions */
    if (ioctl(s,  SIOCGIFINDEX, chosen) < 0)
    {
        //TODO: FIXME
        //udpc_fatal(-1, "Error getting index for %s: %s", net_if->name, strerror(errno));
    }

    net_if->index = chosen->ifr_ifindex;
#endif

    /* Broadcast */
    if (ioctl(s,  SIOCGIFBRDADDR, chosen) < 0)
    {
        //TODO: FIXME
        //udpc_fatal(-1, "Error getting broadcast address for %s: %s", net_if->name, strerror(errno));
    }

    net_if->bcast = getSinAddr(&chosen->ifr_ifru.ifru_broadaddr);

    close(s);

#else /* __MINGW32__ */

    /* WINSOCK initialization */
    wVersionRequested = MAKEWORD(2, 0); /* Request Winsock v2.0 */

    if (WSAStartup(wVersionRequested, &wsaData) != 0) /* Load Winsock DLL */
    {
        fprintf(stderr, "WSAStartup() failed");
        exit(1);
    }

    /* End WINSOCK initialization */


    a = 0;
    r = GetIpAddrTable(iptab, &a, TRUE);
    iptab = malloc(a);
    r = GetIpAddrTable(iptab, &a, TRUE);

    a = 0;
    r = GetIfTable(iftab, &a, TRUE);
    iftab = malloc(a);
    r = GetIfTable(iftab, &a, TRUE);

    if (wanted && !strncmp(wanted, "eth", 3) && wanted[3])
    {
        char* ptr;
        int n = strtoul(wanted + 3, &ptr, 10);

        if (!*ptr)
            wantedEtherNo = n;
    }

    for (i = 0; i < iptab->dwNumEntries; i++)
    {
        int goodness = -1;
        unsigned long iaddr;
        int isEther = 0;
        MIB_IFROW* ifrow;

        iprow = &iptab->table[i];
        iaddr = iprow->dwAddr;

        ifrow = getIfRow(iftab, iprow->dwIndex);

        if (ifrow && ifrow->dwPhysAddrLen == 6 && iprow->dwBCastAddr)
        {
            isEther = 1;
            etherNo++;
        }

        if (wanted)
        {
            if (isAddress && iaddr == wantedAddress.s_addr)
            {
                goodness = 8;
            }
            else if (isEther && wantedEtherNo == etherNo)
            {
                goodness = 9;
            }
            else if (ifrow->dwPhysAddrLen)
            {
                int j;
                const char* ptr = wanted;

                for (j = 0; *ptr && j < ifrow->dwPhysAddrLen; j++)
                {
                    int digit = strtoul(ptr, (char**)&ptr, 16);

                    if (digit != ifrow->bPhysAddr[j])
                        break; /* Digit mismatch */

                    if (*ptr == '-' || *ptr == ':')
                    {
                        ptr++;
                    }
                }

                if (!*ptr && j == ifrow->dwPhysAddrLen)
                {
                    goodness = 9;
                }
            }
        }
        else
        {
            if (iaddr == 0)
            {
                /* disregard interfaces whose address is zero */
                goodness = 1;
            }
            else if (iaddr == htonl(0x7f000001))
            {
                /* disregard localhost type devices */
                goodness = 2;
            }
            else if (isEther)
            {
                /* prefer ethernet */
                goodness = 6;
            }
            else if (ifrow->dwPhysAddrLen)
            {
                /* then prefer interfaces which have a physical address */
                goodness = 4;
            }
            else
            {
                goodness = 3;
            }
        }

        goodness = goodness * 2;

        /* If all else is the same, prefer interfaces that
         * have broadcast */
        if (goodness >= lastGoodness)
        {
            /* Privilege broadcast-enabled interfaces */
            if (iprow->dwBCastAddr)
                goodness++;
        }

        if (goodness > lastGoodness)
        {
            chosen = iprow;
            chosenIf = ifrow;
            lastGoodness = goodness;
        }
    }

    if (!chosen)
    {
        fprintf(stderr, "No suitable network interface found%s%s\n",
                wanted ? " for " : "", wanted ? wanted : "");
        fprintf(stderr, "The following interfaces are available:\n");

        for (i = 0; i < iptab->dwNumEntries; i++)
        {
            char buffer[16];
            struct sockaddr_in addr;
            MIB_IFROW* ifrow;
            char* name = NULL;
            iprow = &iptab->table[i];
            addr.sin_addr.s_addr = iprow->dwAddr;
            ifrow = getIfRow(iftab, iprow->dwIndex);
            name = fmtName(ifrow);
            fprintf(stderr, " %15s  %s\n",
                    udpc_getIpString(&addr, buffer),
                    name ? name : "");

            if (name)
                free(name);
        }

        exit(1);
    }

    net_if->bcast.s_addr = net_if->addr.s_addr = chosen->dwAddr;

    if (chosen->dwBCastAddr)
        net_if->bcast.s_addr |= ~chosen->dwMask;

    if (chosenIf)
    {
        net_if->name = fmtName(chosenIf);
    }
    else
    {
        net_if->name = "*";
    }

    free(iftab);
    free(iptab);
#endif /* __MINGW32__ */

    return net_if;
}

#define IP_MREQN ip_mreqn

int fillMreq(net_if_t* net_if, struct in_addr addr,
             struct IP_MREQN* mreq)
{
#ifdef HAVE_STRUCT_IP_MREQN_IMR_IFINDEX
    mreq->imr_ifindex = net_if->index;
    mreq->imr_address.s_addr = 0;
#else
    mreq->imr_interface = net_if->addr;
#endif
    mreq->imr_multiaddr = addr;

    return 0;
}

int mcastOp(int sock, net_if_t* net_if, struct in_addr addr,
            int code, const char* message)
{
    struct IP_MREQN mreq;
    int r;

    fillMreq(net_if, addr, &mreq);
    r = setsockopt(sock, SOL_IP, code, (char*)&mreq, sizeof(mreq));

    if (r < 0)
    {
        perror(message);
        exit(1);
    }

    return 0;
}

int udpc_setMcastDestination(int sock, net_if_t* net_if, struct sockaddr_in* addr)
{
#ifdef WINDOWS
    int r;
    struct sockaddr_in interface_addr;
    struct in_addr if_addr;
    getMyAddress(net_if, &interface_addr);
    if_addr = getSinAddr(&interface_addr);
    r = setsockopt (sock, IPPROTO_IP, IP_MULTICAST_IF,
                    (char*) &if_addr, sizeof(if_addr));

    if (r < 0)
        fatal(1, "Set multicast send interface");

    return 0;
#else
    /* IP_MULTICAST_IF not correctly supported on Cygwin */
    return mcastOp(sock, net_if, getSinAddr(addr), IP_MULTICAST_IF,
                   "Set multicast send interface");
#endif
}

int initSockAddress(addr_type_t addr_type,
                    net_if_t* net_if,
                    in_addr_t ip,
                    unsigned short port,
                    struct sockaddr_in* addr)
{
    memset ((char*) addr, 0, sizeof(struct sockaddr_in));
    addr->sin_family = AF_INET;
    addr->sin_port = htons(port);

    //TODO: FIXME
    //if(!net_if && addr_type != ADDR_TYPE_MCAST)
    //udpc_fatal(1, "initSockAddr without ifname\n");

    switch (addr_type)
    {
        case ADDR_TYPE_UCAST:
            addr->sin_addr = net_if->addr;
            break;

        case ADDR_TYPE_BCAST:
            addr->sin_addr = net_if->bcast;
            break;

        case ADDR_TYPE_MCAST:
            addr->sin_addr.s_addr = ip;
            break;
    }

    return 0;
}

int mcastListen(int sock, net_if_t* net_if, struct sockaddr_in* addr)
{
    return mcastOp(sock, net_if, getSinAddr(addr), IP_ADD_MEMBERSHIP,
                   "Subscribe to multicast group");
}

int udpc_makeSocket(addr_type_t addr_type,
                    net_if_t* net_if,
                    struct sockaddr_in* tmpl,
                    int port)
{
    int ret, s;
    struct sockaddr_in myaddr;
    in_addr_t ip = 0;

#ifdef WINDOWS
    static int lastSocket = -1;
    /* Very ugly hack, but hey!, this is for Windows */

    if (addr_type == ADDR_TYPE_MCAST)
    {
        mcastListen(lastSocket, net_if, tmpl);
        return -1;
    }
    else if (addr_type != ADDR_TYPE_UCAST)
        return -1;

#endif

    s = socket(PF_INET, SOCK_DGRAM, 0);

    if (s < 0)
    {
        perror("make socket");
        exit(1);
    }

    if (addr_type == ADDR_TYPE_MCAST && tmpl != NULL)
    {
        ip = tmpl->sin_addr.s_addr;
    }

    ret = initSockAddress(addr_type, net_if, ip, port, &myaddr);

    //TODO: FIXME
    //if(ret < 0)
//	udpc_fatal(1, "Could not get socket address fot %d/%s",
//		   addr_type, net_if->name);
    if (addr_type == ADDR_TYPE_BCAST)
    {
//cerr << "for addr_type == ADDR_TYPE_BCAST, myaddr.sin_addr.s_addr = 0x" << hex << myaddr.sin_addr.s_addr << dec << endl;
    }

    if (addr_type == ADDR_TYPE_BCAST && myaddr.sin_addr.s_addr == 0)
    {
        /* Attempting to bind to broadcast address on not-broadcast media ... */
        closesocket(s);
        return -1;
    }

    ret = bind(s, (struct sockaddr*) &myaddr, sizeof(myaddr));
    //TODO: FIXME
//    if (ret < 0) {
//	char buffer[16];
//	udpc_fatal(1, "bind socket to %s:%d (%s)\n",
//		   udpc_getIpString(&myaddr, buffer),
//		   udpc_getPort(&myaddr),
//		   strerror(errno));
//    }

    if (addr_type == ADDR_TYPE_MCAST)
        mcastListen(s, net_if, &myaddr);

#ifdef WINDOWS
    lastSocket = s;
#endif
    return s;
}

int udpc_setSocketToBroadcast(int sock)
{
    /* set the socket to broadcast */
    int p = 1;
    return setsockopt(sock, SOL_SOCKET, SO_BROADCAST, (char*)&p, sizeof(int));
}

int udpc_getBroadCastAddress(net_if_t* net_if, struct sockaddr_in* addr,
                             short port)
{
    int r = initSockAddress(ADDR_TYPE_BCAST, net_if, INADDR_ANY, port, addr);

    if (addr->sin_addr.s_addr == 0)
    {
        /* Quick hack to make it work for loopback */
        struct sockaddr_in ucast;
        initSockAddress(ADDR_TYPE_UCAST, net_if, INADDR_ANY, port, &ucast);

        if ((ntohl(ucast.sin_addr.s_addr) & 0xff000000) == 0x7f000000)
            addr->sin_addr.s_addr = ucast.sin_addr.s_addr;
    }

    return r;
}

int safe_inet_aton(const char* address, struct in_addr* ip)
{
    if (!INET_ATON(address, ip))
    {
        //TODO: FIXME
        //udpc_fatal(-1, "Bad address %s", address);
    }

    return 0;
}

int udpc_getMcastAllAddress(struct sockaddr_in* addr, const char* address,
                            short port)
{
    struct in_addr ip;
    int ret;

    if (address == NULL || address[0] == '\0')
        safe_inet_aton("224.0.0.1", &ip);
    else
    {
        if ((ret = safe_inet_aton(address, &ip)) < 0)
            return ret;
    }

    return initSockAddress(ADDR_TYPE_MCAST, NULL, ip.s_addr, port, addr);
}

void setPort(struct sockaddr_in* addr, unsigned short port)
{
    ((struct sockaddr_in*) addr)->sin_port = htons(port);
}

int isMcastAddress(struct sockaddr_in* addr)
{
    int ip = ntohl(addr->sin_addr.s_addr) >> 24;
    return ip >= 0xe0 && ip < 0xf0;
}

void udpc_clearIp(struct sockaddr_in* addr)
{
    addr->sin_addr.s_addr = 0;
    addr->sin_family = AF_INET;
}

void udpc_setSendBuf(int sock, unsigned int bufsize)
{
    if (setsockopt(sock, SOL_SOCKET, SO_SNDBUF, (char*)&bufsize, sizeof(bufsize)) < 0)
        perror("Set send buffer");
}

int udpc_setTtl(int sock, int ttl)
{
    /* set the socket to broadcast */
    return setsockopt(sock, SOL_IP, IP_MULTICAST_TTL, (char*)&ttl, sizeof(int));
}

int udpc_getMyAddress(net_if_t* net_if, struct sockaddr_in* addr)
{
    return initSockAddress(ADDR_TYPE_UCAST, net_if, INADDR_ANY, 0, addr);
}

void udpc_getDefaultMcastAddress(net_if_t* net_if, struct sockaddr_in* mcast)
{
    udpc_getMyAddress(net_if, mcast);
    mcast->sin_addr.s_addr &= htonl(0x07ffffff);
    mcast->sin_addr.s_addr |= htonl(0xe8000000);
}

void udpc_copyIpFrom(struct sockaddr_in* dst, struct sockaddr_in* src)
{
    dst->sin_addr = src->sin_addr;
    dst->sin_family = src->sin_family;
}

int udpc_getSelectedSock(int* socks, int nr, fd_set* read_set)
{
    int i;
    int maxFd;
    maxFd = -1;

    for (i = 0; i < nr; i++)
    {
        if (socks[i] == -1)
            continue;

        if (FD_ISSET(socks[i], read_set))
            return socks[i];
    }

    return -1;
}

int udpc_ipIsEqual(struct sockaddr_in* left, struct sockaddr_in* right)
{
    return getSinAddr(left).s_addr == getSinAddr(right).s_addr;
}

void udpc_closeSock(int* socks, int nr, int target)
{
    int i;
    int sock = socks[target];

    socks[target] = -1;

    for (i = 0; i < nr; i++)
        if (socks[i] == sock)
            return;

    closesocket(sock);
}
int prepareForSelect(int* socks, int nr, fd_set* read_set)
{
    int i;
    int maxFd;
    FD_ZERO(read_set);
    maxFd = -1;

    for (i = 0; i < nr; i++)
    {
        if (socks[i] == -1)
            continue;

        FD_SET(socks[i], read_set);

        if (socks[i] > maxFd)
            maxFd = socks[i];
    }

    return maxFd;
}

int doReceive(int s, void* message, size_t len,
              struct sockaddr_in* from, int portBase)
{
    socklen_t slen;
    int r;
    unsigned short port;

    slen = sizeof(*from);
#ifdef LOSSTEST
    loseRecvPacket(s);
#endif
    r = recvfrom(s, message, len, 0, (struct sockaddr*)from, &slen);

    if (r < 0)
        return r;

    port = ntohs(from->sin_port);

    if (port != RECEIVER_PORT(portBase) && port != SENDER_PORT(portBase))
    {
        return -1;
    }

    /*    flprintf("recv: %08x %d\n", *(int*) message, r);*/
    return r;
}

participantsDb_t udpc_makeParticipantsDb(void)
{
    return MALLOC(struct participantsDb);
}

int udpc_removeParticipant(struct participantsDb* db, int i)
{
    if (db->clientTable[i].used)
    {
        db->clientTable[i].used = 0;
        db->nrParticipants--;
    }

    return 0;
}

int udpc_lookupParticipant(struct participantsDb* db, struct sockaddr_in* addr)
{
    int i;

    for (i = 0; i < MAX_CLIENTS; i++)
    {
        if (db->clientTable[i].used &&
                udpc_ipIsEqual(&db->clientTable[i].addr, addr))
        {
            return i;
        }
    }

    return -1;
}

int udpc_nrParticipants(participantsDb_t db)
{
    return db->nrParticipants;
}

int udpc_addParticipant(participantsDb_t db,
                        struct sockaddr_in* addr,
                        int capabilities,
                        unsigned int rcvbuf,
                        int pointopoint)
{
//cerr << "adding a participant..." << endl;
    int i;

    if ((i = udpc_lookupParticipant(db, addr)) >= 0)
        return i;

    for (i = 0; i < MAX_CLIENTS; i++)
    {
        if (!db->clientTable[i].used)
        {
            db->clientTable[i].addr = *addr;
            db->clientTable[i].used = 1;
            db->clientTable[i].capabilities = capabilities;
            db->clientTable[i].rcvbuf = rcvbuf;
            db->nrParticipants++;

            return i;
        }
        else if (pointopoint)
            return -1;
    }

    return -1; /* no space left in participant's table */
}

int selectWithoutConsole(int maxFd,
                         fd_set* read_set, struct timeval* tv)
{
    int ret;

    ret = select(maxFd, read_set, NULL, NULL, tv);

    if (ret < 0)
        return -1;

    return ret;
}

#if 0
void sendHello(struct net_config* net_config, int sock,
               int streaming)
{
    struct hello hello;

    /* send hello message */
    if (streaming)
        hello.opCode = htons(CMD_HELLO_STREAMING);
    else
        hello.opCode = htons(CMD_HELLO);

    hello.reserved = 0;
    hello.capabilities = htonl(net_config->capabilities);
    udpc_copyToMessage(hello.mcastAddr, &net_config->dataMcastAddr);
    hello.blockSize = htons(net_config->blockSize);
    //rgWaitAll(net_config, sock, net_config->controlMcastAddr.sin_addr.s_addr, sizeof(hello));
    BCAST_CONTROL(sock, hello);
}
#endif

int checkClientWait(participantsDb_t db,
                    struct net_config* net_config,
                    time_t* firstConnected)
{
    time_t now;

    if (!udpc_nrParticipants(db) || !firstConnected || !*firstConnected)
        return 0; /* do not start: no receivers */

    now = time(0);

    /*
     * If we have a max_client_wait, start the transfer after first client
     * connected + maxSendWait
     */
    if (net_config->max_receivers_wait &&
            (now >= *firstConnected + net_config->max_receivers_wait))
    {
#ifdef USE_SYSLOG
        syslog(LOG_INFO, "max wait[%d] passed: starting",
               net_config->max_receivers_wait );
#endif
        return 1; /* send-wait passed: start */
    }

    /*
     * Otherwise check to see if the minimum of clients
     *  have checked in.
     */
    else if (udpc_nrParticipants(db) >= net_config->min_receivers &&
             /*
              *  If there are enough clients and there's a min wait time, we'll
              *  wait around anyway until then.
              *  Otherwise, we always transfer
              */
             (!net_config->min_receivers_wait ||
              now >= *firstConnected + net_config->min_receivers_wait))
    {
#ifdef USE_SYSLOG
        syslog(LOG_INFO, "min receivers[%d] reached: starting",
               net_config->min_receivers );
#endif
        return 1;
    }
    else
        return 0;
}

int sendConnectionReply(participantsDb_t db,
                        int sock,
                        struct net_config* config,
                        struct sockaddr_in* client,
                        int capabilities,
                        unsigned int rcvbuf)
{
    struct connectReply reply;

    if (rcvbuf == 0)
        rcvbuf = 65536;

    if (capabilities & CAP_BIG_ENDIAN)
    {
        reply.opCode = htons(CMD_CONNECT_REPLY);
        reply.clNr =
            htonl(udpc_addParticipant(db,
                                      client,
                                      capabilities,
                                      rcvbuf,
                                      config->flags & FLAG_POINTOPOINT));
        reply.blockSize = htonl(config->blockSize);
    }
    else
    {
        //TODO: FIXME
        //udpc_fatal(1, "Little endian protocol no longer supported");
    }

    reply.reserved = 0;

    if (config->flags & FLAG_POINTOPOINT)
    {
        udpc_copyIpFrom(&config->dataMcastAddr, client);
    }

    /* new parameters: always big endian */
    reply.capabilities = ntohl(config->capabilities);
    udpc_copyToMessage(reply.mcastAddr, &config->dataMcastAddr);

    /*reply.mcastAddress = mcastAddress;*/
    //rgWaitAll(config, sock, client->sin_addr.s_addr, sizeof(reply));
    if (SEND(sock, reply, *client) < 0)
    {
        perror("reply add new client");
        return -1;
    }

    return 0;
}

int mainDispatcher(int* fd, int nr,
                   participantsDb_t db,
                   struct net_config* net_config,
                   int* tries,
                   time_t* firstConnected)
{
    struct sockaddr_in client;
    union message fromClient;
    fd_set read_set;
    int ret;
    int msgLength;
    int startNow = 0;
    int selected;

    if (firstConnected && !*firstConnected && udpc_nrParticipants(db))
    {
        *firstConnected = time(0);
#ifdef USE_SYSLOG
        syslog(LOG_INFO,
               "first connection: min wait[%d] secs - max wait[%d] - min clients[%d]",
               net_config->min_receivers_wait, net_config->max_receivers_wait,
               net_config->min_receivers );
#endif
    }

    while (!startNow)
    {
        struct timeval tv;
        struct timeval* tvp;
        int nr_desc;

        int maxFd = prepareForSelect(fd, nr, &read_set);

        if (net_config->rexmit_hello_interval)
        {
            tv.tv_usec = (net_config->rexmit_hello_interval % 1000) * 1000;
            tv.tv_sec = net_config->rexmit_hello_interval / 1000;
            tvp = &tv;
        }
        else if (firstConnected && udpc_nrParticipants(db))
        {
            tv.tv_usec = 0;
            tv.tv_sec = 2;
            tvp = &tv;
        }
        else
            tvp = 0;

        nr_desc = selectWithoutConsole(maxFd + 1, &read_set, tvp);

        if (nr_desc < 0)
        {
            perror("select");
            return -1;
        }

        if (nr_desc > 0)
            /* key pressed, or receiver activity */
            break;

        if (net_config->rexmit_hello_interval)
        {
            /* retransmit hello message */
            udpc_sendHello(net_config, fd[0], 0);
            (*tries)++;

            if (net_config->autostart != 0 && *tries > net_config->autostart)
                startNow = 1;
        }

        if (firstConnected)
            startNow =
                startNow || checkClientWait(db, net_config, firstConnected);
    }

    selected = udpc_getSelectedSock(fd, nr, &read_set);

    if (selected == -1)
        return startNow;

    BZERO(fromClient); /* Zero it out in order to cope with short messages
			* from older versions */

    msgLength = RECV(selected, fromClient, client, net_config->portBase);

    if (msgLength < 0)
    {
        perror("problem getting data from client");
        return 0; /* don't panic if we get weird messages */
    }

    if (net_config->flags & FLAG_ASYNC)
        return 0;

    switch (ntohs(fromClient.opCode))
    {
        case CMD_CONNECT_REQ:
            sendConnectionReply(db, fd[0],
                                net_config,
                                &client,
                                CAP_BIG_ENDIAN |
                                ntohl(fromClient.connectReq.capabilities),
                                ntohl(fromClient.connectReq.rcvbuf));
            return startNow;

        case CMD_GO:
            return 1;

        case CMD_DISCONNECT:
            ret = udpc_lookupParticipant(db, &client);

            if (ret >= 0)
                udpc_removeParticipant(db, ret);

            return startNow;

        default:
            break;
    }

    return startNow;
}

static int isPointToPoint(participantsDb_t db, int flags)
{
    if (flags & FLAG_POINTOPOINT)
        return 1;

    if (flags & (FLAG_NOPOINTOPOINT | FLAG_ASYNC))
        return 0;

    return udpc_nrParticipants(db) == 1;
}

int getProducedAmount(produconsum_t pc)
{
    unsigned int produced = pc->produced;
    unsigned int consumed = pc->consumed;

    if (produced < consumed)
        return produced + 2 * pc->size - consumed;
    else
        return produced - consumed;
}

int _consumeAny(produconsum_t pc, unsigned int minAmount,
                struct timespec* ts)
{
    unsigned int amount;
#if DEBUG
    flprintf("%s: Waiting for %d bytes (%d:%d)\n",
             pc->name, minAmount, pc->consumed, pc->produced);
#endif
    pc->consumerIsWaiting = 1;
    amount = getProducedAmount(pc);

    if (amount >= minAmount || pc->atEnd)
    {
        pc->consumerIsWaiting = 0;
#if DEBUG
        flprintf("%s: got %d bytes\n", pc->name, amount);
#endif
        return amount;
    }

    pthread_mutex_lock(&pc->mutex);

    while ((amount = getProducedAmount(pc)) < minAmount && !pc->atEnd)
    {
#if DEBUG
        flprintf("%s: ..Waiting for %d bytes (%d:%d)\n",
                 pc->name, minAmount, pc->consumed, pc->produced);
#endif

        if (ts == 0)
            pthread_cond_wait(&pc->cond, &pc->mutex);
        else
        {
            int r;
#if DEBUG
            flprintf("Before timed wait\n");
#endif
            r = pthread_cond_timedwait(&pc->cond, &pc->mutex, ts);
#if DEBUG
            flprintf("After timed wait %d\n", r);
#endif

            if (r == ETIMEDOUT)
            {
                amount = getProducedAmount(pc);
                break;
            }
        }
    }

    pthread_mutex_unlock(&pc->mutex);
#if DEBUG
    flprintf("%s: Got them %d (for %d) %d\n", pc->name,
             amount, minAmount, pc->atEnd);
#endif
    pc->consumerIsWaiting = 0;
    return amount;
}

produconsum_t pc_makeProduconsum(int size, const char* name)
{
    produconsum_t pc = MALLOC(struct produconsum);
    pc->size = size;
    pc->produced = 0;
    pc->consumed = 0;
    pc->atEnd = 0;
    pthread_mutex_init(&pc->mutex, NULL);
    pc->consumerIsWaiting = 0;
    pthread_cond_init(&pc->cond, NULL);
    pc->name = name;
    return pc;
}

int pc_consumeAnyWithTimeout(produconsum_t pc, struct timespec* ts)
{
    return _consumeAny(pc, 1, ts);
}

unsigned int pc_getProducerPosition(produconsum_t pc)
{
    return pc->produced % pc->size;
}

unsigned int pc_getWaiting(produconsum_t pc)
{
    return getProducedAmount(pc);
}

int pc_consumeAny(produconsum_t pc)
{
    return _consumeAny(pc, 1, 0);
}

int pc_consume(produconsum_t pc, int amount)
{
    return _consumeAny(pc, amount, 0);
}

void wakeConsumer(produconsum_t pc)
{
    if (pc->consumerIsWaiting)
    {
        pthread_mutex_lock(&pc->mutex);
        pthread_cond_signal(&pc->cond);
        pthread_mutex_unlock(&pc->mutex);
    }
}

void pc_produceEnd(produconsum_t pc)
{
    pc->atEnd = 1;
    wakeConsumer(pc);
}

int pc_consumed(produconsum_t pc, int amount)
{
    unsigned int consumed = pc->consumed;

    if (consumed >= 2 * pc->size - amount)
    {
        consumed += amount - 2 * pc->size;
    }
    else
    {
        consumed += amount;
    }

    pc->consumed = consumed;
    return amount;
}

unsigned int pc_getConsumerPosition(produconsum_t pc)
{
    return pc->consumed % pc->size;
}

void pc_produce(produconsum_t pc, unsigned int amount)
{
    unsigned int produced = pc->produced;
    unsigned int consumed = pc->consumed;

    /* sanity checks:
     * 1. should not produce more than size
     * 2. do not pass consumed+size
     */
    if (amount > pc->size)
    {
        //TODO: FIXME
        //udpc_fatal(1, "Buffer overflow in produce %s: %d > %d \n", pc->name, amount, pc->size);
    }

    produced += amount;

    if (produced >= 2 * pc->size)
        produced -= 2 * pc->size;

    if (produced > consumed + pc->size ||
            (produced < consumed && produced > consumed - pc->size))
    {
        //TODO: FIXME
        //udpc_fatal(1, "Buffer overflow in produce %s: %d > %d [%d] \n", pc->name, produced, consumed, pc->size);
    }

    pc->produced = produced;
    wakeConsumer(pc);
}

void udpc_initFifo(struct fifo* fifo, int blockSize)
{
    fifo->dataBufSize = blockSize * 4096;
    fifo->dataBuffer = (unsigned char*)malloc(fifo->dataBufSize + 4096);
    fifo->dataBuffer += 4096 - (((unsigned long)fifo->dataBuffer) % 4096);

    /* Free memory queue is initially full */
    fifo->freeMemQueue = pc_makeProduconsum(fifo->dataBufSize, "free mem");
    pc_produce(fifo->freeMemQueue, fifo->dataBufSize);

    fifo->data = pc_makeProduconsum(fifo->dataBufSize, "receive");
}

THREAD_RETURN returnChannelMain(void* args)
{
    struct returnChannel* returnChannel = (struct returnChannel*) args;

    while (1)
    {
        struct sockaddr_in from;
        int clNo;
        int pos = pc_getConsumerPosition(returnChannel->freeSpace);
        pc_consumeAny(returnChannel->freeSpace);

        RECV(returnChannel->rcvSock,
             returnChannel->q[pos].msg, from,
             returnChannel->config->portBase);
        clNo = udpc_lookupParticipant(returnChannel->participantsDb, &from);

        if (clNo < 0)
        {
            /* packet from unknown provenance */
            continue;
        }

        returnChannel->q[pos].clNo = clNo;
        pc_consumed(returnChannel->freeSpace, 1);
        pc_produce(returnChannel->incoming, 1);
    }

    return 0;
}

void initReturnChannel(struct returnChannel* returnChannel,
                       struct net_config* config,
                       int sock)
{
    returnChannel->config = config;
    returnChannel->rcvSock = sock;
    returnChannel->freeSpace = pc_makeProduconsum(QUEUE_SIZE, "msg:free-queue");
    pc_produce(returnChannel->freeSpace, QUEUE_SIZE);
    returnChannel->incoming = pc_makeProduconsum(QUEUE_SIZE, "msg:incoming");

    pthread_create(&returnChannel->thread, NULL,
                   returnChannelMain, returnChannel);

}

void senderStatsAddBytes(sender_stats_t ss, long bytes)
{
    if (ss != NULL)
    {
        ss->totalBytes += bytes;

        if (ss->bwPeriod)
        {
            double tdiff, bw;
            struct timeval tv;
            gettimeofday(&tv, 0);
            ss->periodBytes += bytes;

            if (tv.tv_sec - ss->periodStart.tv_sec < ss->bwPeriod - 1)
                return;

            tdiff = (tv.tv_sec - ss->periodStart.tv_sec) * 1000000.0 +
                    tv.tv_usec - ss->periodStart.tv_usec;

            if (tdiff < ss->bwPeriod * 1000000.0)
                return;

            bw = ss->periodBytes * 8.0 / tdiff;
            ss->periodBytes = 0;
            ss->periodStart = tv;
        }
    }
}
int ackSlice(struct slice* slice, struct net_config* net_config,
             struct fifo* fifo, sender_stats_t stats)
{
    if (slice->state == slice::SLICE_ACKED)
        /* already acked */
        return 0;

    if (!(net_config->flags & FLAG_SN))
    {
        if (net_config->discovery == net_config::DSC_DOUBLING)
        {
            net_config->sliceSize += net_config->sliceSize / 4;

            if (net_config->sliceSize >= net_config->max_slice_size)
            {
                net_config->sliceSize = net_config->max_slice_size;
                net_config->discovery = net_config::DSC_REDUCING;
            }
        }
    }

    slice->state = slice::SLICE_ACKED;
    pc_produce(fifo->freeMemQueue, slice->bytes);

    /* Statistics */
    senderStatsAddBytes(stats, slice->bytes);

    /* End Statistics */

    return 0;
}

int isSliceAcked(struct slice* slice)
{
    if (slice->state == slice::SLICE_ACKED)
    {
        return 1;
    }
    else
    {
        return 0;
    }
}

int freeSlice(sender_state_t sendst, struct slice* slice)
{
    int i;
    i = slice - sendst->slices;
    slice->state = slice::SLICE_PRE_FREE;

    while (1)
    {
        int pos = pc_getProducerPosition(sendst->free_slices_pc);

        if (sendst->slices[pos].state == slice::SLICE_PRE_FREE)
            sendst->slices[pos].state = slice::SLICE_FREE;
        else
            break;

        pc_produce(sendst->free_slices_pc, 1);
    }

    return 0;
}

int isSliceXmitted(struct slice* slice)
{
    if (slice->state == slice::SLICE_XMITTED)
    {
        return 1;
    }
    else
    {
        return 0;
    }
}

int getSliceBlocks(struct slice* slice, struct net_config* net_config)
{
    return (slice->bytes + net_config->blockSize - 1) / net_config->blockSize;
}

int sendReqack(struct slice* slice, struct net_config* net_config,
               struct fifo* fifo, sender_stats_t stats,
               int sock)
{
    /* in async mode, just confirm slice... */
    if ((net_config->flags & FLAG_ASYNC) && slice->bytes != 0)
    {
        ackSlice(slice, net_config, fifo, stats);
        return 0;
    }

    if ((net_config->flags & FLAG_ASYNC)
#ifdef BB_FEATURE_UDPCAST_FEC
            &&
            (net_config->flags & FLAG_FEC)
#endif
       )
    {
        return 0;
    }

    if (!(net_config->flags & FLAG_SN) && slice->rxmitId != 0)
    {
        int nrBlocks;
        nrBlocks = getSliceBlocks(slice, net_config);

        if (slice->lastGoodBlock != 0 && slice->lastGoodBlock < nrBlocks)
        {
            net_config->discovery = net_config::DSC_REDUCING;

            if (slice->lastGoodBlock < net_config->sliceSize / 2)
            {
                net_config->sliceSize = net_config->sliceSize / 2;
            }
            else
            {
                net_config->sliceSize = slice->lastGoodBlock;
            }

            if (net_config->sliceSize < 32)
            {
                /* a minimum of 32 */
                net_config->sliceSize = 32;
            }
        }
    }

    slice->lastGoodBlock = 0;
    slice->sl_reqack.ra.opCode = htons(CMD_REQACK);
    slice->sl_reqack.ra.sliceNo = htonl(slice->sliceNo);
    slice->sl_reqack.ra.bytes = htonl(slice->bytes);

    slice->sl_reqack.ra.reserved = 0;
    memcpy((void*)&slice->answeredSet, (void*)&slice->sl_reqack.readySet,
           sizeof(slice->answeredSet));
    slice->nrAnswered = slice->nrReady;

    /* not everybody is ready yet */
    slice->needRxmit = 0;
    memset(slice->rxmitMap, 0, sizeof(slice->rxmitMap));
    memset(slice->isXmittedMap, 0, sizeof(slice->isXmittedMap));
    slice->sl_reqack.ra.rxmit = htonl(slice->rxmitId);

    //rgWaitAll(net_config, sock, net_config->dataMcastAddr.sin_addr.s_addr, sizeof(slice->sl_reqack));
    BCAST_DATA(sock, slice->sl_reqack);
    return 0;
}

struct slice* findSlice(struct slice* slice1,
                        struct slice* slice2,
                        int sliceNo)
{
    if (slice1 != NULL && slice1->sliceNo == sliceNo)
        return slice1;

    if (slice2 != NULL && slice2->sliceNo == sliceNo)
        return slice2;

    return NULL;
}

void markParticipantAnswered(slice_t slice, int clNo)
{
    if (BIT_ISSET(clNo, slice->answeredSet))
        /* client already has answered */
        return;

    slice->nrAnswered++;
    SET_BIT(clNo, slice->answeredSet);
}

int udpc_isParticipantValid(struct participantsDb* db, int i)
{
    return db->clientTable[i].used;
}

void senderSetAnswered(sender_stats_t ss, int clNo)
{
    if (ss != NULL)
        ss->clNo = clNo;
}

int handleOk(sender_state_t sendst,
             struct slice* slice,
             int clNo)
{
    if (slice == NULL)
        return 0;

    if (!udpc_isParticipantValid(sendst->rc.participantsDb, clNo))
    {
        //udpc_flprintf("Invalid participant %d\n", clNo);
        return 0;
    }

    if (BIT_ISSET(clNo, slice->sl_reqack.readySet))
    {
        /* client is already marked ready */
    }
    else
    {
        SET_BIT(clNo, slice->sl_reqack.readySet);
        slice->nrReady++;
        senderSetAnswered(sendst->stats, clNo);
        markParticipantAnswered(slice, clNo);
    }

    return 0;
}

int handleDisconnect1(struct slice* slice, int clNo)
{
    if (slice != NULL)
    {
        if (BIT_ISSET(clNo, slice->sl_reqack.readySet))
        {
            /* avoid counting client both as left and ready */
            CLR_BIT(clNo, slice->sl_reqack.readySet);
            slice->nrReady--;
        }

        if (BIT_ISSET(clNo, slice->answeredSet))
        {
            slice->nrAnswered--;
            CLR_BIT(clNo, slice->answeredSet);
        }
    }

    return 0;
}

int handleDisconnect(participantsDb_t db,
                     struct slice* slice1,
                     struct slice* slice2,
                     int clNo)
{
    handleDisconnect1(slice1, clNo);
    handleDisconnect1(slice2, clNo);
    udpc_removeParticipant(db, clNo);
    return 0;
}

int handleRetransmit(sender_state_t sendst,
                     struct slice* slice,
                     int clNo, unsigned char* map, int rxmit)
{
    unsigned int i;

    if (!udpc_isParticipantValid(sendst->rc.participantsDb, clNo))
    {
        //udpc_flprintf("Invalid participant %d\n", clNo);
        return 0;
    }

    if (slice == NULL)
        return 0;

    if (rxmit < slice->rxmitId)
    {
        /* late answer to previous Req Ack */
        return 0;
    }

    for (i = 0; i < sizeof(slice->rxmitMap) / sizeof(char); i++)
    {
        slice->rxmitMap[i] |= ~map[i];
    }

    slice->needRxmit = 1;
    markParticipantAnswered(slice, clNo);
    return 0;
}
int handleNextMessage(sender_state_t sendst,
                      struct slice* xmitSlice,
                      struct slice* rexmitSlice)
{
    int pos = pc_getConsumerPosition(sendst->rc.incoming);
    union message* msg = &sendst->rc.q[pos].msg;
    int clNo = sendst->rc.q[pos].clNo;

    pc_consumeAny(sendst->rc.incoming);

    switch (ntohs(msg->opCode))
    {
        case CMD_OK:
            handleOk(sendst,
                     findSlice(xmitSlice, rexmitSlice, ntohl(msg->ok.sliceNo)), clNo);
            break;

        case CMD_DISCONNECT:
            handleDisconnect(sendst->rc.participantsDb,
                             xmitSlice, rexmitSlice, clNo);
            break;

        case CMD_RETRANSMIT:
            handleRetransmit(sendst,
                             findSlice(xmitSlice, rexmitSlice,
                                       ntohl(msg->retransmit.sliceNo)),
                             clNo,
                             msg->retransmit.map,
                             msg->retransmit.rxmit);
            break;

        default:
            //TODO: FIXME
            //udpc_flprintf("Bad command %04x\n", (unsigned short) msg->opCode);
            break;
    }

    pc_consumed(sendst->rc.incoming, 1);
    pc_produce(sendst->rc.freeSpace, 1);
    return 0;
}

int sendRawData(int sock,
                struct net_config* config,
                char* header, int headerSize,
                unsigned char* data, int dataSize)
{
    struct iovec iov[2];
    struct msghdr hdr;
    int packetSize;
    int ret;

    iov[0].iov_base = header;
    iov[0].iov_len = headerSize;

    iov[1].iov_base = data;
    iov[1].iov_len = dataSize;

    hdr.msg_name = &config->dataMcastAddr;
    hdr.msg_namelen = sizeof(struct sockaddr_in);
    hdr.msg_iov = iov;
    hdr.msg_iovlen  = 2;
    initMsgHdr(&hdr);

    packetSize = dataSize + headerSize;
    //rgWaitAll(config, sock, config->dataMcastAddr.sin_addr.s_addr, packetSize);
    ret = sendmsg(sock, &hdr, 0);

    if (ret < 0)
    {
        //TODO: FIXME
    }

    return 0;
}
int transmitDataBlock(sender_state_t sendst, struct slice* slice, int i)
{
    struct fifo* fifo = sendst->fifo;
    struct net_config* config = sendst->config;
    struct dataBlock msg;

    idbassert(i < MAX_SLICE_SIZE);

    msg.opCode  = htons(CMD_DATA);
    msg.sliceNo = htonl(slice->sliceNo);
    msg.blockNo = htons(i);

    msg.reserved = 0;
    msg.reserved2 = 0;
    msg.bytes = htonl(slice->bytes);

    sendRawData(sendst->socket, config,
                (char*) &msg, sizeof(msg),
                fifo->dataBuffer +
                (slice->base + i * config->blockSize) % fifo->dataBufSize,
                config->blockSize);
    return 0;
}

void senderStatsAddRetransmissions(sender_stats_t ss, int retransmissions)
{
    if (ss != NULL)
    {
        ss->retransmissions += retransmissions;
    }
}

int sendSlice(sender_state_t sendst, struct slice* slice,
              int retransmitting)
{
    struct net_config* config = sendst->config;

    int nrBlocks, i, rehello;
#ifdef BB_FEATURE_UDPCAST_FEC
    int fecBlocks;
#endif
    int retransmissions = 0;

    if (retransmitting)
    {
        slice->nextBlock = 0;

        if (slice->state != slice::SLICE_XMITTED)
            return 0;
    }
    else
    {
        if (slice->state != slice::SLICE_NEW)
            return 0;
    }

    nrBlocks = getSliceBlocks(slice, config);
#ifdef BB_FEATURE_UDPCAST_FEC

    if ((config->flags & FLAG_FEC) && !retransmitting)
    {
        fecBlocks = config->fec_redundancy * config->fec_stripes;
    }
    else
    {
        fecBlocks = 0;
    }

#endif

    if ((sendst->config->flags & FLAG_STREAMING))
    {
        rehello = nrBlocks - sendst->config->rehelloOffset;

        if (rehello < 0)
            rehello = 0;
    }
    else
    {
        rehello = -1;
    }

    /* transmit the data */
    for (i = slice->nextBlock; i < nrBlocks
#ifdef BB_FEATURE_UDPCAST_FEC
            + fecBlocks
#endif
            ; i++)
    {
        if (retransmitting)
        {
            if (!BIT_ISSET(i, slice->rxmitMap) ||
                    BIT_ISSET(i, slice->isXmittedMap))
            {
                /* if slice is not in retransmit list, or has _already_
                 * been retransmitted, skip it */
                if (i > slice->lastGoodBlock)
                    slice->lastGoodBlock = i;

                continue;
            }

            SET_BIT(i, slice->isXmittedMap);
            retransmissions++;
        }

        if (i == rehello)
        {
            udpc_sendHello(sendst->config, sendst->socket, 1);
        }

        if (i < nrBlocks)
            transmitDataBlock(sendst, slice, i);

#ifdef BB_FEATURE_UDPCAST_FEC
        else
            transmitFecBlock(sendst, slice, i - nrBlocks);

#endif

        if (!retransmitting && pc_getWaiting(sendst->rc.incoming))
        {
            i++;
            break;
        }
    }

    if (retransmissions)
        senderStatsAddRetransmissions(sendst->stats, retransmissions);

    slice->nextBlock = i;

    if (i == nrBlocks
#ifdef BB_FEATURE_UDPCAST_FEC
            + fecBlocks
#endif
       )
    {
        slice->needRxmit = 0;

        if (!retransmitting)
            slice->state = slice::SLICE_XMITTED;

        return 2;
    }

    return 1;
}

int doRetransmissions(sender_state_t sendst,
                      struct slice* slice)
{
    if (slice->state == slice::SLICE_ACKED)
        return 0; /* nothing to do */

    /* FIXME: reduce slice size if needed */
    if (slice->needRxmit)
    {
        /* do some retransmissions */
        sendSlice(sendst, slice, 1);
    }

    return 0;
}

struct slice* makeSlice(sender_state_t sendst, int sliceNo)
{
    struct net_config* config = sendst->config;
    struct fifo* fifo = sendst->fifo;
    int i;
    struct slice* slice = NULL;

    pc_consume(sendst->free_slices_pc, 1);
    i = pc_getConsumerPosition(sendst->free_slices_pc);
    slice = &sendst->slices[i];
    idbassert(slice->state == slice::SLICE_FREE);
    BZERO(*slice);
    pc_consumed(sendst->free_slices_pc, 1);

    slice->base = pc_getConsumerPosition(sendst->fifo->data);
    slice->sliceNo = sliceNo;
    slice->bytes = pc_consume(fifo->data, 10 * config->blockSize);

    /* fixme: use current slice size here */
    if (slice->bytes > config->blockSize * config->sliceSize)
        slice->bytes = config->blockSize * config->sliceSize;

    pc_consumed(fifo->data, slice->bytes);
    slice->nextBlock = 0;
    slice->state = slice::SLICE_NEW;
    BZERO(slice->sl_reqack.readySet);
    slice->nrReady = 0;
#ifdef BB_FEATURE_UDPCAST_FEC
    slice->fec_data = sendst->fec_data + (i * config->fec_stripes *
                                          config->fec_redundancy *
                                          config->blockSize);
#endif
    return slice;
}

void cancelReturnChannel(struct returnChannel* returnChannel)
{
    /* No need to worry about the pthread_cond_wait in produconsum, because
     * at the point where we enter here (to cancel the thread), we are sure
     * that nobody else uses that produconsum any more
     */
    pthread_cancel(returnChannel->thread);
    pthread_join(returnChannel->thread, NULL);
}

THREAD_RETURN netSenderMain(void*	args0)
{
    sender_state_t sendst = (sender_state_t) args0;
    struct net_config* config = sendst->config;
    struct timeval tv;
    struct timespec ts;
    int atEnd = 0;
    int nrWaited = 0;
    unsigned long waitAverage = 10000; /* Exponential average of last wait times */

    struct slice* xmitSlice = NULL; /* slice being transmitted a first time */
    struct slice* rexmitSlice = NULL; /* slice being re-transmitted */
    int sliceNo = 0;

    /* transmit the data */
    if (config->default_slice_size == 0)
    {
#ifdef BB_FEATURE_UDPCAST_FEC

        if (config->flags & FLAG_FEC)
        {
            config->sliceSize =
                config->fec_stripesize * config->fec_stripes;
        }
        else
#endif
            if (config->flags & FLAG_ASYNC)
                config->sliceSize = 1024;
            else if (sendst->config->flags & FLAG_SN)
            {
                sendst->config->sliceSize = 112;
            }
            else
                sendst->config->sliceSize = 130;

        sendst->config->discovery = net_config::DSC_DOUBLING;
    }
    else
    {
        config->sliceSize = config->default_slice_size;
#ifdef BB_FEATURE_UDPCAST_FEC

        if ((config->flags & FLAG_FEC) &&
                (config->sliceSize > 128 * config->fec_stripes))
            config->sliceSize = 128 * config->fec_stripes;

#endif
    }

#ifdef BB_FEATURE_UDPCAST_FEC

    if ( (sendst->config->flags & FLAG_FEC) &&
            config->max_slice_size > config->fec_stripes * 128)
        config->max_slice_size = config->fec_stripes * 128;

#endif

    if (config->sliceSize > config->max_slice_size)
        config->sliceSize = config->max_slice_size;

    idbassert(config->sliceSize <= MAX_SLICE_SIZE);

    do
    {
        /* first, cleanup rexmit Slice if needed */

        if (rexmitSlice != NULL)
        {
            if (rexmitSlice->nrReady ==
                    udpc_nrParticipants(sendst->rc.participantsDb))
            {
#if DEBUG
                flprintf("slice is ready\n");
#endif
                ackSlice(rexmitSlice, sendst->config, sendst->fifo,
                         sendst->stats);
            }

            if (isSliceAcked(rexmitSlice))
            {
                freeSlice(sendst, rexmitSlice);
                rexmitSlice = NULL;
            }
        }

        /* then shift xmit slice to rexmit slot, if possible */
        if (rexmitSlice == NULL &&  xmitSlice != NULL &&
                isSliceXmitted(xmitSlice))
        {
            rexmitSlice = xmitSlice;
            xmitSlice = NULL;
            sendReqack(rexmitSlice, sendst->config, sendst->fifo, sendst->stats,
                       sendst->socket);
        }

        /* handle any messages */
        if (pc_getWaiting(sendst->rc.incoming))
        {
#if DEBUG
            flprintf("Before message %d\n",
                     pc_getWaiting(sendst->rc.incoming));
#endif
            handleNextMessage(sendst, xmitSlice, rexmitSlice);

            /* restart at beginning of loop: we may have acked the rxmit
             * slice, makeing it possible to shift the pipe */
            continue;
        }

        /* do any needed retransmissions */
        if (rexmitSlice != NULL && rexmitSlice->needRxmit)
        {
            doRetransmissions(sendst, rexmitSlice);
            /* restart at beginning: new messages may have arrived during
             * retransmission  */
            continue;
        }

        /* if all participants answered, send req ack */
        if (rexmitSlice != NULL &&
                rexmitSlice->nrAnswered ==
                udpc_nrParticipants(sendst->rc.participantsDb))
        {
            rexmitSlice->rxmitId++;
            sendReqack(rexmitSlice, sendst->config, sendst->fifo, sendst->stats,
                       sendst->socket);
        }

        if (xmitSlice == NULL && !atEnd)
        {
#if DEBUG
            flprintf("SN=%d\n", sendst->config->flags & FLAG_SN);
#endif

            if ((sendst->config->flags & FLAG_SN) ||
                    rexmitSlice == NULL)
            {
#ifdef BB_FEATURE_UDPCAST_FEC

                if (sendst->config->flags & FLAG_FEC)
                {
                    int i;
                    pc_consume(sendst->fec_data_pc, 1);
                    i = pc_getConsumerPosition(sendst->fec_data_pc);
                    xmitSlice = &sendst->slices[i];
                    pc_consumed(sendst->fec_data_pc, 1);
                }
                else
#endif
                {
                    xmitSlice = makeSlice(sendst, sliceNo++);
                }

                if (xmitSlice->bytes == 0)
                    atEnd = 1;
            }
        }

        if (xmitSlice != NULL && xmitSlice->state == slice::SLICE_NEW)
        {
            sendSlice(sendst, xmitSlice, 0);
#if DEBUG
            flprintf("%d Interrupted at %d/%d\n", xmitSlice->sliceNo,
                     xmitSlice->nextBlock,
                     getSliceBlocks(xmitSlice, sendst->config));
#endif
            continue;
        }

        if (atEnd && rexmitSlice == NULL && xmitSlice == NULL)
            break;

        if (sendst->config->flags & FLAG_ASYNC)
            break;

#if DEBUG
        flprintf("Waiting for timeout...\n");
#endif
        gettimeofday(&tv, 0);
        ts.tv_sec = tv.tv_sec;
        ts.tv_nsec = (long int)((tv.tv_usec + 1.1 * waitAverage) * 1000);

#ifdef WINDOWS
        /* Windows has a granularity of 1 millisecond in its timer. Take this
         * into account here */
#define GRANULARITY 1000000
        ts.tv_nsec += 3 * GRANULARITY / 2;
        ts.tv_nsec -= ts.tv_nsec % GRANULARITY;
#endif

#define BILLION 1000000000

        while (ts.tv_nsec >= BILLION)
        {
            ts.tv_nsec -= BILLION;
            ts.tv_sec++;
        }

        if (rexmitSlice->rxmitId > 10)
            /* after tenth retransmission, wait minimum one second */
            ts.tv_sec++;

        if (pc_consumeAnyWithTimeout(sendst->rc.incoming, &ts) != 0)
        {
#if DEBUG
            flprintf("Have data\n");
#endif
            {
                struct timeval tv2;
                unsigned long timeout;
                gettimeofday(&tv2, 0);
                timeout =
                    (tv2.tv_sec - tv.tv_sec) * 1000000 +
                    tv2.tv_usec - tv.tv_usec;

                if (nrWaited)
                    timeout += waitAverage;

                waitAverage += 9; /* compensate against rounding errors */
                waitAverage = (long unsigned int)((0.9 * waitAverage + 0.1 * timeout));
            }
            nrWaited = 0;
            continue;
        }

        if (rexmitSlice == NULL)
        {
            //TODO: FIXME
            //udpc_flprintf("Weird. Timeout and no rxmit slice");
            break;
        }

        if (nrWaited > 5)
        {
#ifndef WINDOWS
            /* on Cygwin, we would get too many of those messages... */
            nrWaited = 0;
#endif
        }

        nrWaited++;

        if (rexmitSlice->rxmitId > config->retriesUntilDrop)
        {
            int i;

            for (i = 0; i < MAX_CLIENTS; i++)
            {
                if (udpc_isParticipantValid(sendst->rc.participantsDb, i) &&
                        !BIT_ISSET(i, rexmitSlice->sl_reqack.readySet))
                {
                    udpc_removeParticipant(sendst->rc.participantsDb, i);

                    if (udpc_nrParticipants(sendst->rc.participantsDb) == 0)
                        goto exit_main_loop;
                }
            }

            continue;
        }

        rexmitSlice->rxmitId++;
        sendReqack(rexmitSlice, sendst->config, sendst->fifo, sendst->stats,
                   sendst->socket);
    }
    while (udpc_nrParticipants(sendst->rc.participantsDb) ||
            (config->flags & FLAG_ASYNC));

exit_main_loop:
    cancelReturnChannel(&sendst->rc);
    pc_produceEnd(sendst->fifo->freeMemQueue);
    return 0;
}

int spawnNetSender(struct fifo* fifo,
                   int sock,
                   struct net_config* config,
                   participantsDb_t db)
{
    int i;

    sender_state_t sendst = MALLOC(struct senderState);
    sendst->fifo = fifo;
    sendst->socket = sock;
    sendst->config = config;
    //sendst->stats = stats;
#ifdef BB_FEATURE_UDPCAST_FEC

    if (sendst->config->flags & FLAG_FEC)
        sendst->fec_data =  xmalloc(NR_SLICES *
                                    config->fec_stripes *
                                    config->fec_redundancy *
                                    config->blockSize);

#endif
    sendst->rc.participantsDb = db;
    initReturnChannel(&sendst->rc, sendst->config, sendst->socket);

    sendst->free_slices_pc = pc_makeProduconsum(NR_SLICES, "free slices");
    pc_produce(sendst->free_slices_pc, NR_SLICES);

    for (i = 0; i < NR_SLICES; i++)
        sendst->slices[i].state = slice::SLICE_FREE;

#ifdef BB_FEATURE_UDPCAST_FEC

    if (sendst->config->flags & FLAG_FEC)
    {
        /* Free memory queue is initially full */
        fec_init();
        sendst->fec_data_pc = pc_makeProduconsum(NR_SLICES, "fec data");

        pthread_create(&sendst->fec_thread, NULL, fecMain, sendst);
    }

#endif

    pthread_create(&fifo->thread, NULL, netSenderMain, sendst);
    return 0;
}

struct sockaddr_in* udpc_getParticipantIp(participantsDb_t db, int i)
{
    return &db->clientTable[i].addr;
}

int udpc_getParticipantCapabilities(participantsDb_t db, int i)
{
    return db->clientTable[i].capabilities;
}

unsigned int udpc_getParticipantRcvBuf(participantsDb_t db, int i)
{
    return db->clientTable[i].rcvbuf;
}

int pc_consumeContiguousMinAmount(produconsum_t pc, int amount)
{
    int n = _consumeAny(pc, amount, 0);
    int l = pc->size - (pc->consumed % pc->size);

    if (n > l)
        n = l;

    return n;

}

#define BLOCKSIZE 4096

void localReader(struct fifo* fifo, const uint8_t* buf, uint32_t len)
{
//cerr << "starting to send " << len << " bytes" << endl;
    uint32_t offset = 0;

    while (1)
    {
        int pos = pc_getConsumerPosition(fifo->freeMemQueue);
        int bytes =
            pc_consumeContiguousMinAmount(fifo->freeMemQueue, BLOCKSIZE);

        if (bytes > (pos + bytes) % BLOCKSIZE)
            bytes -= (pos + bytes) % BLOCKSIZE;

        if (offset + bytes > len) bytes = len - offset;

//cerr << "sending " << bytes << " bytes from bs..." << endl;
        //bytes = read(in, fifo->dataBuffer + pos, bytes);
        memcpy(fifo->dataBuffer + pos, buf + offset, bytes);
        offset += bytes;

        if (bytes == 0)
        {
            /* the end */
            pc_produceEnd(fifo->data);
            break;
        }
        else
        {
            pc_consumed(fifo->freeMemQueue, bytes);
            pc_produce(fifo->data, bytes);
        }
    }

//cerr << "done sending" << endl;
}

unsigned int udpc_getRcvBuf(int sock)
{
    unsigned int bufsize;
    socklen_t len = sizeof(int);

    if (getsockopt(sock, SOL_SOCKET, SO_RCVBUF, (char*)&bufsize, &len) < 0)
        return -1;

    return bufsize;
}

int sendConnectReq(struct client_config* client_config,
                   struct net_config* net_config,
                   int haveServerAddress)
{
//cerr << "sending a connect request" << endl;
    struct connectReq connectReq;

    if (net_config->flags & FLAG_PASSIVE)
        return 0;

    connectReq.opCode = htons(CMD_CONNECT_REQ);
    connectReq.reserved = 0;
    connectReq.capabilities = htonl(RECEIVER_CAPABILITIES);
    connectReq.rcvbuf = htonl(udpc_getRcvBuf(client_config->S_UCAST));

    if (haveServerAddress)
        return SSEND(connectReq);
    else
        return BCAST_CONTROL(client_config->S_UCAST, connectReq);
}

void udpc_setRcvBuf(int sock, unsigned int bufsize)
{
    if (setsockopt(sock, SOL_SOCKET, SO_RCVBUF,
                   (char*) &bufsize, sizeof(bufsize)) < 0)
    {
        //TODO: FIXME
        //perror("Set receiver buffer");
    }
}

void udpc_copyFromMessage(struct sockaddr_in* dst, unsigned char* src)
{
    memcpy((char*) &dst->sin_addr, src, sizeof(struct in_addr));
}

unsigned short udpc_getPort(struct sockaddr_in* addr)
{
    return ntohs(((struct sockaddr_in*) addr)->sin_port);
}

int udpc_selectSock(int* socks, int nr, int startTimeout)
{
    fd_set read_set;
    int r;
    int maxFd;
    struct timeval tv, *tvp;

    if (startTimeout)
    {
        tv.tv_sec = startTimeout;
        tv.tv_usec = 0;
        tvp = &tv;
    }
    else
    {
        tvp = NULL;
    }

    maxFd = prepareForSelect(socks, nr, &read_set);
    r = select(maxFd + 1, &read_set, NULL, NULL, tvp);

    if (r < 0)
        return r;

    return udpc_getSelectedSock(socks, nr, &read_set);
}

void udpc_zeroSockArray(int* socks, int nr)
{
    int i;

    for (i = 0; i < nr; i++)
        socks[i] = -1;
}

unsigned char* getBlockSpace(struct clientState* clst)
{
    int pos;

    if (clst->localPos)
    {
        clst->localPos--;
        return clst->localBlockAddresses[clst->localPos];
    }

    pc_consume(clst->freeBlocks_pc, 1);
    pos = pc_getConsumerPosition(clst->freeBlocks_pc);
    pc_consumed(clst->freeBlocks_pc, 1);
    return clst->blockAddresses[pos];
}

void setNextBlock(struct clientState* clst)
{
    clst->nextBlock = getBlockSpace(clst);
}

int setupMessages(struct clientState* clst)
{
    /* the messages received from the server */
    clst->data_iov[0].iov_base = (void*)&clst->Msg;
    clst->data_iov[0].iov_len = sizeof(clst->Msg);

    /* namelen set just before reception */
    clst->data_hdr.msg_iov = clst->data_iov;
    /* iovlen set just before reception */

    initMsgHdr(&clst->data_hdr);
    return 0;
}

struct slice* initSlice(struct clientState* clst,
                        struct slice* slice,
                        int sliceNo)
{
    idbassert(slice->state == slice::SLICE_FREE || slice->state == slice::SLICE_RECEIVING);

    slice->magic = SLICEMAGIC;
    slice->state = slice::SLICE_RECEIVING;
    slice->blocksTransferred = 0;
    slice->dataBlocksTransferred = 0;
    BZERO(slice->retransmit);
    slice->freePos = 0;
    slice->bytes = 0;

    if (clst->currentSlice != NULL && !clst->currentSlice->bytesKnown)
    {
        //udpc_fatal(1, "Previous slice size not known\n");
    }

    if (clst->currentSliceNo != sliceNo - 1)
    {
        //udpc_fatal(1, "Slice no mismatch %d <-> %d\n", sliceNo, clst->currentSliceNo);
    }

    slice->bytesKnown = 0;
    slice->sliceNo = sliceNo;

    BZERO(slice->missing_data_blocks);
#ifdef BB_FEATURE_UDPCAST_FEC
    BZERO(slice->fec_stripes);
    BZERO(slice->fec_blocks);
    BZERO(slice->fec_descs);
#endif
    clst->currentSlice = slice;
    clst->currentSliceNo = sliceNo;
    return slice;
}
struct slice* newSlice(struct clientState* clst, int sliceNo)
{
    struct slice* slice = NULL;
    int i;

    pc_consume(clst->free_slices_pc, 1);
    i = pc_getConsumerPosition(clst->free_slices_pc);
    pc_consumed(clst->free_slices_pc, 1);
    slice = &clst->slices[i];
    idbassert(slice->state == slice::SLICE_FREE);

    /* wait for free data memory */
    slice->base = pc_getConsumerPosition(clst->fifo->freeMemQueue);
    pc_consume(clst->fifo->freeMemQueue,
               clst->net_config->blockSize * MAX_SLICE_SIZE);
    initSlice(clst, slice, sliceNo);
    return slice;
}

#define ADR(x, bs) (fifo->dataBuffer + \
	(slice->base+(x)*bs) % fifo->dataBufSize)

void closeAllExcept(struct clientState* clst, int fd)
{
    int i;
    int* socks = clst->client_config->socks;

    if (clst->selectedFd >= 0)
        return;

    clst->selectedFd = fd;

    for (i = 1; i < NR_CLIENT_SOCKS; i++)
        if (socks[i] != -1 && socks[i] != fd)
            udpc_closeSock(socks, NR_CLIENT_SOCKS, i);
}

struct slice* rcvFindSlice(struct clientState* clst, int sliceNo)
{
    if (! clst->currentSlice)
    {
        /* Streaming mode? */
        clst->currentSliceNo = sliceNo - 1;
        return newSlice(clst, sliceNo);
    }

    if (sliceNo <= clst->currentSliceNo)
    {
        struct slice* slice = clst->currentSlice;
        int pos = slice - clst->slices;
        idbassert(slice == NULL || slice->magic == SLICEMAGIC);

        while (slice->sliceNo != sliceNo)
        {
            if (slice->state == slice::SLICE_FREE)
                return NULL;

            idbassert(slice->magic == SLICEMAGIC);
            pos--;

            if (pos < 0)
                pos += NR_SLICES;

            slice = &clst->slices[pos];
        }

        return slice;
    }

    if (sliceNo != clst->currentSliceNo + 1)
    {
        //TODO: FIXME
        //udpc_flprintf("Slice %d skipped\n", sliceNo-1);
        //exit(1);
    }

    if ((clst->net_config->flags & FLAG_STREAMING) &&
            sliceNo != clst->currentSliceNo)
    {
        return initSlice(clst, clst->currentSlice, sliceNo);
    }

    if (sliceNo > clst->receivedSliceNo + 2)
    {

        //TODO: FIXME
#if 0
        slice_t slice = rcvFindSlice(clst, clst->receivedSliceNo + 1);
        udpc_flprintf("Dropped by server now=%d last=%d\n", sliceNo, clst->receivedSliceNo);

        if (slice != NULL)
            printMissedBlockMap(clst, slice);

        exit(1);
#endif
    }

    return newSlice(clst, sliceNo);
}

void setSliceBytes(struct slice* slice,
                   struct clientState* clst,
                   int bytes)
{
    idbassert(slice->magic == SLICEMAGIC);

    if (slice->bytesKnown)
    {
        if (slice->bytes != bytes)
        {
            //TODO: FIXME
            //udpc_fatal(1, "Byte number mismatch %d <-> %d\n", bytes, slice->bytes);
        }
    }
    else
    {
        slice->bytesKnown = 1;
        slice->bytes = bytes;

        if (bytes == 0)
            clst->netEndReached = 1;
    }
}

void advanceReceivedPointer(struct clientState* clst)
{
    int pos = clst->receivedPtr;

    while (1)
    {
        slice_t slice = &clst->slices[pos];

        if (
#ifdef BB_FEATURE_UDPCAST_FEC
            slice->state != SLICE_FEC &&
            slice->state != SLICE_FEC_DONE &&
#endif
            slice->state != slice::SLICE_DONE)
            break;

        pos++;
        clst->receivedSliceNo = slice->sliceNo;

        if (pos >= NR_SLICES)
            pos -= NR_SLICES;
    }

    clst->receivedPtr = pos;
}

void receiverStatsAddBytes(receiver_stats_t rs, long bytes)
{
    if (rs != NULL)
        rs->totalBytes += bytes;
}

void cleanupSlices(struct clientState* clst, unsigned int doneState)
{
    while (1)
    {
        int pos = pc_getProducerPosition(clst->free_slices_pc);
        int bytes;
        slice_t slice = &clst->slices[pos];

        if (slice->state != (signed)doneState)
            break;

        receiverStatsAddBytes(clst->stats, slice->bytes);
        bytes = slice->bytes;

        /* signal data received */
        if (bytes == 0)
        {
            pc_produceEnd(clst->fifo->data);
        }
        else
            pc_produce(clst->fifo->data, slice->bytes);

        /* free up slice structure */
        clst->slices[pos].state = slice::SLICE_FREE;
        pc_produce(clst->free_slices_pc, 1);

        /* if at end, exit this thread */
        if (!bytes)
        {
            clst->endReached = 2;
        }
    }
}

void checkSliceComplete(struct clientState* clst,
                        struct slice* slice)
{
    int blocksInSlice;

    idbassert(slice->magic == SLICEMAGIC);

    if (slice->state != slice::SLICE_RECEIVING)
        /* bad starting state */
        return;

    /* is this slice ready ? */
    idbassert(clst->net_config->blockSize != 0);
    blocksInSlice = (slice->bytes + clst->net_config->blockSize - 1) /
                    clst->net_config->blockSize;

    if (blocksInSlice == slice->blocksTransferred)
    {
        pc_consumed(clst->fifo->freeMemQueue, slice->bytes);
        clst->net_config->flags &= ~FLAG_STREAMING;

        if (blocksInSlice == slice->dataBlocksTransferred)
            slice->state = slice::SLICE_DONE;
        else
        {
#ifdef BB_FEATURE_UDPCAST_FEC
            idbassert(clst->use_fec == 1);
            slice->state = SLICE_FEC;
#else
            idbassert(0);
#endif
        }

        advanceReceivedPointer(clst);
#ifdef BB_FEATURE_UDPCAST_FEC

        if (clst->use_fec)
        {
            int n = pc_getProducerPosition(clst->fec_data_pc);
            idbassert(slice->state == SLICE_DONE || slice->state == SLICE_FEC);
            clst->fec_slices[n] = slice;
            pc_produce(clst->fec_data_pc, 1);
        }
        else
#endif
            cleanupSlices(clst, slice::SLICE_DONE);
    }
}

int processDataBlock(struct clientState* clst,
                     int sliceNo,
                     int blockNo,
                     int bytes)
{
//cerr << "processDataBlock(): " << sliceNo << ", " << blockNo << ", " << bytes << endl;
    struct fifo* fifo = clst->fifo;
    struct slice* slice = rcvFindSlice(clst, sliceNo);
    unsigned char* shouldAddress, *isAddress;

    idbassert(slice == NULL || slice->magic == SLICEMAGIC);

    if (slice == NULL ||
            slice->state == slice::SLICE_FREE ||
            slice->state == slice::SLICE_DONE
#ifdef BB_FEATURE_UDPCAST_FEC
            ||
            slice->state == SLICE_FEC ||
            slice->state == SLICE_FEC_DONE
#endif
       )
    {
        /* an old slice. Ignore */
//cerr << "ignore" << endl;
        return 0;
    }

    if (sliceNo > clst->currentSliceNo + 2)
    {
        //TODO: FIXME
        //udpc_fatal(1, "We have been dropped by sender\n");
    }

    if (BIT_ISSET(blockNo, slice->retransmit.map))
    {
        /* we already have this packet, ignore */
#if 0
        flprintf("Packet %d:%d not for us\n", sliceNo, blockNo);
#endif
//cerr << "dup" << endl;
        return 0;
    }

    if (slice->base % clst->net_config->blockSize)
    {
        //TODO: FIXME
        //udpc_fatal(1, "Bad base %d, not multiple of block size %d\n", slice->base, clst->net_config->blockSize);
//cerr << "bad base" << endl;
    }

//cerr << "good slice" << endl;

    shouldAddress = ADR(blockNo, clst->net_config->blockSize);
    isAddress = (unsigned char*)clst->data_hdr.msg_iov[1].iov_base;

    if (shouldAddress != isAddress)
    {
        /* copy message to the correct place */
        memcpy(shouldAddress, isAddress,  clst->net_config->blockSize);
    }

    if (clst->client_config->sender_is_newgen && bytes != 0)
        setSliceBytes(slice, clst, bytes);

    if (clst->client_config->sender_is_newgen && bytes == 0)
        clst->netEndReached = 0;

    SET_BIT(blockNo, slice->retransmit.map);
#ifdef BB_FEATURE_UDPCAST_FEC

    if (slice->fec_stripes)
    {
        int stripe = blockNo % slice->fec_stripes;
        slice->missing_data_blocks[stripe]--;
        idbassert(slice->missing_data_blocks[stripe] >= 0);

        if (slice->missing_data_blocks[stripe] <
                slice->fec_blocks[stripe])
        {
            int blockIdx;
            /* FIXME: FEC block should be enqueued in local queue here...*/
            slice->fec_blocks[stripe]--;
            blockIdx = stripe + slice->fec_blocks[stripe] * slice->fec_stripes;
            idbassert(slice->fec_descs[blockIdx].adr != 0);
            clst->localBlockAddresses[clst->localPos++] =
                slice->fec_descs[blockIdx].adr;
            slice->fec_descs[blockIdx].adr = 0;
            slice->blocksTransferred--;
        }
    }

#endif
    slice->dataBlocksTransferred++;
    slice->blocksTransferred++;

    while (slice->freePos < MAX_SLICE_SIZE &&
            BIT_ISSET(slice->freePos, slice->retransmit.map))
        slice->freePos++;

    checkSliceComplete(clst, slice);
    return 0;
}

int sendOk(struct client_config* client_config, unsigned int sliceNo)
{
    struct ok ok;
    ok.opCode = htons(CMD_OK);
    ok.reserved = 0;
    ok.sliceNo = htonl(sliceNo);
    return SSEND(ok);
}

int sendRetransmit(struct clientState* clst,
                   struct slice* slice,
                   int rxmit)
{
    struct client_config* client_config = clst->client_config;

    idbassert(slice->magic == SLICEMAGIC);
    slice->retransmit.opCode = htons(CMD_RETRANSMIT);
    slice->retransmit.reserved = 0;
    slice->retransmit.sliceNo = htonl(slice->sliceNo);
    slice->retransmit.rxmit = htonl(rxmit);
    return SSEND(slice->retransmit);
}

int processReqAck(struct clientState* clst,
                  int sliceNo, int bytes, int rxmit)
{
    struct slice* slice = rcvFindSlice(clst, sliceNo);
    int blocksInSlice;
    char* readySet = (char*) clst->data_hdr.msg_iov[1].iov_base;

    idbassert(slice == NULL || slice->magic == SLICEMAGIC);

    {
        struct timeval tv;
        gettimeofday(&tv, 0);
        /* usleep(1); DEBUG: FIXME */
    }

    if (BIT_ISSET(clst->client_config->clientNumber, readySet))
    {
        /* not for us */
        return 0;
    }

    if (slice == NULL)
    {
        /* an old slice => send ok */
        return sendOk(clst->client_config, sliceNo);
    }

    setSliceBytes(slice, clst, bytes);
    idbassert(clst->net_config->blockSize != 0);
    blocksInSlice = (slice->bytes + clst->net_config->blockSize - 1) /
                    clst->net_config->blockSize;

    if (blocksInSlice == slice->blocksTransferred)
    {
        /* send ok */
        sendOk(clst->client_config, slice->sliceNo);
    }
    else
    {
        sendRetransmit(clst, slice, rxmit);
    }

    checkSliceComplete(clst, slice); /* needed for the final 0 sized slice */
    advanceReceivedPointer(clst);
#ifdef BB_FEATURE_UDPCAST_FEC

    if (!clst->use_fec)
        cleanupSlices(clst, SLICE_DONE);

#endif
    return 0;
}

int udpc_isAddressEqual(struct sockaddr_in* a, struct sockaddr_in* b)
{
    return !memcmp((char*) a, (char*)b, 8);
}
int dispatchMessage(struct clientState* clst)
{
    int ret;
    struct sockaddr_in lserver;
    struct fifo* fifo = clst->fifo;
    int fd = -1;
    struct client_config* client_config = clst->client_config;

    /* set up message header */
    if (clst->currentSlice != NULL &&
            clst->currentSlice->freePos < MAX_SLICE_SIZE)
    {
        struct slice* slice = clst->currentSlice;
        idbassert(slice == NULL || slice->magic == SLICEMAGIC);
        clst->data_iov[1].iov_base =
            ADR(slice->freePos, clst->net_config->blockSize);
    }
    else
    {
        clst->data_iov[1].iov_base = clst->nextBlock;
    }

    clst->data_iov[1].iov_len = clst->net_config->blockSize;
    clst->data_hdr.msg_iovlen = 2;

    clst->data_hdr.msg_name = &lserver;
    clst->data_hdr.msg_namelen = sizeof(struct sockaddr_in);

    while (clst->endReached || clst->netEndReached)
    {
        int oldEndReached = clst->endReached;
        int nr_desc;
        struct timeval tv;
        fd_set read_set;

        int maxFd = prepareForSelect(client_config->socks,
                                     NR_CLIENT_SOCKS, &read_set);

        tv.tv_sec = clst->net_config->exitWait / 1000;
        tv.tv_usec = (clst->net_config->exitWait % 1000) * 1000;
        nr_desc = select(maxFd,  &read_set, 0, 0,  &tv);

        if (nr_desc < 0)
        {
            break;
        }

        fd = udpc_getSelectedSock(client_config->socks, NR_CLIENT_SOCKS, &read_set);

        if (fd >= 0)
            break;

        /* Timeout expired */
        if (oldEndReached >= 2)
        {
            clst->endReached = 3;
            return 0;
        }
    }

    if (fd < 0)
        fd = clst->selectedFd;

    if (fd < 0)
    {
        struct timeval tv, *tvp;
        fd_set read_set;
        int maxFd = prepareForSelect(client_config->socks,
                                     NR_CLIENT_SOCKS, &read_set);
        clst->promptPrinted = 1;

        if (clst->net_config->startTimeout == 0)
        {
            tvp = NULL;
        }
        else
        {
            tv.tv_sec = clst->net_config->startTimeout;
            tv.tv_usec = 0;
            tvp = &tv;
        }

//cerr << "waiting for data..." << endl;
        ret = selectWithoutConsole(maxFd + 1, &read_set, tvp);

        if (ret < 0)
        {
            perror("Select");
            return 0;
        }

        if (ret == 0)
        {
            clst->endReached = 3;
            clst->netEndReached = 3;
            pc_produceEnd(clst->fifo->data);
            return 1;
        }

        fd = udpc_getSelectedSock(clst->client_config->socks,
                                  NR_CLIENT_SOCKS, &read_set);
    }

#ifdef LOSSTEST
    loseRecvPacket(fd);
    ret = RecvMsg(fd, &clst->data_hdr, 0);
#else
    ret = recvmsg(fd, &clst->data_hdr, 0);
//cerr << "got some data on fd " << fd << endl;
#endif

    if (ret < 0)
    {
#if DEBUG
        flprintf("data recvfrom %d: %s\n", fd, strerror(errno));
#endif
        return -1;
    }

#if 0
    fprintf(stderr, "received packet for slice %d, block %d\n",
            ntohl(Msg.sliceNo), ntohs(db.blockNo));
#endif

    if (!udpc_isAddressEqual(&lserver,
                             &clst->client_config->serverAddr))
    {
        return -1;
    }

    switch (ntohs(clst->Msg.opCode))
    {
        case CMD_DATA:
//cerr << "got CMD_DATA" << endl;
            closeAllExcept(clst, fd);
            //udpc_receiverStatsStartTimer(clst->stats);
            clst->client_config->isStarted = 1;
            return processDataBlock(clst,
                                    ntohl(clst->Msg.dataBlock.sliceNo),
                                    ntohs(clst->Msg.dataBlock.blockNo),
                                    ntohl(clst->Msg.dataBlock.bytes));
#ifdef BB_FEATURE_UDPCAST_FEC

        case CMD_FEC:
//cerr << "got CMD_FEC" << endl;
            closeAllExcept(clst, fd);
            //receiverStatsStartTimer(clst->stats);
            clst->client_config->isStarted = 1;
            return processFecBlock(clst,
                                   ntohs(clst->Msg.fecBlock.stripes),
                                   ntohl(clst->Msg.fecBlock.sliceNo),
                                   ntohs(clst->Msg.fecBlock.blockNo),
                                   ntohl(clst->Msg.fecBlock.bytes));
#endif

        case CMD_REQACK:
//cerr << "got CMD_REQACK" << endl;
            closeAllExcept(clst, fd);
            //receiverStatsStartTimer(clst->stats);
            clst->client_config->isStarted = 1;
            return processReqAck(clst,
                                 ntohl(clst->Msg.reqack.sliceNo),
                                 ntohl(clst->Msg.reqack.bytes),
                                 ntohl(clst->Msg.reqack.rxmit));

        case CMD_HELLO_STREAMING:
        case CMD_HELLO_NEW:
        case CMD_HELLO:
//cerr << "got CMD_HELLO" << endl;
            /* retransmission of hello to find other participants ==> ignore */
            return 0;

        default:
            break;
    }

    return -1;
}

THREAD_RETURN netReceiverMain(void* args0)
{
    struct clientState* clst = (struct clientState*) args0;

    clst->currentSliceNo = 0;
    setupMessages(clst);

    clst->currentSliceNo = -1;
    clst->currentSlice = NULL;
    clst->promptPrinted = 0;

    if (! (clst->net_config->flags & FLAG_STREAMING))
        newSlice(clst, 0);
    else
    {
        clst->currentSlice = NULL;
        clst->currentSliceNo = 0;
    }

    while (clst->endReached < 3)
    {
        dispatchMessage(clst);
    }

#ifdef BB_FEATURE_UDPCAST_FEC

    if (clst->use_fec)
        pthread_join(clst->fec_thread, NULL);

#endif
    return 0;
}

int spawnNetReceiver(struct fifo* fifo,
                     struct client_config* client_config,
                     struct net_config* net_config)
{
    int i;
    struct clientState*  clst = MALLOC(struct clientState);
    clst->fifo = fifo;
    clst->client_config = client_config;
    clst->net_config = net_config;
    //clst->stats = stats;
    clst->endReached = 0;
    clst->netEndReached = 0;
    clst->selectedFd = -1;

    clst->free_slices_pc = pc_makeProduconsum(NR_SLICES, "free slices");
    pc_produce(clst->free_slices_pc, NR_SLICES);

    for (i = 0; i < NR_SLICES; i++)
        clst->slices[i].state = slice::SLICE_FREE;

    clst->receivedPtr = 0;
    clst->receivedSliceNo = 0;

#ifdef BB_FEATURE_UDPCAST_FEC
    fec_init(); /* fec new involves memory
		 * allocation. Better do it here */
    clst->use_fec = 0;
    clst->fec_data_pc = pc_makeProduconsum(NR_SLICES, "fec data");
#endif

#define NR_BLOCKS 4096
    clst->freeBlocks_pc = pc_makeProduconsum(NR_BLOCKS, "free blocks");
    pc_produce(clst->freeBlocks_pc, NR_BLOCKS);
    clst->blockAddresses = (unsigned char**)calloc(NR_BLOCKS, sizeof(char*));
    clst->localBlockAddresses = (unsigned char**)calloc(NR_BLOCKS, sizeof(char*));
    clst->blockData = (unsigned char*)malloc(NR_BLOCKS * net_config->blockSize);

    for (i = 0; i < NR_BLOCKS; i++)
        clst->blockAddresses[i] = clst->blockData + i * net_config->blockSize;

    clst->localPos = 0;

    setNextBlock(clst);
    return pthread_create(&client_config->thread, NULL, netReceiverMain, clst);
}

unsigned int pc_getSize(produconsum_t pc)
{
    return pc->size;
}

int writer(struct fifo* fifo, SBS outbs)
{
//cerr << "start appending to outbs" << endl;
    outbs->restart();
    int fifoSize = pc_getSize(fifo->data);

    while (1)
    {
        int pos = pc_getConsumerPosition(fifo->data);
        int bytes = pc_consumeContiguousMinAmount(fifo->data, BLOCKSIZE);

        if (bytes == 0)
        {
//cerr << "done appending to outbs: " << outbs->length() << endl;
            return 0;
        }

        /*
         * If we have more than blocksize, round down to nearest blocksize
         * multiple
         */
        if (pos + bytes != fifoSize &&
                bytes > (pos + bytes) % BLOCKSIZE)
            bytes -= (pos + bytes) % BLOCKSIZE;

        /* make sure we don't write to big a chunk... Better to
         * liberate small chunks one by one rather than attempt to
         * write out a bigger chunk and block reception for too
         * long */
        if (bytes > 128 * 1024)
            bytes = 64 * 1024;

        //bytes = write(outFile, fifo->dataBuffer + pos, bytes);
//cerr << "appending " << bytes << " bytes to outbs..." << endl;
        outbs->append(fifo->dataBuffer + pos, bytes);
        pc_consumed(fifo->data, bytes);
        pc_produce(fifo->freeMemQueue, bytes);
    }
}
}

namespace multicast
{

MulticastImpl::MulticastImpl(int min_receivers, const string& ifName, int portBase, int bufSize) :
    fIfName(ifName),
    fDb(0)
{
    udpc_clearIp(&fNet_config.dataMcastAddr);
    fNet_config.mcastRdv = 0;
    fNet_config.blockSize = 1024; //1456;
    fNet_config.sliceSize = 16;
    fNet_config.portBase = portBase;
    fNet_config.nrGovernors = 0;
    fNet_config.flags = FLAG_NOKBD;
    fNet_config.capabilities = 0;
    fNet_config.min_slice_size = 16;
    fNet_config.max_slice_size = 1024;
    fNet_config.default_slice_size = 0;
    fNet_config.ttl = 1;
    fNet_config.rexmit_hello_interval = 2000;
    fNet_config.autostart = 0;
    fNet_config.requestedBufSize = bufSize;

    fNet_config.min_receivers = min_receivers;
    fNet_config.max_receivers_wait = 0;
    fNet_config.min_receivers_wait = 0;

    fNet_config.retriesUntilDrop = 200;

    fNet_config.rehelloOffset = 50;

    fStat_config.log = 0;
    fStat_config.bwPeriod = 0;
    fStat_config.printUncompressedPos = -1;
    fStat_config.statPeriod = DEFLT_STAT_PERIOD;

    fNet_config.net_if = 0;

    //full-duplex
    fNet_config.flags |= FLAG_SN;

    if (fIfName.empty())
        fIfName = "eth0";

    fSock[0] = -1;
    fSock[1] = -1;
    fSock[2] = -1;
}

MulticastImpl::~MulticastImpl()
{
    delete fDb;

    for (int i = 0; i < 3; i++)
    {
        if (fSock[i] >= 0)
        {
            shutdown(fSock[i], SHUT_RDWR);
            close(fSock[i]);
        }
    }
}

void MulticastImpl::startSender()
{
    int tries;
    time_t firstConnected = 0;
    time_t* firstConnectedP;

    //participantsDb_t db;

    /* make the socket and print banner */
    //int fSock[3];
    int nr = 0;
    int fd;
    int r;
    int j;

    fNet_config.net_if = udpc_getNetIf(fIfName.c_str());

    fSock[nr++] = udpc_makeSocket(ADDR_TYPE_UCAST,
                                  fNet_config.net_if,
                                  NULL,
                                  SENDER_PORT(fNet_config.portBase));
//cerr << "sock[" << (nr-1) << "] = " << fSock[(nr-1)] << endl;

    if (! (fNet_config.flags & (FLAG_SN | FLAG_NOTSN)) )
    {
        if (udpc_isFullDuplex(fSock[0], fNet_config.net_if->name) == 1)
        {
            fNet_config.flags |= FLAG_SN;
        }
    }

    fd = udpc_makeSocket(ADDR_TYPE_BCAST,
                         fNet_config.net_if,
                         NULL,
                         SENDER_PORT(fNet_config.portBase));

    if (fd >= 0)
        fSock[nr++] = fd;

//cerr << "sock[" << (nr-1) << "] = " << fSock[(nr-1)] << endl;

    if (fNet_config.requestedBufSize)
        udpc_setSendBuf(fSock[0], fNet_config.requestedBufSize);

    fNet_config.controlMcastAddr.sin_addr.s_addr = 0;

    if (fNet_config.ttl == 1 && fNet_config.mcastRdv == NULL)
    {
        udpc_getBroadCastAddress(fNet_config.net_if,
                                 &fNet_config.controlMcastAddr,
                                 RECEIVER_PORT(fNet_config.portBase));
        udpc_setSocketToBroadcast(fSock[0]);
    }

    if (fNet_config.controlMcastAddr.sin_addr.s_addr == 0)
    {
        udpc_getMcastAllAddress(&fNet_config.controlMcastAddr,
                                fNet_config.mcastRdv,
                                RECEIVER_PORT(fNet_config.portBase));

        /* Only do the following if controlMcastAddr is indeed an
           mcast address ... */
        if (isMcastAddress(&fNet_config.controlMcastAddr))
        {
            udpc_setMcastDestination(fSock[0], fNet_config.net_if,
                                     &fNet_config.controlMcastAddr);
            udpc_setTtl(fSock[0], fNet_config.ttl);
            fSock[nr++] = udpc_makeSocket(ADDR_TYPE_MCAST,
                                          fNet_config.net_if,
                                          &fNet_config.controlMcastAddr,
                                          SENDER_PORT(fNet_config.portBase));
//cerr << "sock[" << (nr-1) << "] = " << fSock[(nr-1)] << endl;
        }
    }

    if (!(fNet_config.flags & FLAG_POINTOPOINT) &&
            udpc_ipIsZero(&fNet_config.dataMcastAddr))
    {
        udpc_getDefaultMcastAddress(fNet_config.net_if,
                                    &fNet_config.dataMcastAddr);
    }

    if (fNet_config.flags & FLAG_POINTOPOINT)
    {
        udpc_clearIp(&fNet_config.dataMcastAddr);
    }

    setPort(&fNet_config.dataMcastAddr, RECEIVER_PORT(fNet_config.portBase));

    fNet_config.capabilities = SENDER_CAPABILITIES;

    if (fNet_config.flags & FLAG_ASYNC)
        fNet_config.capabilities |= CAP_ASYNC;

    udpc_sendHello(&fNet_config, fSock[0], 0);

    fDb = udpc_makeParticipantsDb();
    tries = 0;

    if (fNet_config.min_receivers || fNet_config.min_receivers_wait ||
            fNet_config.max_receivers_wait)
        firstConnectedP = &firstConnected;
    else
        firstConnectedP = NULL;

    while (!(r = mainDispatcher(fSock, nr, fDb, &fNet_config, &tries, firstConnectedP)))
        ;

    for (j = 1; j < nr; j++)
        if (fSock[j] != fSock[0])
            closesocket(fSock[j]);

    if (r == 1)
    {
        int i;

        for (i = 1; i < nr; i++)
            udpc_closeSock(fSock, nr, i);
    }

    //doTransfer(fSock[0], fDb, &fNet_config, &fStat_config);
}

void MulticastImpl::doTransfer(const uint8_t* buf, uint32_t len)
{
    int i;
    int ret;
    struct fifo fifo;
    int isPtP = isPointToPoint(fDb, fNet_config.flags);

    fNet_config.rcvbuf = 0;

    for (i = 0; i < MAX_CLIENTS; i++)
        if (udpc_isParticipantValid(fDb, i))
        {
            unsigned int pRcvBuf = udpc_getParticipantRcvBuf(fDb, i);

            if (isPtP)
                udpc_copyIpFrom(&fNet_config.dataMcastAddr,
                                udpc_getParticipantIp(fDb, i));

            fNet_config.capabilities &=
                udpc_getParticipantCapabilities(fDb, i);

            if (pRcvBuf != 0 &&
                    (fNet_config.rcvbuf == 0 || fNet_config.rcvbuf > pRcvBuf))
                fNet_config.rcvbuf = pRcvBuf;
        }

    if (isMcastAddress(&fNet_config.dataMcastAddr))
        udpc_setMcastDestination(fSock[0], fNet_config.net_if,
                                 &fNet_config.dataMcastAddr);

    if (! (fNet_config.capabilities & CAP_BIG_ENDIAN))
    {
        //TODO: FIXME
        //udpc_fatal(1, "Peer with incompatible endianness");
    }

    if (! (fNet_config.capabilities & CAP_NEW_GEN))
    {
        fNet_config.dataMcastAddr = fNet_config.controlMcastAddr;
        fNet_config.flags &= ~(FLAG_SN | FLAG_ASYNC);
    }

    if (fNet_config.flags & FLAG_BCAST)
        fNet_config.dataMcastAddr = fNet_config.controlMcastAddr;

    udpc_initFifo(&fifo, fNet_config.blockSize);
    ret = spawnNetSender(&fifo, fSock[0], &fNet_config, fDb);
    localReader(&fifo, buf, len);

    pthread_join(fifo.thread, NULL);

}

void MulticastImpl::startReceiver()
{
    union serverControlMsg Msg;
    int connectReqSent = 0;
    struct sockaddr_in myIp;
    int haveServerAddress;

    fClient_config.sender_is_newgen = 0;

    fNet_config.net_if = udpc_getNetIf(fIfName.c_str());
    {
        fprintf(stderr, "net_if:\n\taddr = 0x%x\n\tbcast = 0x%x\n\tname = %s\n\tindex = %d\n", fNet_config.net_if->addr.s_addr, fNet_config.net_if->bcast.s_addr, fNet_config.net_if->name, fNet_config.net_if->index);
    }

    udpc_zeroSockArray(fClient_config.socks, NR_CLIENT_SOCKS);

    fClient_config.S_UCAST = udpc_makeSocket(ADDR_TYPE_UCAST,
                             fNet_config.net_if,
                             0, RECEIVER_PORT(fNet_config.portBase));
//cerr << "S_UCAST = " << fClient_config.S_UCAST << endl;
    fClient_config.S_BCAST = udpc_makeSocket(ADDR_TYPE_BCAST,
                             fNet_config.net_if,
                             0, RECEIVER_PORT(fNet_config.portBase));
//cerr << "S_BCAST = " << fClient_config.S_BCAST << endl;

    if (fNet_config.ttl == 1 && fNet_config.mcastRdv == NULL)
    {
        udpc_getBroadCastAddress(fNet_config.net_if,
                                 &fNet_config.controlMcastAddr,
                                 SENDER_PORT(fNet_config.portBase));
        udpc_setSocketToBroadcast(fClient_config.S_UCAST);
    }
    else
    {
        udpc_getMcastAllAddress(&fNet_config.controlMcastAddr,
                                fNet_config.mcastRdv,
                                SENDER_PORT(fNet_config.portBase));

        if (isMcastAddress(&fNet_config.controlMcastAddr))
        {
            udpc_setMcastDestination(fClient_config.S_UCAST, fNet_config.net_if,
                                     &fNet_config.controlMcastAddr);
            udpc_setTtl(fClient_config.S_UCAST, fNet_config.ttl);

            fClient_config.S_MCAST_CTRL =
                udpc_makeSocket(ADDR_TYPE_MCAST,
                                fNet_config.net_if,
                                &fNet_config.controlMcastAddr,
                                RECEIVER_PORT(fNet_config.portBase));
//cerr << "S_MCAST_CTRL = " << fClient_config.S_MCAST_CTRL << endl;
            // TODO: subscribe address as receiver to!
        }
    }

    udpc_clearIp(&fNet_config.dataMcastAddr);

    connectReqSent = 0;
    haveServerAddress = 0;

    fClient_config.clientNumber = 0; /*default number for asynchronous transfer*/

    while (1)
    {
        // int len;
        int msglen;
        int sock;

        if (!connectReqSent)
        {
            if (sendConnectReq(&fClient_config, &fNet_config,
                               haveServerAddress) < 0)
            {
                //TODO: FIXME
                //perror("sendto to locate server");
            }

            connectReqSent = 1;
        }

        haveServerAddress = 0;

//cerr << "waiting for msg..." << flush << endl;
        sock = udpc_selectSock(fClient_config.socks, NR_CLIENT_SOCKS,
                               fNet_config.startTimeout);

//cerr << "got something" << endl;
        if (sock < 0)
        {
            //TODO: FIXME
        }

        // len = sizeof(server);
        msglen = RECV(sock,
                      Msg, fClient_config.serverAddr, fNet_config.portBase);

        if (msglen < 0)
        {
            //TODO: FIXME
            //perror("recvfrom to locate server");
            //exit(1);
        }

        if (udpc_getPort(&fClient_config.serverAddr) !=
                SENDER_PORT(fNet_config.portBase))
            /* not from the right port */
            continue;

        switch (ntohs(Msg.opCode))
        {
            case CMD_CONNECT_REPLY:
//cerr << "got conrep" << endl;
                fClient_config.clientNumber = ntohl(Msg.connectReply.clNr);
                fNet_config.blockSize = ntohl(Msg.connectReply.blockSize);

                if (ntohl(Msg.connectReply.capabilities) & CAP_NEW_GEN)
                {
                    fClient_config.sender_is_newgen = 1;
                    udpc_copyFromMessage(&fNet_config.dataMcastAddr,
                                         Msg.connectReply.mcastAddr);
                }

                if (fClient_config.clientNumber == -1)
                {
                    //TODO: FIXME
                    //udpc_fatal(1, "Too many clients already connected\n");
                }

                goto break_loop;

            case CMD_HELLO_STREAMING:
            case CMD_HELLO_NEW:
            case CMD_HELLO:
//cerr << "got hello" << endl;
                connectReqSent = 0;

                if (ntohs(Msg.opCode) == CMD_HELLO_STREAMING)
                    fNet_config.flags |= FLAG_STREAMING;

                if (ntohl(Msg.hello.capabilities) & CAP_NEW_GEN)
                {
                    fClient_config.sender_is_newgen = 1;
                    udpc_copyFromMessage(&fNet_config.dataMcastAddr,
                                         Msg.hello.mcastAddr);
                    fNet_config.blockSize = ntohs(Msg.hello.blockSize);

                    if (ntohl(Msg.hello.capabilities) & CAP_ASYNC)
                        fNet_config.flags |= FLAG_PASSIVE;

                    if (fNet_config.flags & FLAG_PASSIVE)
                        goto break_loop;
                }

                haveServerAddress = 1;
                continue;

            case CMD_CONNECT_REQ:
            case CMD_DATA:
            case CMD_FEC:
                continue;

            default:
                break;
        }

        //TODO: FIXME
        //udpc_fatal(1, "Bad server reply %04x. Other transfer in progress?\n", (unsigned short) ntohs(Msg.opCode));
    }

break_loop:

    udpc_getMyAddress(fNet_config.net_if, &myIp);

    if (!udpc_ipIsZero(&fNet_config.dataMcastAddr)  &&
            !udpc_ipIsEqual(&fNet_config.dataMcastAddr, &myIp) &&
            (udpc_ipIsZero(&fNet_config.controlMcastAddr) ||
             !udpc_ipIsEqual(&fNet_config.dataMcastAddr, &fNet_config.controlMcastAddr)
            ))
    {
        fClient_config.S_MCAST_DATA =
            udpc_makeSocket(ADDR_TYPE_MCAST, fNet_config.net_if,
                            &fNet_config.dataMcastAddr,
                            RECEIVER_PORT(fNet_config.portBase));
//cerr << "S_MCAST_DATA = " << fClient_config.S_MCAST_DATA << endl;
    }


    if (fNet_config.requestedBufSize)
    {
        int i;

        for (i = 0; i < NR_CLIENT_SOCKS; i++)
            if (fClient_config.socks[i] != -1)
                udpc_setRcvBuf(fClient_config.socks[i], fNet_config.requestedBufSize);
    }

}

void MulticastImpl::receive(SBS outbs)
{
    struct fifo fifo;

    udpc_initFifo(&fifo, fNet_config.blockSize);
    fifo.data = pc_makeProduconsum(fifo.dataBufSize, "receive");
    fClient_config.isStarted = 0;

    spawnNetReceiver(&fifo, &fClient_config, &fNet_config);
    writer(&fifo, outbs);

    pthread_join(fClient_config.thread, NULL);
}

}

