Subversion Repositories planix.SVN

/* network i/o */

#include "all.h"

#include "io.h"

#include <fcall.h>		/* 9p2000 */

#include <thread.h>

enum {

	Maxfdata	= 8192,

	Nqueue		= 200,		/* queue size (tunable) */

	Netclosed	= 0,		/* Connection state */

	Netopen,

};

/*

 * the kernel file server read packets directly from

 * its ethernet(s) and did all the protocol processing.

 * if the incoming packets were 9p (over il/ip), they

 * were queued for the server processes to operate upon.

 *

 * in user mode, we have one process per incoming connection

 * instead, and those processes get just the data, minus

 * tcp and ip headers, so they just see a stream of 9p messages,

 * which they then queue for the server processes.

 *

 * there used to be more queueing (in the kernel), with separate

 * processes for ethernet input, il input, 9p processing, il output

 * and ethernet output, and queues connecting them.  we now let

 * the kernel's network queues, protocol stacks and processes do

 * much of this work.

 *

 * partly as a result of this, we can now process 9p messages

 * transported via tcp, exploit multiple x86 processors, and

 * were able to shed 70% of the file server's source, by line count.

 *

 * the upshot is that Ether (now Network) is no longer a perfect fit for

 * the way network i/o is done now.  the notion of `connection'

 * is being introduced to complement it.

 */

typedef struct Network Network;

typedef struct Netconn Netconn;

typedef struct Conn9p Conn9p;

/* a network, not necessarily an ethernet */

struct Network {

	int	ctlrno;

	char	iname[NAMELEN];

	char	oname[NAMELEN];

	char	*dialstr;

	char	anndir[40];

	char	lisdir[40];

	int	annfd;			/* fd from announce */

};

/* an open tcp (or other transport) connection */

struct Netconn {

	Queue*	reply;		/* network output */

	char*	raddr;		/* remote caller's addr */

	Chan*	chan;		/* list of tcp channels */

	int	alloc;		/* flag: allocated */

	int	state;

	Conn9p*	conn9p;		/* not reference-counted */

	Lock;

};

/*

 * incoming 9P network connection from a given machine.

 * typically will multiplex 9P sessions for multiple users.

 */

struct Conn9p {

	QLock;

	Ref;

	int	fd;

	char*	dir;

	Netconn*netconn;		/* cross-connection */

	char*	raddr;

};

static Network netif[Maxnets];

static struct {

	Lock;

	Chan*	chan;

} netchans;

static Queue *netoq;		/* only one network output queue is needed */

char *annstrs[Maxnets] = {

	"tcp!*!9fs",

};

/* never returns nil */

static Chan*

getchan(Conn9p *conn9p)

{

	Netconn *netconn;

	Chan *cp, *xcp;

	lock(&netchans);

	/* look for conn9p's Chan */

	xcp = nil;

	for(cp = netchans.chan; cp; cp = netconn->chan) {

		netconn = cp->pdata;

		if(!netconn->alloc)

			xcp = cp;		/* remember free Chan */

		else if(netconn->raddr != nil &&

		    strcmp(conn9p->raddr, netconn->raddr) == 0) {

			unlock(&netchans);

			return cp;		/* found conn9p's Chan */

		}

	}

	/* conn9p's Chan not found; if no free Chan, allocate & fill in one */

	cp = xcp;

	if(cp == nil) {

		cp = fs_chaninit(Devnet, 1, sizeof(Netconn));

		netconn = cp->pdata;

		netconn->chan = netchans.chan;

		netconn->state = Netopen;	/* a guess */

		/* cross-connect netconn and conn9p */

		netconn->conn9p = conn9p;	/* not reference-counted */

		conn9p->netconn = netconn;

		netchans.chan = cp;

	}

	/* fill in Chan's netconn */

	netconn = cp->pdata;

	netconn->raddr = strdup(conn9p->raddr);

	/* fill in Chan */

	cp->send = serveq;

	if (cp->reply == nil)

		cp->reply = netoq;

	netconn->reply = netoq;

	cp->protocol = nil;

	cp->msize = 0;

	cp->whotime = 0;

	strncpy(cp->whochan, conn9p->raddr, sizeof cp->whochan);

//	cp->whoprint = tcpwhoprint;

	netconn->alloc = 1;

	unlock(&netchans);

	return cp;

}

static char *

fd2name(int fd)

{

	char data[128];

	if (fd2path(fd, data, sizeof data) < 0)

		return strdup("/GOK");

	return strdup(data);

}

static void

hangupdfd(int dfd)

{

	int ctlfd;

	char *end, *data;

	data = fd2name(dfd);

	close(dfd);

	end = strstr(data, "/data");

	if (end != nil)

		strcpy(end, "/ctl");

	ctlfd = open(data, OWRITE);

	if (ctlfd >= 0) {

		hangup(ctlfd);

		close(ctlfd);

	}

	free(data);

}

void

closechan(int n)

{

	Chan *cp;

	for(cp = chans; cp; cp = cp->next)

		if(cp->whotime != 0 && cp->chan == n)

			fileinit(cp);

}

void

nethangup(Chan *cp, char *msg, int dolock)

{

	Netconn *netconn;

	netconn = cp->pdata;

	netconn->state = Netclosed;

	if(msg != nil)

		print("hangup! %s %s\n", msg, netconn->raddr);

	fileinit(cp);

	cp->whotime = 0;

	strcpy(cp->whoname, "<none>");

	if(dolock)

		lock(&netchans);

	netconn->alloc = 0;

	free(netconn->raddr);

	netconn->raddr = nil;

	if(dolock)

		unlock(&netchans);

}

void

chanhangup(Chan *cp, char *msg, int dolock)

{

	Netconn *netconn = cp->pdata;

	Conn9p *conn9p = netconn->conn9p;

	if (conn9p->fd > 0)

		hangupdfd(conn9p->fd);	/* drop it */

	nethangup(cp, msg, dolock);

}

/*

 * returns length of next 9p message (including the length) and

 * leaves it in the first few bytes of abuf.

 */

static long

size9pmsg(int fd, void *abuf, uint n)

{

	int m;

	uchar *buf = abuf;

	if (n < BIT32SZ)

		return -1;	/* caller screwed up */

	/* read count */

	m = readn(fd, buf, BIT32SZ);

	if(m != BIT32SZ){

		if(m < 0)

			return -1;

		return 0;

	}

	return GBIT32(buf);

}

static int

readalloc9pmsg(int fd, Msgbuf **mbp)

{

	int m, len;

	uchar lenbuf[BIT32SZ];

	Msgbuf *mb;

	*mbp = nil;

	len = size9pmsg(fd, lenbuf, BIT32SZ);

	if (len <= 0)

		return len;

	if(len <= BIT32SZ || len > IOHDRSZ+Maxfdata){

		werrstr("bad length in 9P2000 message header");

		return -1;

	}

	if ((mb = mballoc(len, nil, Mbeth1)) == nil)

		panic("readalloc9pmsg: mballoc failed");

	*mbp = mb;

	memmove(mb->data, lenbuf, BIT32SZ);

	len -= BIT32SZ;

	m = readn(fd, mb->data+BIT32SZ, len);

	if(m < len)

		return 0;

	return BIT32SZ+m;

}

static void

connection(void *v)

{

	int n;

	char buf[64];

	Chan *chan9p;

	Conn9p *conn9p = v;

	Msgbuf *mb;

	NetConnInfo *nci;

	incref(conn9p);			/* count connections */

	nci = getnetconninfo(conn9p->dir, conn9p->fd);

	if (nci == nil)

		panic("connection: getnetconninfo(%s, %d) failed",

			conn9p->dir, conn9p->fd);

	conn9p->raddr = nci->raddr;

	chan9p = getchan(conn9p);

	print("new connection on %s pid %d from %s\n",

		conn9p->dir, getpid(), conn9p->raddr);

	/*

	 * reading from a pipe or a network device

	 * will give an error after a few eof reads.

	 * however, we cannot tell the difference

	 * between a zero-length read and an interrupt

	 * on the processes writing to us,

	 * so we wait for the error.

	 */

	while (conn9p->fd > 0 && (n = readalloc9pmsg(conn9p->fd, &mb)) >= 0) {

		if(n == 0)

			continue;

		mb->param = (uintptr)conn9p;	/* has fd for replies */

		mb->chan = chan9p;

		assert(mb->magic == Mbmagic);

		incref(conn9p);			/* & count packets in flight */

		fs_send(serveq, mb);		/* to 9P server processes */

		/* mb will be freed by receiving process */

	}

	rerrstr(buf, sizeof buf);

	qlock(conn9p);

	print("connection hung up from %s\n", conn9p->dir);

	if (conn9p->fd > 0)		/* not poisoned yet? */

		hangupdfd(conn9p->fd);	/* poison the fd */

	nethangup(chan9p, "remote hung up", 1);

	closechan(chan9p->chan);

	conn9p->fd = -1;		/* poison conn9p */

	if (decref(conn9p) == 0) {	/* last conn.? turn the lights off */

		free(conn9p->dir);

		qunlock(conn9p);

		free(conn9p);

	} else

		qunlock(conn9p);

	freenetconninfo(nci);

	if(buf[0] == '\0' || strstr(buf, "hungup") != nil)

		exits("");

	sysfatal("mount read, pid %d", getpid());

}

static void

neti(void *v)

{

	int lisfd, accfd;

	Network *net;

	Conn9p *conn9p;

	net = v;

	print("net%di\n", net->ctlrno);

	for(;;) {

		lisfd = listen(net->anndir, net->lisdir);

		if (lisfd < 0) {

			print("listen %s failed: %r\n", net->anndir);

			continue;

		}

		/* got new call on lisfd */

		accfd = accept(lisfd, net->lisdir);

		if (accfd < 0) {

			print("accept %d (from %s) failed: %r\n",

				lisfd, net->lisdir);

			continue;

		}

		/* accepted that call */

		conn9p = malloc(sizeof *conn9p);

		conn9p->dir = strdup(net->lisdir);

		conn9p->fd = accfd;

		newproc(connection, conn9p, smprint("9P read %s", conn9p->dir));

		close(lisfd);

	}

}

/* only need one of these for all network connections, thus all interfaces */

static void

neto(void *)

{

	int len, datafd;

	Msgbuf *mb;

	Conn9p *conn9p;

	print("neto\n");

	for(;;) {

		/* receive 9P answer from 9P server processes */

		while((mb = fs_recv(netoq, 0)) == nil)

			continue;

		if(mb->data == nil) {

			print("neto: pkt nil cat=%d free=%d\n",

				mb->category, mb->flags&FREE);

			if(!(mb->flags & FREE))

				mbfree(mb);

			continue;

		}

		/* send answer back over the network connection in the reply */

		len = mb->count;

		conn9p = (Conn9p *)mb->param;

		assert(conn9p);

		qlock(conn9p);

		datafd = conn9p->fd;

		assert(len >= 0);

		/* datafd < 0 probably indicates poisoning by the read side */

		if (datafd < 0 || write(datafd, mb->data, len) != len) {

			print( "network write error (%r);");

			print(" closing connection for %s\n", conn9p->dir);

			nethangup(getchan(conn9p), "network write error", 1);

			if (datafd > 0)

				hangupdfd(datafd);	/* drop it */

			conn9p->fd = -1;		/* poison conn9p */

		}

		mbfree(mb);

		if (decref(conn9p) == 0)

			panic("neto: zero ref count");

		qunlock(conn9p);

	}

}

void

netstart(void)

{

	int netorun = 0;

	Network *net;

	if(netoq == nil)

		netoq = newqueue(Nqueue, "network reply");

	for(net = &netif[0]; net < &netif[Maxnets]; net++){

		if(net->dialstr == nil)

			continue;

		sprint(net->oname, "neto");

		if (netorun++ == 0)

			newproc(neto, nil, net->oname);

		sprint(net->iname, "net%di", net->ctlrno);

		newproc(neti, net, net->iname);

	}

}

void

netinit(void)

{

	Network *net;

	for (net = netif; net < netif + Maxnets; net++) {

		net->dialstr = annstrs[net - netif];

		if (net->dialstr == nil)

			continue;

		net->annfd = announce(net->dialstr, net->anndir);

		/* /bin/service/tcp564 may already have grabbed the port */

		if (net->annfd < 0)

			sysfatal("can't announce %s: %r", net->dialstr);

		print("netinit: announced on %s\n", net->dialstr);

	}

}