mariadb-columnstore-engine/utils/multicast/impl.cpp

/* 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);
}

}