/* $Id: protocol.c,v 1.208 2004/11/04 21:25:17 graziano Exp $ */

#include "config_portability.h"

#include <unistd.h>
#include <signal.h> 
#include <netinet/in.h>   /* sometimes required for #include <arpa/inet.h> */
#include <netinet/tcp.h>  /* TCP_NODELAY */
#ifdef HAVE_INTTYPES_H
#	include <inttypes.h>
#endif
#include <arpa/inet.h>    /* inet_ntoa() */
#include <netdb.h>        /* getprotobyname() */
#if TIME_WITH_SYS_TIME
#	include <sys/time.h>
#	include <time.h>
#else
#	if HAVE_SYS_TIME_H
#		include <sys/time.h>
#	else
#		include <time.h>
#	endif
#endif
#include <errno.h>        /* errno */
#include <sys/wait.h>     /* waitpid() */
#include <sys/socket.h>   /* getpeername() socket() */
#include <stdlib.h>
#ifdef HAVE_FCNTL_H
#	include <fcntl.h>
#endif

#include "diagnostic.h"
#include "osutil.h"
#include "protocol.h"
#include "strutil.h"
#include "dnsutil.h"
#include "timeouts.h"


static void *lock = NULL;		/* local mutex */

/* Global variables: they need to be protected with locks when accessed */
#define MAX_NOTIFIES 40
static SocketFunction disconnectFunctions[MAX_NOTIFIES];
static NewConnectionFunction newFunctions[MAX_NOTIFIES];
static fd_set connectedEars;
static fd_set connectedPipes;
static fd_set connectedSockets;
static fd_set inUse;

/* this pipe is the one used by SocketIsAvailable to notify that a socket
 * needs to be listen to. It sends a message, thus forcing
 * ListenForMessage to be restarted: avoid missing messages when using
 * threads. ReadAvailable is the read end, and WriteAvailable is the
 * write end. */
static int ReadAvailable = NO_SOCKET;
static int WriteAvailable = NO_SOCKET;
static int written = 0;

/* This is used when it's time to call CloseDisconnections(): when
 * receiving a SIGPIPE, this flag is set to 1 so that IncomingRequest
 * will call CloseDisconnections to pick up the disconnected socket  */
static short needDisconnect = 0;


/* initialize the static variables! */
static void 
InitModule() {
	static int done = 0;
	int filedes[2];

	if (done == 0) {
		GetNWSLock(&lock);
		for (done = 0; done < MAX_NOTIFIES; done++) {
			disconnectFunctions[done] = NULL;
			newFunctions[done] = NULL;
		}
		FD_ZERO(&connectedEars);
		FD_ZERO(&connectedPipes);
		FD_ZERO(&connectedSockets);
		FD_ZERO(&inUse);

		/* create the pipe for sockes is available messages */
		if (pipe(filedes) == -1) {
			WARN("InitModule: couldn't create pipe!\n");
		} else {
			ReadAvailable = filedes[0];
			WriteAvailable = filedes[1];
			FD_SET(ReadAvailable, &connectedPipes);
		}

		done = 1;
		ReleaseNWSLock(&lock);
	}
}


static int
TcpProtoNumber(void);


/*
 * Remove #sock# from all maintained socket sets.
 *
 * It should be thread safe.
 */
void
ClearSocket(Socket sock) {

	/* operates on global variables */
	GetNWSLock(&lock);
	FD_CLR(sock, &connectedPipes);
	FD_CLR(sock, &connectedSockets);
	FD_CLR(sock, &connectedEars);
	/* clear also the inUse state */
	FD_CLR(sock, &inUse);
	ReleaseNWSLock(&lock);
}


/* It should be thread safe. Set keepalive and nodelay options by default
 * and if #reuse# set the REUSEADDR option. If #bufSize# is bigger then
 * 0 will set RCVBUF and SNDBUF with that value. */
int
ConditionSocket(	Socket sd,
			int reuse,
			int bufSize) {
	int i = 1, ret = 0, size = 1;
	SOCKLEN_T s = sizeof(int);
#ifdef USE_NONBLOCKING
	int flags;
#endif

#ifdef USE_NONBLOCKING
#ifdef HAVE_FCNTL
	/* let's try to set the a non-blocking socket */
	if ((flags = fcntl(sd,  F_GETFL, 0)) < 0) {
		WARN("ConditionSocket: fcntl failed.\n");
	} else if (fcntl(sd,  F_SETFL, flags | O_NONBLOCK) < 0) {
		WARN("ConditionSocket: fcntl failed.\n");
	}
#else
	WARN("ConditionSocket: fcntl wasn't found on this machine\n");
#endif		/* HAVE_FCNTL */
#endif		/* USE_NONBLOCKING */

	ret += setsockopt(sd, SOL_SOCKET, SO_KEEPALIVE, (char *)&i, s);
	ret += setsockopt(sd, TcpProtoNumber(), TCP_NODELAY, (char *)&i, s);

	if (reuse) {
		ret += setsockopt(sd, SOL_SOCKET, SO_REUSEADDR, (char *)&i, s);
	}

	if (bufSize > 0) {
		i = bufSize;
		ret += setsockopt(sd, SOL_SOCKET, SO_RCVBUF, (char *)&i, s);
		ret += setsockopt(sd, SOL_SOCKET, SO_SNDBUF, (char *)&i, s);
		if (!getsockopt(sd, SOL_SOCKET, SO_RCVBUF, (char *)&i, &s)) {
			size = i/bufSize;
			if (size != 2) {
				size = 1;
			}
		}
	}

	/* let's return the size of the socket buffer or 0 for failure */
	if (ret != 0) {
		WARN("ConditionSocket: setsockopt failed\n");
		ret = 0;
	} else {
		ret = size;
	}

	return (ret);
}


/*
 * Notifies all registered functions that #sock# has been closed.
 */
static void
DoDisconnectNotification(Socket sock) {
	int i;
	SocketFunction tmp;

	InitModule();

	/* now we have to play musicale chairs: we need to call the
	 * function while *not* keeping the lock (since the user might
	 * call function in this module) but we need the lock in case
	 * disconnectFunctions changes */
	for(i = 0; i < MAX_NOTIFIES; i++) {
		GetNWSLock(&lock);
		tmp = disconnectFunctions[i];
		ReleaseNWSLock(&lock);
		if (tmp != NULL) {
			tmp(sock);
		}
	}
}

/* notifies that a nex connection has been made on #ear# */
static void
DoNotifyConnection(	Socket ear, 
			Socket newSock) {
	int i;
	NewConnectionFunction tmp;

	InitModule();

	/* read note on DoNotifyOnDisconnection for why we do this weird
	 * tmp business */
	for(i = 0; i < MAX_NOTIFIES; i++) {
		GetNWSLock(&lock);
		tmp = newFunctions[i];
		ReleaseNWSLock(&lock);
		if (tmp != NULL) {
			tmp(ear, newSock);
		}
	}
}



/*
 * Returns the tcp protocol number from the network protocol data base.
 * 
 * getprotobyname() is not thread safe. We need to lock it.
 */
static int
TcpProtoNumber(void) {
	struct protoent *fetchedEntry;
	static int returnValue = 0;

	if(returnValue == 0) {
		GetNWSLock(&lock);
		fetchedEntry = getprotobyname("tcp");
		if(fetchedEntry != NULL) {
			returnValue = fetchedEntry->p_proto;
		}
		ReleaseNWSLock(&lock);
	}

	return returnValue;
}


/* thread safe */
int
CallAddrBuff(	IPAddress addr,
		unsigned short port,
		int bufSize,
		Socket *sock,
		double timeOut) {

	struct sockaddr_in server; /* remote host address */
	Socket sd;
	double start,
		ltimeout;
	int tmp_errno, ret, done;
	char *peer;
#ifdef USE_NONBLOCKING
	struct timeval tout;
	fd_set rds, wds;
	int err, len;
#endif

	InitModule();

	/* sanity check */
	if (sock == NULL) {
		ERROR("CallAddrBuff: socket pointer is NULL\n");
		return 0;
	}

	tmp_errno = ret = 0;		/* makes the compiler happy */

	memset((char *)&server, 0, sizeof(server));
	server.sin_addr.s_addr = addr.addr;
	server.sin_family = AF_INET;
	server.sin_port = htons((u_short)port);

	sd = socket(AF_INET, SOCK_STREAM, 0);
	if(sd < 0) {
		*sock = NO_SOCKET;
		WARN("CallAddrBuff: cannot create socket to server\n");
		return 0;
	}

	ConditionSocket(sd, 0, bufSize);

	/* set the adaptive timeout or the user selected one */
	ltimeout = 0;
	if (timeOut >= 0) {
		ltimeout = timeOut;
	} else {
		/* adaptive timeouts */
		ltimeout = GetTimeOut(CONN, addr, 0);
	}
#ifdef USE_ALARM_SIGNAL
	if (ltimeout > 0) {
		DDEBUG1("CallAddrBuff: setting timer to %.2f\n", ltimeout);
	}
#endif

	/* let's time it */
	start = MicroTime();

	done = 0;
	while (!done) {
#ifdef USE_ALARM_SIGNAL
		if (timeOut != 0) {
			/* do we still have time left? */
			ltimeout -= (MicroTime() - start)/1000000;
			if (ltimeout > 0) {
				/* set the timer */
				ret = (unsigned int) ltimeout;
				ret = (ret > 0) ? ret : 1;
				SetRealTimer(ret);
			} else {
				/* timeout */
				done = 1;
				break;
			}
		}
#endif

  
		ret = connect(sd, (struct sockaddr *)&server, sizeof(server));
		/* save a copy of errno */
		tmp_errno = errno;

#ifdef USE_ALARM_SIGNAL
		/* reset the timer */
		if (timeOut != 0) {
			RESETREALTIMER;
		}
#endif
		done = 1;
	}

#ifdef USE_NONBLOCKING
	if (ret < 0 && tmp_errno == EINPROGRESS) {
		/* let's wait till it's ready */
		FD_ZERO(&rds); FD_SET(sd, &rds);
		FD_ZERO(&wds); FD_SET(sd, &wds);

		/* set the timeout */
		tout.tv_sec = (unsigned long)ltimeout;
		tout.tv_usec = (unsigned long)(ltimeout - (double)tout.tv_sec)*1000000;
		ret = PortabilitySelect(sd+1, &rds, &wds, NULL, &tout, &tmp_errno);
		/* let's see what happened */
		if (ret > 0) {
			/* we got something: let's check there is no error */
			len = sizeof(err);
			ret = getsockopt(sd, SOL_SOCKET, SO_ERROR, (void *) &err, (SOCKLEN_T *)&len);
			if (ret == -1) {
				WARN("CallAddrBuff: getsockopt failed\n");
				tmp_errno = errno;
			} else if (err != 0) {
				ret = -1;
				tmp_errno = err;
			} else {
				/* good */
				ret = 0;
			}
		} else  if (ret == 0) {
			/* that was a time out: we handle that later */
			ret = -1;
			tmp_errno = ETIMEDOUT;
		}
	}
#endif

	/* get how long it took */
	start = (MicroTime() - start)/1000000;

	if (timeOut != 0) {
#ifdef USE_ALARM_SIGNAL
		RESETREALTIMER;
#endif
		if (timeOut < 0) {
			/* set the adaptive timeouts */
			if (ret == 0) {
				SetTimeOut(CONN, addr, start, 0, 0);
			} else if (ret == -1 && tmp_errno == ETIMEDOUT) {
				SetTimeOut(CONN, addr, start, 0, 1);
			}
		}
	}


	/* let's see if we got the connection */
	if (ret < 0) {
		close(sd);

		WARN3("CallAddrBuff: connect failed (%d %s) in %.2f seconds\n", tmp_errno, strerror(tmp_errno), start);

		/* no good socket */
		*sock = NO_SOCKET;

		/* failed */
		ret = 0;
	} else {
		/* save the good socket */
		*sock = sd;

		/* print log message */
		peer = IPAddressMachine_r(addr);
		INFO4("CallAddrBuff: connected socket %d to %s:%d in %.2f seconds\n", sd, peer, port, start);
		FREE(peer);

		GetNWSLock(&lock);
		FD_SET(sd, &connectedSockets);
		ReleaseNWSLock(&lock);
	
		/* everything is cool */
		ret = 1;
	}

	return ret;
}


/* it should be thread safe (we lock up access to connected*) */
void
CloseConnections(int closeEars,
                 int closePipes,
                 int closeSockets) {
	Socket dead;
	int i;

	InitModule();

	dead = NO_SOCKET;
	for(i = 0; i < FD_SETSIZE; i++) {
		GetNWSLock(&lock);
		if(closeEars && FD_ISSET(i, &connectedEars)) {
			dead = i;
		}
		if(closePipes && FD_ISSET(i, &connectedPipes)) {
			dead = i;
		}
		if(closeSockets && FD_ISSET(i, &connectedSockets)) {
			dead = i;
		}
		ReleaseNWSLock(&lock);
		DROP_SOCKET(&dead);
	}
}


/*
 * Returns 1 or 0 depending on whether or not #sd# is a connected socket.
 *
 * it should be thread safe.
 */
static int
IsConnected(Socket sd) {
	struct sockaddr peer_name_buff;

	SOCKLEN_T peer_name_buff_size = sizeof(peer_name_buff);
	return(getpeername(sd, &peer_name_buff, &peer_name_buff_size) >= 0);
}


/* thread safe */
int
CloseDisconnections(void) {
	Socket dead, i;
	int returnValue = 0, tmp;

	InitModule();

	for(i = 0; i < FD_SETSIZE; i++) {
		GetNWSLock(&lock);
		tmp = FD_ISSET(i, &connectedSockets);
		ReleaseNWSLock(&lock);
		if(tmp && !IsConnected(i)) {
			dead = i;
			DROP_SOCKET(&dead);
			returnValue++;
		}
	}

	return(returnValue);
}


/* thread safe */
int
CloseSocket(Socket *sock,
            int waitForPeer) {
	fd_set readFDs;
	Socket sd = *sock;
	struct timeval timeout;
	int tmp_errno, gotPipe;

	InitModule();

	/* sanity check */
	if (sock == NULL) {
		ERROR("CloseSocket: passing a NULL argument\n");
		return 0;
	}

	/* check if we need to do work */
	if(*sock <= NO_SOCKET) {
		return 1;  /* Already closed; nothing to do. */
	}

	INFO1("CloseSocket: Closing connection %d\n", *sock);

	GetNWSLock(&lock);
	gotPipe = FD_ISSET(sd, &connectedPipes);
	ReleaseNWSLock(&lock);

#ifdef USE_NONBLOCKING
	if (!gotPipe) {
		/* let's get it synchronous again for closing */
		tmp_errno = fcntl(sd,  F_GETFL, NULL);
		if (tmp_errno < 1) {
			WARN1("CloseSocket: GETFL failed on %d.\n", sd);
		} else {
			tmp_errno &= (~O_NONBLOCK);
			if (fcntl(sd,  F_SETFL, tmp_errno) < 0) {
				WARN1("CloseSocket: SETFL failed on %d\n", sd);
			}
		}
	}
#endif

	if(waitForPeer > 0) {
		FD_ZERO(&readFDs);
		FD_SET(sd, &readFDs);
		timeout.tv_sec = waitForPeer;
		timeout.tv_usec = 0;

		if (PortabilitySelect(sd+1, &readFDs, NULL, NULL, NULL, NULL) <= 0) {
			WARN3("CloseSocket: no response on select sd=%d (%d %s)\n", sd, errno, strerror(errno));
			return 0;
		}
	}

	if(!gotPipe) {
		if(shutdown(sd, 2) < 0) {
			tmp_errno = errno;

			/* The other side may have beaten us to the reset. */
			if ((tmp_errno!=ENOTCONN) && (tmp_errno!=ECONNRESET)) {
				WARN3("CloseSocket: shutdown error (%d %s) on %d\n", tmp_errno, strerror(tmp_errno), sd);
			}
		}
	}

	if(close(sd) < 0) {
		tmp_errno = errno;
		WARN3("CloseSocket: close error on socket %d (%d %s)\n", sd, errno, strerror(errno));
	}

	ClearSocket(sd);
	DoDisconnectNotification(sd);
	*sock = NO_SOCKET;

	return(1);
}


#define READ_END 0
#define WRITE_END 1

int
CreateLocalChild(pid_t *pid,
                 Socket *parentToChild,
                 Socket *childToParent) {

	int childWrite[2];
	int parentWrite[2];
	int myEnd;

	InitModule();

	if(parentToChild != NULL) {
		if(pipe(parentWrite) == -1) {
			FAIL2("CreateLocalChild: couldn't get pipe (%d %s)\n", errno, strerror(errno));
		}
	}

	if(childToParent != NULL) {
		if(pipe(childWrite) == -1) {
			if(parentToChild != NULL) {
				close(parentWrite[0]);
				close(parentWrite[1]);
			}
			FAIL2("CreateLocalChild: couldn't get pipe (%d %s)\n", errno, strerror(errno));
		}
	}

	*pid = fork();

	if(*pid == -1) {
		if(parentToChild != NULL) {
			close(parentWrite[0]);
			close(parentWrite[1]);
		}
		if(childToParent != NULL) {
			close(childWrite[0]);
			close(childWrite[1]);
		}
		FAIL2("CreateLocalChild: couldn't fork (%d %s)\n", errno, strerror(errno));
	}

	/* Close descriptors that this process won't be using. */
	if(parentToChild != NULL) {
		myEnd = (*pid == 0) ? READ_END : WRITE_END;
		close(parentWrite[1 - myEnd]);
		GetNWSLock(&lock);
		FD_SET(parentWrite[myEnd], &connectedPipes);
		ReleaseNWSLock(&lock);
		*parentToChild = parentWrite[myEnd];
	}

	if(childToParent != NULL) {
		myEnd = (*pid == 0) ? WRITE_END : READ_END;
		close(childWrite[1 - myEnd]);
		GetNWSLock(&lock);
		FD_SET(childWrite[myEnd], &connectedPipes);
		ReleaseNWSLock(&lock);
		*childToParent = childWrite[myEnd];
	}

	return(1);
}


static void
SocketFailure(int sig) {
	HoldSignal(SIGPIPE);
	needDisconnect = 1;
	if(signal(SIGPIPE, SocketFailure) == SIG_ERR) {
		WARN("SocketFailure: error resetting signal\n");
	}
	ReleaseSignal(SIGPIPE);
}

/* it should be thread safe */
int
EstablishAnEarBuff(	unsigned short startingPort,
			unsigned short endingPort,
			Socket *ear,
			unsigned short *earPort,
			int bufSize) {

	unsigned short port;
	Socket sd = NO_SOCKET;
	struct sockaddr_in server;
	int tmp_errno;
	SOCKLEN_T size;
	
	InitModule();

	size = sizeof(server);

	for(port = startingPort; port <= endingPort; port++) {
		server.sin_port = htons((u_short)port);
		server.sin_addr.s_addr = INADDR_ANY;
		server.sin_family = AF_INET;
		if((sd = socket(AF_INET, SOCK_STREAM, 0)) < 0) {
			ERROR("EstablishAnEar: socket allocation failed\n");
			return 0;
		}
		/* let's ge the socket in some reasonable default */
		if (bufSize <= 0) {
			bufSize = 32*1024;
		}
		ConditionSocket(sd, 1, bufSize);

		if (bind(sd, (struct sockaddr *)&server, size) == 0) {
			if (listen(sd, 5) == 0) {
				/* success: let's be anal */
				if (getsockname(sd, (struct sockaddr *)&server, &size) ==  0) {

					break;
				} else {
					tmp_errno = errno;
					WARN2("EstablishAnEar: getsockname failed (%d %s)\n", errno, strerror(errno));
				}
			} else {
				tmp_errno = errno;
				WARN2("EstablishAnEar: listen failed (%d %s)\n", errno, strerror(errno));
			}
		} else {
			tmp_errno = errno;
			WARN2("EstablishAnEar: bind failed (%d %s)\n", errno, strerror(errno));
		}
		close(sd);
	}

	if(port > endingPort) {
		FAIL2("EstablishAnEar: couldn't find a port between %d and %d\n", startingPort, endingPort);
	}

	/* let's deal with SIGPIPE */
	signal(SIGPIPE, SocketFailure);

	GetNWSLock(&lock);
	FD_SET(sd, &connectedEars);
	ReleaseNWSLock(&lock);

	if (ear != NULL) {
		*ear = sd;
	}
	if (earPort != NULL) {
		*earPort = ntohs(server.sin_port);
	}

	DDEBUG1("EstablistAnEar: connected socket %d\n", sd);

	return(1);
}

/* the following 2 defines are used internally over the pipe for socket
 * is available */
#define	AvailableMessage "pippo"
#define AvailableMessageLen 5

/* thread safe ... hopefully. We #ifdef HAVE_THREAD_H for performance
 * reasons when threads are not required. */
int
IncomingRequest(double timeOut,
                Socket *sd) {

	Socket dead, i;
	char lookahead;
	Socket newSock, nextToService;
	struct sockaddr_in peer_in;
	SOCKLEN_T peer_in_len = sizeof(peer_in);
	fd_set readFds;
	struct timeval tout, *p_tout;
	double wakeup, now = 0;		/* keep the time */
	int tmp_errno, n, done = -1, recvd, howMany;
#ifdef HAVE_PTHREAD_H
	static short running = 0;
#endif

	InitModule();

	/* sanity check */
	if (sd == NULL) {
		ERROR("IncomingRequest: NULL pointer\n");
		return 0;
	}

	/* nextToService is used to make sure that every connection gets
	 * a chance to be serviced.  If we always started checking at 0,
	 * very active connections with small descriptor numbers could
	 * starve larger-descriptor connections.  */
	/* static Socket nextToService = 0; */
	/* Obi: I don't use the "static" approach since may be trouble
	 * with thread. Instead of locking I use the CurrentTime() to
	 * (hopefully) start from a different place each time (look for
	 * nextToService later) */
	nextToService = -1;

	/* let's check we are the only one running here ... */
#ifdef HAVE_PTHREAD_H
	GetNWSLock(&lock);
	if (running != 0) {
		ReleaseNWSLock(&lock);
		WARN("IncomingRequest: another instance is running!\n");
		return 0;
	}
	running = 1;
	ReleaseNWSLock(&lock);
#endif

	/* let's see if we need to check disconnected socket */
	if (needDisconnect) {
		needDisconnect = 0;
		(void)CloseDisconnections();
	}

	/* we use wakeup to initialize nextService later on, so we
	 * initialize it here even though we don't really use it when
	 * timeOut <= 0 */
	wakeup = MicroTime() + 1000000*timeOut;
	*sd = NO_SOCKET;

	while (done == -1) {
		/* we check the timeout only if we have a timeout to take
		 * care of */
		if (timeOut > 0) {
			now = MicroTime();
			/* Got experience of a -1 from MicroTime so we
			 * need to check here. Plus we check if the
			 * timeout is expired only if requested by the
			 * caller (timeout > 0) */
			if (now == -1 || now >= wakeup) {
				done = 0;	/* didn't find anything */
				break;		/* let's get out of here */
			}
		}

		/* Construct in readFds the union of connected ears,
		 * pipes, and sockets. */
		FD_ZERO(&readFds);
		GetNWSLock(&lock);
		for(n = 0, i = 0; i < FD_SETSIZE; i++) {
			if (FD_ISSET(i, &inUse)) {
				continue;
			}
			if (FD_ISSET(i, &connectedSockets) || FD_ISSET(i, &connectedPipes) || FD_ISSET(i, &connectedEars)) {
				FD_SET(i, &readFds);
				n = i;
			}
		}
		ReleaseNWSLock(&lock);

		/* we cannot have more than FD_SETSIZE sockets */
		if (++n > FD_SETSIZE) {
			WARN("IncomingRequest: too many open sockets!!\n");
			done = 0;
			break;
		}

		if (timeOut > 0) { 
			/* set the timeout */
			tout.tv_sec = (unsigned long)((wakeup - now)/1000000);
			tout.tv_usec = (unsigned long)(wakeup - now) % 1000000;
			p_tout = &tout;
		} else if (timeOut < 0) {
			/* polling only */
			tout.tv_sec = 0;
			tout.tv_usec = 0;
			p_tout = &tout;
			/* we just do one loop around */
			done = 0;
		} else { 
			/* no timeout: good luck! */
			p_tout = NULL;
		}

		howMany = PortabilitySelect(n, &readFds, NULL, NULL, p_tout, &tmp_errno);
		if (howMany == -1) {
			WARN2("IncomingRequest: select error (%d %s)\n", tmp_errno, strerror(tmp_errno));
			done = 0;	/* didn't find anything */
			break;			 /* done here */
		} else if (howMany == 0) {
			continue;	 /* this was a timeout */
		} 

		/* first of all we check if there is something on the
		 * internal pipe, if so we clear it and restart */
		if (FD_ISSET(ReadAvailable, &readFds)) {
			char tmp[AvailableMessageLen + 1];

			RecvBytes(ReadAvailable, tmp, AvailableMessageLen, -1);
			tmp[AvailableMessageLen] = '\0';
			if (strcmp(tmp, AvailableMessage) != 0) {
				WARN("IncomingRequest: something wrong in the internal pipe!\n");
			}
			/* let's restart the whole thing */
			GetNWSLock(&lock);
			written = 0;
			ReleaseNWSLock(&lock);
			continue;
		}

		/* we need to set the nextToService if not already set */
		if (nextToService == -1 || nextToService >= n) {
			nextToService = ((short)wakeup % n);
			if (nextToService < 0) {
				nextToService *= -1;
			}
		}

		/* we do an all around loop to check all sockets
		 * starting from nextToService until we have one
		 * to service */
		for (i = -1; i != nextToService; i = (i+1) % n) {
			short isEar;	/* used to be able to lock global
					   variables */

			if (i == -1) {
				/* first time around */
				i = nextToService;
			}

			if(!FD_ISSET(i, &readFds)) {
				/* nothing to do here */
				continue;
			}

			/* we are in business */
			howMany--;

			GetNWSLock(&lock);
			isEar = FD_ISSET(i, &connectedEars);
			ReleaseNWSLock(&lock);

			if(isEar) {
				/* Add the new connection
				 * to connectedSockets. */
				newSock = accept(i, (struct sockaddr *)&peer_in, &peer_in_len);
				tmp_errno = errno;
				
				if(newSock == -1) {
					newSock = i;
					DROP_SOCKET(&newSock);
					WARN2("IncomingRequest: accept failed (%d %s)\n", tmp_errno, strerror(tmp_errno));
				} else {
					char *peer;

					ConditionSocket(newSock, 0, 32*1024);
					peer = PeerName_r(newSock);
					if (peer != NULL) {
						DDEBUG3("IncomingRequest: connected socket %d to %s:%d\n", newSock, peer, PeerNamePort(newSock));
						FREE(peer);
					} else {
						DDEBUG1("IncomingRequest: connected socket %d\n", newSock);
					}

					/* operating on a global variable */
					GetNWSLock(&lock);
					FD_SET(newSock, &connectedSockets);
					ReleaseNWSLock(&lock);

					/* notify the new connection */
					DoNotifyConnection(i, newSock);
				}
			} else if(IsPipe(i)) {
				/* we found a good one */
				*sd = i;
				done = 1;
				break;
			} else {
				/* Existing socket connection. */
				recvd = recv(i, &lookahead, 1, MSG_PEEK);
				tmp_errno = errno;
				if (recvd < 0) {
#ifdef USE_NONBLOCKING
					/* just in case we are in
					 * progress skip this
					 * socket */
					if (tmp_errno != EAGAIN && tmp_errno != EWOULDBLOCK) {
#endif
					WARN3("IncomingRequest: socket %d failed (%d %s)\n", i, tmp_errno, strerror(tmp_errno));
					/* done with this socket */
					dead = i;
					DROP_SOCKET(&dead);
#ifdef USE_NONBLOCKING
					}
#endif
				} else if (recvd == 0) {
					/* This is how we find out about
					 * connections closed by a peer.
					 * Drop it from our list of known
					 * connections. */
					DDEBUG1("IncomingRequest: Dropping closed connection %d\n", i);
					/* done with this socket */
					dead = i;
					DROP_SOCKET(&dead);
				} else {
					*sd = i;
#ifdef HAVE_PTHREAD_H
					/* the socket is in use:
					 * client needs to call
					 * SocketIsAvailable to
					 * free it */
					/* it only makes sense in
					 * a threaded environment */
					GetNWSLock(&lock);
					FD_SET(i, &inUse);
					ReleaseNWSLock(&lock);
#endif
					done = 1;
					break;
				}
			}
			/* if we have serviced as many sockets as select
			 * told us we are done here */
			if (howMany <= 0) {
				break;
			}
		}
	}

	/* done */
#ifdef HAVE_PTHREAD_H
	GetNWSLock(&lock);
	running = 0;
	ReleaseNWSLock(&lock);
#endif

	return done;
}


/* thread safe */
int 
SocketIsAvailable(Socket sd) {
	int toSignal = 0;

	/* sanity check */
	if (sd < 0 || sd > FD_SETSIZE) {
		return 0;
	}

	GetNWSLock(&lock);
	FD_CLR(sd, &inUse);
	if (written == 0) {
		toSignal = 1;
		written = 1;
	}
	ReleaseNWSLock(&lock);

	/* and now let's notify that we got a new socket to listen to */
	if (WriteAvailable != NO_SOCKET && toSignal) {
		SendBytes(WriteAvailable, AvailableMessage, AvailableMessageLen, -1);
	}

	return 1;
}

/* thread safe */
int
SocketInUse(Socket sd) {

	/* sanity check */
	if (sd < 0 || sd > FD_SETSIZE) {
		return 0;
	}

	InitModule();

	GetNWSLock(&lock);
	/* let's check if we know anything about this socket */
	if (FD_ISSET(sd, &connectedPipes) || FD_ISSET(sd, &connectedSockets) || FD_ISSET(sd, &connectedEars)) {
		FD_SET(sd, &inUse);
	} else {
		WARN1("SocketInUse: %d not a connected socket!\n", sd);
	}
	ReleaseNWSLock(&lock);

	return 1;
}

/* thread safe */
int
IsOkay(Socket sd) {
	fd_set readFds;
	fd_set writeFds;
	struct timeval tv;
	double tmp;

	if(sd < 0) {
		return 0;
	}

	FD_ZERO(&readFds);
	FD_ZERO(&writeFds);
	FD_SET(sd, &readFds);
	FD_SET(sd, &writeFds);
	tmp = GetTimeOut(SEND, Peer(sd), 1);
	tv.tv_sec  = (unsigned long)tmp; 
	tv.tv_usec = (unsigned long)((tmp - (unsigned long)tmp)*1000000);

	return (PortabilitySelect(sd+1, &readFds, &writeFds, NULL, &tv, NULL) > 0);

}


/* thread safe */
int
NotifyOnDisconnection(SocketFunction notifyFn) {
	int i;

	InitModule();

	/* operating on global variables */
	GetNWSLock(&lock);
	for(i = 0; i < MAX_NOTIFIES; i++) {
		if(disconnectFunctions[i] == NULL) {
			disconnectFunctions[i] = notifyFn;
			break;
		}
	}
	ReleaseNWSLock(&lock);
	
	return (i < MAX_NOTIFIES);
}

int 
NotifyOnNewConnection(	NewConnectionFunction newFn) {
	int i;

	InitModule();

	/* operating on global variables */
	GetNWSLock(&lock);
	for(i = 0; i < MAX_NOTIFIES; i++) {
		if(newFunctions[i] == NULL) {
			newFunctions[i] = newFn;
			break;
		}
	}
	ReleaseNWSLock(&lock);
	
	return (i < MAX_NOTIFIES);
}


#define MAXPASSES 40
static pid_t passedPids[MAXPASSES];
static Socket passedSockets[MAXPASSES];


int
PassSocket(Socket *sock,
           pid_t child) {

	int i, childStat;

	/* Clean up any sockets previously passed to children who have
	 * exited. */
	for(i = 0; i < MAXPASSES; i++) {
		if(passedPids[i] != 0) {
			if((waitpid(passedPids[i], &childStat, WNOHANG) < 0) || WIFEXITED(childStat)) {
				INFO1("PassSocket: Reclaiming connection %d\n", passedSockets[i]);
				DROP_SOCKET(&passedSockets[i]);
				passedPids[i] = 0;
			}
		}
	}

	/* Record this socket in passedSockets and remove all other
	 * memory of it. */
	for(i = 0; i < MAXPASSES; i++) {
		if(passedPids[i] == 0) {
			INFO2("PassSocket: Passing connection %d to %d\n", *sock, child);
			passedPids[i] = child;
			passedSockets[i] = *sock;
			ClearSocket(*sock);
			*sock = NO_SOCKET;
			return(1);
		}
	}

	return(0);
}


/* here to be used by dnsutil.c (GetPeerName) */
int
IsPipe(Socket sd) {
	int ret = 0;

	InitModule();

	GetNWSLock(&lock);
	if (FD_ISSET(sd, &connectedPipes)) {
		ret = 1;
	}
	ReleaseNWSLock(&lock);

	return ret;
}
	
int
RecvBytesWait(	Socket sd,
		void *bytes,
		size_t byteSize,
		double timeOut,
		int dontWait) {
	double when, myTimeOut;
	int tmp, tmp_errno;
	char *nextByte;
	fd_set readFds;
	int recvd, totalRecvd;
	struct timeval tout, *tv;

	/* sanity check */
	if (sd < 0 || bytes == NULL || byteSize <= 0) {
		WARN("RecvBytesWait: parameters out of range!\n");
		return 0;
	}

	FD_ZERO(&readFds);
	FD_SET(sd, &readFds);

	/* is there a timeout? */
	if (timeOut > 0) {
		/* let's compute when the time is over */
		when = CurrentTime() + timeOut;

#ifdef USE_ALARM_SIGNAL
		/* set the alarm (we get to the bigger integer to not
		 * interfere with the select that deals with doubles) */
		myTimeOut = timeOut;
		SetRealTimer((myTimeOut >= 1) ? (int)(myTimeOut+1) : 1);
#endif
		/* we'll use the timeout in select */
		tv = &tout;
	} else {
		when = 0;
		tv = NULL;
	}

	totalRecvd = 0;			/* so far no bytes received */
	for(nextByte=(char*)bytes; totalRecvd < byteSize; totalRecvd += recvd) {
		recvd = 0;

		/* set the timeout if requested by the user */
		if (timeOut > 0) {
			myTimeOut = when - CurrentTime();
			if (myTimeOut <= 0) {
				break;
			}
			tout.tv_sec = (unsigned long) myTimeOut;
			tout.tv_usec = (unsigned long)((myTimeOut - tout.tv_sec)*1000000);
		}

		tmp =  select(sd+1, &readFds, NULL, NULL, tv);
		tmp_errno = errno;

		if (tmp == -1) {
			if (tmp_errno == EINTR) {
				/* retry */
				continue;
			}
			WARN3("RecvBytesWait: select on %d failed (%d %s)\n", sd, tmp_errno, strerror(tmp_errno));
			break;
		} else if (tmp == 0) {
			/* timed out */
			WARN2("RecvBytesWait: Socket %d timed out (%.0fs)\n", sd, timeOut);

			break;
		}

		/* read & recv are the same if flags is 0: we use read so
		 * we don't need to check if it's a pipe */
	 	recvd = read(sd, nextByte, byteSize - totalRecvd);
		tmp_errno = errno;

		if(recvd < 0) {
#ifdef USE_NONBLOCKING
			/* socket is not ready */
			if(tmp_errno == EAGAIN || tmp_errno == EWOULDBLOCK)  {
				/* if we've been asked not to wait we get
				 * out of here */
				if (dontWait) {
					break;
				}
				continue;
			}
#endif
			WARN5("RecvBytesWait: socket %d failed after %d/%d bytes (%d %s)\n", sd, totalRecvd, byteSize, tmp_errno, strerror(tmp_errno));
			break;
		} else if (recvd == 0) {
			/* socket got disconnected: bye bye */
			INFO1("RecvBytesWait: socket %d got disconnected\n", sd);
			break;
		}
		/* all is good */
		nextByte += recvd;
	}

#ifdef USE_ALARM_SIGNAL
	if (timeOut > 0) {
		RESETREALTIMER;
	}
#endif
	
	return totalRecvd;
}


/* it should be thread safe. */
int
SendBytesWait(	Socket sd,
		const void *bytes,
		size_t byteSize,
		double timeOut,
		int dontWait) {

	char *nextByte;
	int sent, totalSent, tmp;
	double when, myTimeOut;
	fd_set writeFds;
	int tmp_errno;
	struct timeval *tv, tout;

	/* sanity check */
	if (sd < 0 || bytes == NULL || byteSize <= 0) {
		WARN("SendBytesWait: parameters out of range!\n");
		return 0;
	}

	/* select the adaptive timeouts or the passed in one. */
	if (timeOut > 0) {
		/* let's compute when the time is over */
		when = CurrentTime() + timeOut;
#ifdef USE_ALARM_SIGNAL
		/* set the alarm to the upper second to not get in the
		 * way of the select which deals with doubles */
		myTimeOut = timeOut;
		SetRealTimer((myTimeOut >= 1) ? (int)(myTimeOut + 1) : 1);
#endif
		tv = &tout;
	} else {
		when = 0;
		tv = NULL;
	}

	FD_ZERO(&writeFds);
	FD_SET(sd, &writeFds);

	totalSent = 0;			/* nothing sent yet */
	for(nextByte = (char*)bytes; totalSent < byteSize; totalSent += sent) {
		sent = 0;

		/* set the timeout, and if we timed out get out */
		if (timeOut > 0) {
			/* 0 is the special flag for don't use touts */
			myTimeOut = when - CurrentTime();
			if (myTimeOut <= 0) {
				break;
			}
			tout.tv_sec = (unsigned long) myTimeOut;
			tout.tv_usec = (unsigned long)((myTimeOut - tout.tv_sec)*1000000);
		} 

		tmp =  select(sd+1, NULL, &writeFds, NULL, tv);
		tmp_errno = errno;

		if (tmp == -1) {
			if (tmp_errno == EINTR) {
				/* retry */
				continue;
			}
			WARN3("SendBytesWait: select on %d failed (%d %s)\n", sd, tmp_errno, strerror(tmp_errno));
			break;
		} else if (tmp == 0) {
			/* timed out */
			WARN2("SendBytesWait: Socket %d timed out (%.0fs)\n", sd, timeOut);
			break;
		}

		/* write & send are the same if flags is 0: we use write
		 * so we don't have to check if it's a pipe */
		sent = write(sd, nextByte, byteSize - totalSent);
		tmp = errno;

		if(sent < 0) {
#ifdef USE_NONBLOCKING
			/* the socket is not ready yet */
			if (tmp == EAGAIN || tmp == EWOULDBLOCK)  {
				/* if we've been asked not to wait we get
				 * out of here */
				if (dontWait) {
					break;
				}
				continue;
			}
#endif
			WARN5("SendBytesWait: send on socket %d failed after %d/%d bytes (%d %s)\n", sd, totalSent, byteSize, tmp, strerror(tmp));
			break;
		}
		nextByte += sent;
	}

#ifdef USE_ALARM_SIGNAL
	/* reset sigalalrm */
	if (timeOut > 0) {
		RESETREALTIMER;
	}
#endif

	return totalSent;
}

int
ReflectSocket(	Socket from,
		Socket to) {
#ifdef USE_NONBLOCKING
	fd_set Fds;
	struct timeval tv;
	int tmp, ret, recvd;
	char buf[512];

	ret = 0;

	FD_ZERO(&Fds);
	FD_SET(from, &Fds);

	/* we don't wait here: we just go for what we got */
	tv.tv_usec = 0;
	tv.tv_sec = 0;
	ret = PortabilitySelect(from+1, &Fds, NULL, NULL, &tv, &tmp);
	if (ret == -1) {
		/* error case */
		ERROR2("ReflectSocket: select returned %d on socket %d\n", tmp, from);
		return 0;
	} else if (ret == 0) {
		INFO1("ReflectSocket: socket %d timed out\n", from);
		return 1;
	} 

	/* good to go: read what we can and dump it on
	 * the #to# socket: first we need to check that
	 * #to# is writable. */
	FD_ZERO(&Fds);
	FD_SET(to, &Fds);

	tv.tv_usec = 0;
	tv.tv_sec = 0;

	ret = PortabilitySelect(to+1, NULL, &Fds, NULL, &tv, &tmp);
	if (ret == -1) {
		/* error case */
		ERROR2("ReflectSocket: select returned %d on socket %d\n", tmp, to);
		return 0;
	} else if (ret == 0) {
		INFO1("ReflectSocket: socket %d timed out\n", from);
		return 1;
	}

	/* let's read */
	recvd = read(from, buf, sizeof(buf));
	tmp = errno;

	/* let's check if something bad happened */
	if (recvd < 0) {
		if (tmp == EAGAIN || tmp == EWOULDBLOCK) {
			/* done here for now */
			return 1;
		} else {
			WARN3("ReflectSocket: socket %d failed (%d %s)\n", to, tmp, strerror(tmp));
			return 0;
		}
	}

	/* let's dump it */
	ret = write(to, buf, recvd);
	tmp = errno;

	/* let's check if something bad happened */
	if (ret < 0) {
		if (tmp != EAGAIN && tmp != EWOULDBLOCK) {
			WARN3("ReflectSocket: socket %d failed (%d %s)\n", to, tmp, strerror(tmp));
			return 0;
		}
	}
	
	/* done for now */
	return 1;
#else
	ERROR("ReflectSocket: I need non blocking socket to work!\n");
	return 0;
#endif
}