/*======================================================================
                    C O M M . C 
                    doc: Thu Mar  5 12:32:02 1992
                    dlm: Thu Aug 19 20:50:21 1993
                    (c) 1992 ant@julia
                    uE-Info: 68 70 T 0 11 72 2 2 8 ofnI
======================================================================*/

/*#define		VERBOSE		/* show retrys */

#include	<stdio.h>
#include	<signal.h>
#include	<netdb.h>	/* prefer system over rpc/netdb.h */
#include	<errno.h>
#include	<sys/time.h>
#include	<sys/types.h>
#include	<sys/socket.h>
#include	<sys/param.h>
#include	<rpc/rpc.h>
#include	<arpa/inet.h>
#include	<netinet/in.h>
#include	"inetray.h"
#include	"inetray.start.h"
#include	"rcfile.h"
#include	"config.h"
#include	"common.h"
#include	"utils.h"
#include	"version.h"

hInfo		*hosts = NULL;		/* list of hosts */
int		sfdmax = 0;		/* max fd for ressocks */
int		nRunning = 0;		/* number of workers running */

static int	sock;			/* listening socket */

extern int	errno;			/* to restart syscalls */
extern int	key;			/* session key */

void shutDown(sig)			/* shut down operation */
int sig;
{
	int i,res;
	void killAll(),flushAll(),waitAll(),terminateAll(),closeAll();

	for (i=1; i<NSIG; i++)			/* ignore all signals */
		signal(i,SIG_IGN);
	fprintf(stderr,"\nShutting down");
	if (sig != 0)
		fprintf(stderr," due to Signal #%d",sig);
	fprintf(stderr,"... [kill"); 
	killAll();				/* kill workers */
	fprintf(stderr,", flush");
	flushAll();				/* flush result sockets */
	fprintf(stderr,", wait");
	waitAll();				/* wait for workers */
	fprintf(stderr,", terminate");
	terminateAll();				/* terminate servers */
	fprintf(stderr,", close");
	closeAll();				/* close sockets */
	fprintf(stderr,"]\n");
	exit(0);
}

void abort1(host)			/* this one is broken */
hInfo *host;
{
	fprintf(stderr,
		"\tWarning: Lost connection with %s(%d) (check remote error messages)\n",
	        host->name,host->wid);
	host->broken = TRUE;		/* this host crashed */
	close(host->sock);
	if (--nRunning == 0) {		/* still servers running? */
	        fprintf(stderr,"We're alone now...");
	        shutDown(0);
	}
}

void IRbroadcast(fNum,inP,in)		/* bcast INIT or TERMINATE */
u_long fNum; char *in; xdrproc_t inP;
{
	int 	i,res,exists,uid,nSvcs;
	char	*addr;

	exists = getHost(TRUE,&addr,&nSvcs,&uid);	/* hosts */
	while (exists) {
		if (fNum == INIT) {			/* INIT is special */
			((iPrm *)in)->uid = uid;
			((iPrm *)in)->nSvcs = nSvcs;
			if (nSvcs > 1) nSvcs = 1;
		}
		for (i=0; i<nSvcs; i++) {
			res = sendrpc(addr,INETRAY+i,IRV1,fNum,
					inP,in,xdr_void,NULL);
			if ((res != 0) &&
			    ((fNum != TERMINATE) || (res != RPC_PROGNOTREGISTERED))) {
				fprintf(stderr,
					"\nWarning: sendrpc(%s,%d): ",
					addr,fNum);
				clnt_perrno(res);
				addr[0] = '\0';		/* skip in future */
			}
		}
		exists = getHost(FALSE,&addr,&nSvcs,&uid);
	}

	exists = getNet(TRUE,&addr,&nSvcs,&uid);	/* nets */
	while (exists) {
		if (fNum == INIT) {
			((iPrm *)in)->uid = uid;
			((iPrm *)in)->nSvcs = nSvcs;
			if (nSvcs > 1) nSvcs = 1;
		}
		for (i=0; i<nSvcs; i++) {
			res = clnt_dirbcast(addr,INETRAY+i,IRV1,fNum,
						inP,in,xdr_void,NULL,NULL);
			if (res != 0) {
				fprintf(stderr,
					"\nWarning: clnt_dirbcast(%s,%d): ",
					addr,fNum);
				clnt_perrno(res);
				addr[0] = '\0';		/* skip in future */
			}
		}
		exists = getNet(FALSE,&addr,&nSvcs,&uid);
	}

	getLocal(&addr,&nSvcs,&uid);			/* LAN */
        if (fNum == INIT) {
                ((iPrm *)in)->uid = uid;
		((iPrm *)in)->nSvcs = nSvcs;
		if (nSvcs > 1) nSvcs = 1;
	}
        for (i=0; i<nSvcs; i++) {
	        res = clnt_broadcast(INETRAY+i,IRV1,fNum,
	                                inP,in,xdr_void,NULL,NULL);
	        if (res != 0) {
	                fprintf(stderr,
	                        "\nWarning: clnt_broadcast(%d): ",fNum);
	                clnt_perrno(res);
		}
	}
}

void IRSbroadcast()					/* bcast START */
{
	int 	res,exists,nSvcs,uid;
	char	*addr;

	exists = getHost(TRUE,&addr,&nSvcs,&uid);	/* hosts */
	while (exists) {
		if (nSvcs > 0) {
	                res = sendrpc(addr,STARTER,IRSV1,START,
	                              xdr_void,NULL,xdr_void,NULL);
	                if ((res != 0) && (res != RPC_PROGNOTREGISTERED)) {
	                        fprintf(stderr,
	                                "\nWarning: sendrpc(%s,START): ",addr);
	                        clnt_perrno(res);
				addr[0] = '\0';		/* skip in future */
			}
		}
		exists = getHost(FALSE,&addr,&nSvcs,&uid);
	}

	exists = getNet(TRUE,&addr,&nSvcs,&uid);	/* nets */
	while (exists) {
		if (nSvcs > 0) {
	                res = clnt_dirbcast(addr,STARTER,IRSV1,START,
	                                    xdr_void,NULL,xdr_void,NULL,NULL);
	                if (res != 0) {
	                        fprintf(stderr,
	                                "\nWarning: clnt_dirbcast(%s,START): ",addr);
	                        clnt_perrno(res);
			}
		}
		exists = getNet(FALSE,&addr,&nSvcs,&uid);
	}

	getLocal(&addr,&nSvcs,&uid);			/* LAN */
	if (nSvcs > 0) {
	        res = clnt_broadcast(STARTER,IRSV1,START,
	                             xdr_void,NULL,xdr_void,NULL,NULL);
	        if (res != 0) {
	                fprintf(stderr,"\nWarning: clnt_broadcast(START): ");
	                clnt_perrno(res);
		}
	}
}

static readit(fdp,buf,nbyte)			/* xdr aux routine */
char *fdp,*buf; int nbyte;
{
	int nread,fd;

	fd = *(int *)fdp;
reStartRead:
	nread = read(fd,buf,nbyte);
	if (nread < 0) {
		if (errno == EINTR) {
			goto reStartRead;
		}
		perror("read");
	}
	if (nread == 0) nread = -1;
	return nread;
}

static void getAnswers()				/* wait for answers */
{
	int	aLen,res;
	hInfo	*new,*old,*cur;
	fd_set	sockSet;
	char	buf[64],*cp;
	struct hostent 		*host;
	struct servent 		*s;
	struct sockaddr_in	addr;
	struct timeval		tout,noDefault;

	tout.tv_sec = rTimeout;				/* timeout */
	tout.tv_usec = 0;
	noDefault.tv_sec = noDefault.tv_usec = -1;	/* use supplied val */
	do {
	        do {
		restartSelect:
		        FD_ZERO(&sockSet);		/* select() on socket */
		        FD_SET(sock,&sockSet);
	                res = select(sock+1,&sockSet,NULL,NULL,&tout);
	                if (res < 0) {
	       			if (errno == EINTR) goto restartSelect;
	                        perror("select");
	                        exit(1);
			}
		} while ((res == 0) && (nRunning < minWorkers));
		if (res == 0) break;			/* done */
		
		new = (hInfo *)malloc(sizeof(hInfo)); 	/* alloc node */
		if (new == NULL) {
			fprintf(stderr,"malloc() failed\n");
			exit(1);
		}

		new->reTrys = 0;
	       	aLen = sizeof(addr);			/* accept result */
		new->sock = accept(sock,&addr,&aLen);
		if (new->sock < 0) {
			perror("accept");
			exit(1);
		}
		sprintf(buf,"<%s>",inet_ntoa(addr.sin_addr));
		fprintf(stderr,"\n  %-18.17s",buf);
		if (new->sock > sfdmax) sfdmax = new->sock;

		xdrrec_create(&(new->xdrs),0,0,		/* make xdr_stuff */
				(char *)&(new->sock),readit,NULL);
		new->xdrs.x_op = XDR_DECODE;
	        if (!xdrrec_skiprecord(&(new->xdrs))) {	/* read next record*/
	                fprintf(stderr,"xdrrec_skiprecord() failed\n");
	                exit(1);
		}
                if (!xdr_int(&(new->xdrs),&(new->wid))) {/* read info */
			fprintf(stderr,"xdr_int() failed\n");
			exit(1);
		}
                if (!xdr_int(&(new->xdrs),&(new->pid))) {
			fprintf(stderr,"xdr_int() failed\n");
			exit(1);
		}
		cp = new->rUser;
                if (!xdr_string(&(new->xdrs),&cp,32)) {
			fprintf(stderr,"xdr_string() failed\n");
			exit(1);
		}
		cp = new->rwd;
                if (!xdr_string(&(new->xdrs),&cp,MAXPATHLEN)) {
			fprintf(stderr,"xdr_string() failed\n");
			exit(1);
		}
		cp = new->rVersion;
                if (!xdr_string(&(new->xdrs),&cp,8)) {
			fprintf(stderr,"xdr_string() failed\n");
			exit(1);
		}
		
		host = gethostbyaddr(&addr.sin_addr,	/* create client */
			sizeof(addr.sin_addr),AF_INET);
		if (host == NULL) {
			perror("gethostbyaddr");
			exit(1);
		}
		strncpy(new->name,host->h_name,MAXHOSTNAMELEN);
		memcpy((char *)&new->addr,
			(char *)&addr.sin_addr,
			sizeof(addr.sin_addr));
		new->clnt = clnt_create(new->name,
				INETRAY+new->wid,IRV1,"udp");
		new->done = 0;
		if (new->clnt == NULL) {
			fprintf(stderr,"Warning: clnt_create(%s,%d)",new->name,new->wid);
			clnt_pcreateerror("");
			new->done = -1;
		} else if (!clnt_control(new->clnt,CLSET_TIMEOUT,&noDefault)) {
			fprintf(stderr,"clnt_control() failed\n");
			exit(1);
		}

		for (old = NULL, cur = hosts;		/* sort into list */
		     (cur != NULL) &&
		     (cur->wid > new->wid);
		     old = cur, cur = cur->next) ;
		new->next = cur;			/* link it in list */
		if (old == NULL) hosts = new;
		else		 old->next = new;

		sprintf(buf,"[%s]",new->rVersion);	/* version # */
		fprintf(stderr,"%-8.7s",buf);

		fprintf(stderr,"%-14.13s",new->name);	/* name */
		fprintf(stderr,"%5d ",new->pid);	/* pid */

		if (!doThis(new->name) ||		/* drop it */
	            (nRunning == maxWorkers)) {		
			new->done = -1;			/* sentinel */
			fprintf(stderr,"IGNORED");
		} else if (new->rVersion[0] != VERSION[0]) {
			new->done = -1;			/* sentinel */
			fprintf(stderr,"VERSION MISMATCH");
		} else if (new->done == 0) {
			new->broken = FALSE;
			nRunning++;			/* found one */
			fprintf(stderr,"%-10.9s%s",new->rUser,new->rwd);
		}
	} FOREVER;

	for (old = NULL, cur = hosts;			/* drop from list */
	     cur != NULL;
	     cur = cur->next) {
		if (cur->done == 0) {
			old = cur;
			continue;
		}
		if (cur->clnt != NULL) {		/* create failed */
			res = clnt_call(cur->clnt,TERMINATE,
						xdr_int,&key,
						xdr_void,NULL,
						now); 
			if (res != 5) {
				fprintf(stderr,"clnt_call(%s)",cur->name);
				clnt_perrno(res);
				exit(1);
			}
		}
		if (old == NULL) hosts = cur->next;	/* drop from list */
		else 		 old->next = cur->next;
	}
}

void registerSvc(hName,cLine,cwd,sIn)	/* register servers */
char *hName,*cLine,*cwd,sIn;		/* params for INIT */
{
	int	i;
	iPrm	param;
	char	bindErr = FALSE;
	struct sockaddr_in addr;

	param.rName = hName;		/* compose parameters */
	param.cmdLine = cLine;
	stripHead(STRIPHEAD,cwd);
	stripHome(geteuid(),cwd);
	param.cwd = cwd;
	param.key = key;
	param.rPort = portNumber;
	param.sendIn = sIn;
	
	/* listen on socket */
	if ((sock = socket(PF_INET,SOCK_STREAM,0)) < 0) {
		perror("socket");
		exit(1);
	}
	addr.sin_family = AF_INET;
	addr.sin_port = htons(portNumber);
	addr.sin_addr.s_addr = INADDR_ANY;
    restartBind:
	if (bind(sock,&addr,sizeof(addr)) < 0) {
		if (errno == EADDRINUSE) {
			if (!bindErr) {
				fprintf(stderr,"\nAddress in use! Retrying: .");
				bindErr = TRUE;
			} else  fprintf(stderr,".");
			sleep(5);
			goto restartBind;
		}
		perror("bind");
		exit(1);
	}
	if (listen(sock,5) < 0) {
		perror("listen");
		exit(1);
	}

	fprintf(stderr,"start"); 
	IRSbroadcast();				/* start servers */
	fprintf(stderr,", delay"); 
	sleep(rTimeout);			/* allow startup */
	fprintf(stderr,", init"); 
	IRbroadcast(INIT,xdr_iPrm,&param);	/* start all workers */
	fprintf(stderr,"]"); 
	getAnswers();				/* wait for replys */
	close(sock);
}

void distribute(bp)			/* distributed buffered stdin */
int bp;
{
	hInfo	*host;
	char	buf[STRLEN];
	int	nRead;

	do {
		nRead = read(bp,buf,STRLEN);
		if (nRead < 0) {
			perror("read");
			exit(1);
		}
		if (nRead == 0) {
			buf[0] = '\0';
			nRead = 1;
		}
		for (host=hosts;host!=NULL;host=host->next) {
			if (write(host->sock,buf,nRead) != nRead) {
				perror("write");
				exit(1);
			}
		}
	} while (buf[nRead-1] != '\0');
}

void closeAll()				/* close all sockets */
{
	hInfo 	*host;
	
	for (host = hosts; host != NULL; host = host->next) {
		if (host->broken) continue;
		xdr_destroy(&(host->xdrs));
		clnt_destroy(host->clnt);
		close(host->sock);
	}
}

void flushAll()				/* flush all sockets */
{
	hInfo 	*host;
	fd_set 	sockSet;
	char	buf[1024];
	int	res;
	struct timeval tout;

	tout.tv_sec = FLUSHTIMEOUT;
	tout.tv_usec = 0;
	do {
	restartSelect:
	        FD_ZERO(&sockSet);
		for (host = hosts; host != NULL; host = host->next)
			if (!host->broken)
				FD_SET(host->sock,&sockSet);
	        res = select(sfdmax+1,&sockSet,NULL,NULL,&tout);
       		if (res < 0) {
       			if (errno == EINTR) goto restartSelect;
	                perror("select");
			exit(1);
		}
		if (res == 0) return;	/* done */
		for (host = hosts; host != NULL; host = host->next) {
			if (host->broken) continue;
			if (FD_ISSET(host->sock,&sockSet)) {
				if (read(host->sock,buf,1024) < 0) {
					perror("read");
					exit(1);
				}
			}
		}
        } FOREVER;
}

void killAll()				/* kill all servers */
{
	hInfo 	*host;
	int	res,nTrys,rpcerr;

	for (host = hosts; host != NULL; host = host->next) {
		if (host->broken) continue;
		for (nTrys=CALLRETRY; nTrys > 0; nTrys--) {
			rpcerr = clnt_call(host->clnt,KILL,
					   xdr_int,&key,
					   xdr_int,&res,
					   callTOut);
			if (rpcerr == RPC_SUCCESS) break;
			if (rpcerr == RPC_TIMEDOUT) {
#ifdef VERBOSE
				fprintf(stderr,
					"KILL retry to %s(%d)\n",
					host->name,host->wid);
#endif
				host->reTrys++;
				if (nTrys > 1) continue;
			}
			fprintf(stderr,"kill(): ");
			abort1(host);
		}
	}
}

void waitAll()				/* wait on all clients */
{
	hInfo 	*host;
	int	res,nTrys,rpcerr;

	for (host = hosts; host != NULL; host = host->next) {
		if (host->broken) continue;
		for (nTrys=CALLRETRY; nTrys > 0; nTrys--) {
			rpcerr = clnt_call(host->clnt,WAIT,
					   xdr_int,&key,
					   xdr_int,&res,
					   callTOut);
			if (rpcerr == RPC_SUCCESS) break;
			if (rpcerr == RPC_TIMEDOUT) {
#ifdef VERBOSE
				fprintf(stderr,
					"WAIT retry to %s(%d)\n",
					host->name,host->wid);
#endif
				host->reTrys++;
				if (nTrys > 1) continue;
			}
			fprintf(stderr,"wait(): ");
			abort1(host);
		}
	}
}

void terminateAll()			/* terminate all svcs */
{
	hInfo 	*host;
	fd_set 	sockSet;
	int	res,n,time=0;

	IRbroadcast(TERMINATE,xdr_int,&key);	/* terminate all */
	sleep(1); time++;

	while (time < TERMTIMEOUT) {
	restartSelect:
	        FD_ZERO(&sockSet);
        	for (n=0,host=hosts; host!=NULL; host=host->next) {
        		if (host->broken) continue;
			FD_SET(host->sock,&sockSet);
			n++;
		}
	        res = select(sfdmax+1,&sockSet,NULL,NULL,&now);
       		if (res < 0) {
       			if (errno == EINTR) goto restartSelect;
	                perror("select");
			exit(1);
		}
		if (n == res) return;		/* done */
		for (host = hosts; host != NULL; host = host->next) {
			host->broken = FD_ISSET(host->sock,&sockSet);
			if (!host->broken) {
#ifdef VERBOSE
				fprintf(stderr,
					"TERM retry to %s(%d)\n",
					host->name,host->wid);
#endif
				host->reTrys++;
				(void)clnt_call(host->clnt,TERMINATE,
						xdr_int,&key,
						xdr_void,NULL,now);
			}
		}
		sleep(1); time++;
        }
        fprintf(stderr,"\nWarning: The following servers may not have terminated:\n");
        for (host = hosts; host != NULL; host = host->next) {
                if (!FD_ISSET(host->sock,&sockSet)) 
                        fprintf(stderr,"\t%s[%d]\n",host->name,host->pid);
        }
}