/*  * 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. */ /* 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 voidhangupdfd(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);} voidclosechan(int n){	Chan *cp; 	for(cp = chans; cp; cp = cp->next)		if(cp->whotime != 0 && cp->chan == n)			fileinit(cp);} voidnethangup(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);} voidchanhangup(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 longsize9pmsg(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 intreadalloc9pmsg(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 voidconnection(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 voidneti(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 voidneto(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);	}} voidnetstart(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);	}} voidnetinit(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);	}}