harvey/sys/src/cmd/aan.c

/*
 * This file is part of the UCB release of Plan 9. It is subject to the license
 * terms in the LICENSE file found in the top-level directory of this
 * distribution and at http://akaros.cs.berkeley.edu/files/Plan9License. No
 * part of the UCB release of Plan 9, including this file, may be copied,
 * modified, propagated, or distributed except according to the terms contained
 * in the LICENSE file.
 */

#include <u.h>
#include <libc.h>
#include <auth.h>
#include <fcall.h>
#include <thread.h>

#define NS(x)		((int64_t)x)
#define US(x)		(NS(x) * 1000LL)
#define MS(x)		(US(x) * 1000LL)
#define S(x)		(MS(x) * 1000LL)

#define LOGNAME	"aan"

enum {
	Synctime = S(8),
	Nbuf = 10,
	K = 1024,
	Bufsize = 8 * K,
	Stacksize = 8 * K,
	Timer = 0,		/* Alt channels. */
	Unsent = 1,
	Maxto = 24 * 3600,	/* A full day to reconnect. */
	Hdrsz = 12,
};

typedef struct Endpoints Endpoints;
struct Endpoints {
	char	*lsys;
	char	*lserv;
	char	*rsys;
	char	*rserv;
};

typedef struct {
	uint32_t	nb;		/* Number of data bytes in this message */
	uint32_t	msg;		/* Message number */
	uint32_t	acked;		/* Number of messages acked */
} Hdr;

typedef struct {
	Hdr	hdr;
	uint8_t	buf[Bufsize];
} Buf;

static char	*Logname = LOGNAME;
static int	client;
static int	debug;
static char	*devdir;
static char	*dialstring;
static int	done;
static int	inmsg;
static int	maxto = Maxto;
static int	netfd;

static Channel	*empty;
static Channel	*unacked;
static Channel	*unsent;

static Alt a[] = {
	/*	c	v	 op   */
	{ 	nil,	nil,	CHANRCV	},	/* timer */
	{	nil,	nil,	CHANRCV	},	/* unsent */
	{ 	nil,	nil,	CHANEND	},
};

static void	dmessage(int, char *, ...);
static void	freeendpoints(Endpoints *);
static void	fromclient(void*);
static void	fromnet(void*);
static Endpoints *getendpoints(char *);
static void	packhdr(Hdr *, uint8_t *);
static void	reconnect(void);
static void	showmsg(int, char *, Buf *);
static void	synchronize(void);
static void	timerproc(void *);
static void	unpackhdr(Hdr *, uint8_t *);
static int	writen(int, uint8_t *, int);

static void
usage(void)
{
	fprint(2, "Usage: %s [-cd] [-m maxto] dialstring|netdir\n", argv0);
	threadexitsall("usage");
}

static int
catch(void *v, char *s)
{
	if (strstr(s, "alarm") != nil) {
		syslog(0, Logname, "Timed out waiting for client on %s, exiting...",
			   devdir);
		threadexitsall(nil);
	}
	return 0;
}

void
threadmain(int argc, char **argv)
{
	int i, fd, failed, delta;
	int64_t synctime, now;
	char *p;
	unsigned char buf[Hdrsz];
	Buf *b, *eb;
	Channel *timer;
	Hdr hdr;

	ARGBEGIN {
	case 'c':
		client++;
		break;
	case 'd':
		debug++;
		break;
	case 'm':
		maxto = strtol(EARGF(usage()), (char **)nil, 0);
		break;
	default:
		usage();
	} ARGEND;

	if (argc != 1)
		usage();

	if (!client) {
		devdir = argv[0];
		if ((p = strstr(devdir, "/local")) != nil)
			*p = '\0';
	}else
		dialstring = argv[0];

	if (debug > 0) {
		fd = open("#c/cons", OWRITE|OCEXEC);
		dup(fd, 2);
	}

	fmtinstall('F', fcallfmt);

	atnotify(catch, 1);

	unsent = chancreate(sizeof(Buf *), Nbuf);
	unacked= chancreate(sizeof(Buf *), Nbuf);
	empty  = chancreate(sizeof(Buf *), Nbuf);
	timer  = chancreate(sizeof(unsigned char *), 1);

	for (i = 0; i != Nbuf; i++) {
		eb = malloc(sizeof(Buf));
		sendp(empty, eb);
	}

	netfd = -1;

	if (proccreate(fromnet, nil, Stacksize) < 0)
		sysfatal("Cannot start fromnet; %r");

	reconnect();		/* Set up the initial connection. */
	synchronize();

	if (proccreate(fromclient, nil, Stacksize) < 0)
		sysfatal("cannot start fromclient; %r");

	if (proccreate(timerproc, timer, Stacksize) < 0)
		sysfatal("Cannot start timerproc; %r");

	a[Timer].c = timer;
	a[Unsent].c = unsent;
	a[Unsent].v = &b;

	synctime = nsec() + Synctime;
	failed = 0;
	while (!done) {
		if (failed) {
			/* Wait for the netreader to die. */
			while (netfd >= 0) {
				dmessage(1, "main; waiting for netreader to die\n");
				sleep(1000);
			}

			/* the reader died; reestablish the world. */
			reconnect();
			synchronize();
			failed = 0;
		}

		now = nsec();
		delta = (synctime - nsec()) / MS(1);

		if (delta <= 0) {
			hdr.nb = 0;
			hdr.acked = inmsg;
			hdr.msg = -1;
			packhdr(&hdr, buf);
			if (writen(netfd, buf, sizeof(buf)) < 0) {
				dmessage(2, "main; writen failed; %r\n");
				failed = 1;
				continue;
			}
			synctime = nsec() + Synctime;
			assert(synctime > now);
		}

		switch (alt(a)) {
		case Timer:
			break;
		case Unsent:
			sendp(unacked, b);

			b->hdr.acked = inmsg;
			packhdr(&b->hdr, buf);
			if (writen(netfd, buf, sizeof(buf)) < 0 ||
			    writen(netfd, b->buf, b->hdr.nb) < 0) {
				dmessage(2, "main; writen failed; %r\n");
				failed = 1;
			}

			if (b->hdr.nb == 0)
				done = 1;
			break;
		}
	}
	syslog(0, Logname, "exiting...");
	threadexitsall(nil);
}

static void
fromclient(void *v)
{
	Buf *b;
	static int outmsg;

	do {
		b = recvp(empty);
		if ((int)(b->hdr.nb = read(0, b->buf, Bufsize)) <= 0) {
			if ((int)b->hdr.nb < 0)
				dmessage(2, "fromclient; Cannot read 9P message; %r\n");
			else
				dmessage(2, "fromclient; Client terminated\n");
			b->hdr.nb = 0;
		}
		b->hdr.msg = outmsg++;

		showmsg(1, "fromclient", b);
		sendp(unsent, b);
	} while (b->hdr.nb != 0);
}

static void
fromnet(void *v)
{
	int len, acked, i;
	uint8_t buf[Hdrsz];
	Buf *b, *rb;
	static int lastacked;

	b = (Buf *)malloc(sizeof(Buf));
	assert(b);

	while (!done) {
		while (netfd < 0) {
			dmessage(1, "fromnet; waiting for connection... (inmsg %d)\n",
					  inmsg);
			sleep(1000);
		}

		/* Read the header. */
		if ((len = readn(netfd, buf, sizeof(buf))) <= 0) {
			if (len < 0)
				dmessage(1, "fromnet; (hdr) network failure; %r\n");
			else
				dmessage(1, "fromnet; (hdr) network closed\n");
			close(netfd);
			netfd = -1;
			continue;
		}
		unpackhdr(&b->hdr, buf);
		dmessage(2, "fromnet: Got message, size %d, nb %d, msg %d\n",
			len, b->hdr.nb, b->hdr.msg);

		if (b->hdr.nb == 0) {
			if  ((int32_t)b->hdr.msg >= 0) {
				dmessage(1, "fromnet; network closed\n");
				break;
			}
			continue;
		}

		if ((len = readn(netfd, b->buf, b->hdr.nb)) <= 0 ||
		    len != b->hdr.nb) {
			if (len == 0)
				dmessage(1, "fromnet; network closed\n");
			else
				dmessage(1, "fromnet; network failure; %r\n");
			close(netfd);
			netfd = -1;
			continue;
		}

		if (b->hdr.msg < inmsg) {
			dmessage(1, "fromnet; skipping message %d, currently at %d\n",
				b->hdr.msg, inmsg);
			continue;
		}

		/* Process the acked list. */
		acked = b->hdr.acked - lastacked;
		for (i = 0; i != acked; i++) {
			rb = recvp(unacked);
			if (rb->hdr.msg != lastacked + i) {
				dmessage(1, "rb %p, msg %d, lastacked %d, i %d\n",
					rb, rb? rb->hdr.msg: -2, lastacked, i);
				assert(0);
			}
			rb->hdr.msg = -1;
			sendp(empty, rb);
		}
		lastacked = b->hdr.acked;
		inmsg++;
		showmsg(1, "fromnet", b);
		if (writen(1, b->buf, len) < 0)
			sysfatal("fromnet; cannot write to client; %r");
	}
	done = 1;
}

static void
reconnect(void)
{
	char err[32], ldir[40];
	int lcfd, fd;
	Endpoints *ep;

	if (dialstring) {
		syslog(0, Logname, "dialing %s", dialstring);
  		while ((fd = dial(dialstring, nil, nil, nil)) < 0) {
			err[0] = '\0';
			errstr(err, sizeof err);
			if (strstr(err, "connection refused")) {
				dmessage(1, "reconnect; server died...\n");
				threadexitsall("server died...");
			}
			dmessage(1, "reconnect: dialed %s; %s\n", dialstring, err);
			sleep(1000);
		}
		syslog(0, Logname, "reconnected to %s", dialstring);
	} else {
		syslog(0, Logname, "waiting for connection on %s", devdir);
		alarm(maxto * 1000);
 		if ((lcfd = listen(devdir, ldir)) < 0)
			sysfatal("reconnect; cannot listen; %r");

		if ((fd = accept(lcfd, ldir)) < 0)
			sysfatal("reconnect; cannot accept; %r");
		alarm(0);
		close(lcfd);

		ep = getendpoints(ldir);
		dmessage(1, "rsys '%s'\n", ep->rsys);
		syslog(0, Logname, "connected from %s", ep->rsys);
		freeendpoints(ep);
	}
	netfd = fd;			/* Wakes up the netreader. */
}

static void
synchronize(void)
{
	Channel *tmp;
	Buf *b;
	uint8_t buf[Hdrsz];

	/*
	 * Ignore network errors here.  If we fail during
	 * synchronization, the next alarm will pick up
	 * the error.
	 */
	tmp = chancreate(sizeof(Buf *), Nbuf);
	while ((b = nbrecvp(unacked)) != nil) {
		packhdr(&b->hdr, buf);
		writen(netfd, buf, sizeof(buf));
		writen(netfd, b->buf, b->hdr.nb);
		sendp(tmp, b);
	}
	chanfree(unacked);
	unacked = tmp;
}

static void
showmsg(int level, char *s, Buf *b)
{
	if (b == nil) {
		dmessage(level, "%s; b == nil\n", s);
		return;
	}
	dmessage(level, "%s;  (len %d) %X %X %X %X %X %X %X %X %X (%p)\n", s,
		b->hdr.nb,
		b->buf[0], b->buf[1], b->buf[2],
		b->buf[3], b->buf[4], b->buf[5],
		b->buf[6], b->buf[7], b->buf[8], b);
}

static int
writen(int fd, uint8_t *buf, int nb)
{
	int n, len = nb;

	while (nb > 0) {
		if (fd < 0)
			return -1;
		if ((n = write(fd, buf, nb)) < 0) {
			dmessage(1, "writen; Write failed; %r\n");
			return -1;
		}
		dmessage(2, "writen: wrote %d bytes\n", n);
		buf += n;
		nb -= n;
	}
	return len;
}

static void
timerproc(void *x)
{
	Channel *timer = x;

	while (!done) {
		sleep((Synctime / MS(1)) >> 1);
		sendp(timer, "timer");
	}
}

static void
dmessage(int level, char *fmt, ...)
{
	va_list arg;

	if (level > debug)
		return;
	va_start(arg, fmt);
	vfprint(2, fmt, arg);
	va_end(arg);
}

static void
getendpoint(char *dir, char *file, char **sysp, char **servp)
{
	int fd, n;
	char buf[128];
	char *sys, *serv;

	sys = serv = 0;
	snprint(buf, sizeof buf, "%s/%s", dir, file);
	fd = open(buf, OREAD);
	if(fd >= 0){
		n = read(fd, buf, sizeof(buf)-1);
		if(n>0){
			buf[n-1] = 0;
			serv = strchr(buf, '!');
			if(serv){
				*serv++ = 0;
				serv = strdup(serv);
			}
			sys = strdup(buf);
		}
		close(fd);
	}
	if(serv == 0)
		serv = strdup("unknown");
	if(sys == 0)
		sys = strdup("unknown");
	*servp = serv;
	*sysp = sys;
}

static Endpoints *
getendpoints(char *dir)
{
	Endpoints *ep;

	ep = malloc(sizeof(*ep));
	getendpoint(dir, "local", &ep->lsys, &ep->lserv);
	getendpoint(dir, "remote", &ep->rsys, &ep->rserv);
	return ep;
}

static void
freeendpoints(Endpoints *ep)
{
	free(ep->lsys);
	free(ep->rsys);
	free(ep->lserv);
	free(ep->rserv);
	free(ep);
}

/* p must be a unsigned char* */
#define	U32GET(p)	(p[0] | p[1]<<8 | p[2]<<16 | p[3]<<24)
#define	U32PUT(p,v)	(p)[0] = (v); (p)[1] = (v)>>8; \
			(p)[2] = (v)>>16; (p)[3] = (v)>>24

static void
packhdr(Hdr *hdr, uint8_t *buf)
{
	uint8_t *p;

	p = buf;
	U32PUT(p, hdr->nb);
	p += 4;
	U32PUT(p, hdr->msg);
	p += 4;
	U32PUT(p, hdr->acked);
}

static void
unpackhdr(Hdr *hdr, uint8_t *buf)
{
	uint8_t *p;

	p = buf;
	hdr->nb = U32GET(p);
	p += 4;
	hdr->msg = U32GET(p);
	p += 4;
	hdr->acked = U32GET(p);
}