1 056fe1ba 2003-11-23 devnull #include <u.h>
  2 8baa0cbd 2004-06-09 devnull #include <errno.h>
  3 056fe1ba 2003-11-23 devnull #include <libc.h>
  4 056fe1ba 2003-11-23 devnull #include <venti.h>
  5 056fe1ba 2003-11-23 devnull #include "queue.h"
  6 056fe1ba 2003-11-23 devnull 
  7 8baa0cbd 2004-06-09 devnull long ventisendbytes, ventisendpackets;
  8 8baa0cbd 2004-06-09 devnull long ventirecvbytes, ventirecvpackets;
  9 8baa0cbd 2004-06-09 devnull 
 10 056fe1ba 2003-11-23 devnull static int
 11 056fe1ba 2003-11-23 devnull _vtsend(VtConn *z, Packet *p)
 12 056fe1ba 2003-11-23 devnull {
 13 056fe1ba 2003-11-23 devnull 	IOchunk ioc;
 14 056fe1ba 2003-11-23 devnull 	int n;
 15 056fe1ba 2003-11-23 devnull 	uchar buf[2];
 16 8baa0cbd 2004-06-09 devnull 
 17 8baa0cbd 2004-06-09 devnull 
 18 056fe1ba 2003-11-23 devnull 	if(z->state != VtStateConnected) {
 19 056fe1ba 2003-11-23 devnull 		werrstr("session not connected");
 20 056fe1ba 2003-11-23 devnull 		return -1;
 21 056fe1ba 2003-11-23 devnull 	}
 22 056fe1ba 2003-11-23 devnull 
 23 056fe1ba 2003-11-23 devnull 	/* add framing */
 24 056fe1ba 2003-11-23 devnull 	n = packetsize(p);
 25 056fe1ba 2003-11-23 devnull 	if(n >= (1<<16)) {
 26 056fe1ba 2003-11-23 devnull 		werrstr("packet too large");
 27 056fe1ba 2003-11-23 devnull 		packetfree(p);
 28 056fe1ba 2003-11-23 devnull 		return -1;
 29 056fe1ba 2003-11-23 devnull 	}
 30 056fe1ba 2003-11-23 devnull 	buf[0] = n>>8;
 31 056fe1ba 2003-11-23 devnull 	buf[1] = n;
 32 056fe1ba 2003-11-23 devnull 	packetprefix(p, buf, 2);
 33 8baa0cbd 2004-06-09 devnull 	ventisendbytes += n+2;
 34 8baa0cbd 2004-06-09 devnull 	ventisendpackets++;
 35 056fe1ba 2003-11-23 devnull 
 36 056fe1ba 2003-11-23 devnull 	for(;;){
 37 056fe1ba 2003-11-23 devnull 		n = packetfragments(p, &ioc, 1, 0);
 38 056fe1ba 2003-11-23 devnull 		if(n == 0)
 39 056fe1ba 2003-11-23 devnull 			break;
 40 056fe1ba 2003-11-23 devnull 		if(write(z->outfd, ioc.addr, ioc.len) < ioc.len){
 41 056fe1ba 2003-11-23 devnull 			packetfree(p);
 42 056fe1ba 2003-11-23 devnull 			return 0;
 43 056fe1ba 2003-11-23 devnull 		}
 44 056fe1ba 2003-11-23 devnull 		packetconsume(p, nil, ioc.len);
 45 056fe1ba 2003-11-23 devnull 	}
 46 056fe1ba 2003-11-23 devnull 	packetfree(p);
 47 056fe1ba 2003-11-23 devnull 	return 1;
 48 056fe1ba 2003-11-23 devnull }
 49 056fe1ba 2003-11-23 devnull 
 50 056fe1ba 2003-11-23 devnull static Packet*
 51 056fe1ba 2003-11-23 devnull _vtrecv(VtConn *z)
 52 056fe1ba 2003-11-23 devnull {
 53 056fe1ba 2003-11-23 devnull 	uchar buf[10], *b;
 54 056fe1ba 2003-11-23 devnull 	int n;
 55 056fe1ba 2003-11-23 devnull 	Packet *p;
 56 056fe1ba 2003-11-23 devnull 	int size, len;
 57 056fe1ba 2003-11-23 devnull 
 58 056fe1ba 2003-11-23 devnull 	if(z->state != VtStateConnected) {
 59 056fe1ba 2003-11-23 devnull 		werrstr("session not connected");
 60 056fe1ba 2003-11-23 devnull 		return nil;
 61 056fe1ba 2003-11-23 devnull 	}
 62 056fe1ba 2003-11-23 devnull 
 63 056fe1ba 2003-11-23 devnull 	p = z->part;
 64 056fe1ba 2003-11-23 devnull 	/* get enough for head size */
 65 056fe1ba 2003-11-23 devnull 	size = packetsize(p);
 66 056fe1ba 2003-11-23 devnull 	while(size < 2) {
 67 056fe1ba 2003-11-23 devnull 		b = packettrailer(p, MaxFragSize);
 68 056fe1ba 2003-11-23 devnull 		assert(b != nil);
 69 a09e80f9 2004-05-23 devnull 		if(0) fprint(2, "%d read hdr\n", getpid());
 70 056fe1ba 2003-11-23 devnull 		n = read(z->infd, b, MaxFragSize);
 71 a09e80f9 2004-05-23 devnull 		if(0) fprint(2, "%d got %d (%r)\n", getpid(), n);
 72 8baa0cbd 2004-06-09 devnull 		if(n==0 || (n<0 && errno!=EINTR))
 73 056fe1ba 2003-11-23 devnull 			goto Err;
 74 056fe1ba 2003-11-23 devnull 		size += n;
 75 056fe1ba 2003-11-23 devnull 		packettrim(p, 0, size);
 76 056fe1ba 2003-11-23 devnull 	}
 77 056fe1ba 2003-11-23 devnull 
 78 056fe1ba 2003-11-23 devnull 	if(packetconsume(p, buf, 2) < 0)
 79 056fe1ba 2003-11-23 devnull 		goto Err;
 80 056fe1ba 2003-11-23 devnull 	len = (buf[0] << 8) | buf[1];
 81 056fe1ba 2003-11-23 devnull 	size -= 2;
 82 056fe1ba 2003-11-23 devnull 
 83 056fe1ba 2003-11-23 devnull 	while(size < len) {
 84 a09e80f9 2004-05-23 devnull 	//	n = len - size;
 85 a09e80f9 2004-05-23 devnull 	//	if(n > MaxFragSize)
 86 056fe1ba 2003-11-23 devnull 			n = MaxFragSize;
 87 056fe1ba 2003-11-23 devnull 		b = packettrailer(p, n);
 88 a09e80f9 2004-05-23 devnull 		if(0) fprint(2, "%d read body %d\n", getpid(), n);
 89 a09e80f9 2004-05-23 devnull 		n = read(z->infd, b, n);
 90 a09e80f9 2004-05-23 devnull 		if(0) fprint(2, "%d got %d (%r)\n", getpid(), n);
 91 a09e80f9 2004-05-23 devnull 		if(n > 0)
 92 a09e80f9 2004-05-23 devnull 			size += n;
 93 a09e80f9 2004-05-23 devnull 		packettrim(p, 0, size);
 94 8baa0cbd 2004-06-09 devnull 		if(n==0 || (n<0 && errno!=EINTR))
 95 056fe1ba 2003-11-23 devnull 			goto Err;
 96 056fe1ba 2003-11-23 devnull 	}
 97 8baa0cbd 2004-06-09 devnull 	ventirecvbytes += len;
 98 8baa0cbd 2004-06-09 devnull 	ventirecvpackets++;
 99 056fe1ba 2003-11-23 devnull 	p = packetsplit(p, len);
100 056fe1ba 2003-11-23 devnull 	return p;
101 056fe1ba 2003-11-23 devnull Err:	
102 056fe1ba 2003-11-23 devnull 	return nil;	
103 056fe1ba 2003-11-23 devnull }
104 056fe1ba 2003-11-23 devnull 
105 056fe1ba 2003-11-23 devnull /*
106 056fe1ba 2003-11-23 devnull  * If you fork off two procs running vtrecvproc and vtsendproc,
107 056fe1ba 2003-11-23 devnull  * then vtrecv/vtsend (and thus vtrpc) will never block except on 
108 056fe1ba 2003-11-23 devnull  * rendevouses, which is nice when it's running in one thread of many.
109 056fe1ba 2003-11-23 devnull  */
110 056fe1ba 2003-11-23 devnull void
111 056fe1ba 2003-11-23 devnull vtrecvproc(void *v)
112 056fe1ba 2003-11-23 devnull {
113 056fe1ba 2003-11-23 devnull 	Packet *p;
114 056fe1ba 2003-11-23 devnull 	VtConn *z;
115 056fe1ba 2003-11-23 devnull 	Queue *q;
116 056fe1ba 2003-11-23 devnull 
117 056fe1ba 2003-11-23 devnull 	z = v;
118 056fe1ba 2003-11-23 devnull 	q = _vtqalloc();
119 056fe1ba 2003-11-23 devnull 
120 056fe1ba 2003-11-23 devnull 	qlock(&z->lk);
121 056fe1ba 2003-11-23 devnull 	z->readq = q;
122 056fe1ba 2003-11-23 devnull 	qlock(&z->inlk);
123 056fe1ba 2003-11-23 devnull 	rwakeup(&z->rpcfork);
124 056fe1ba 2003-11-23 devnull 	qunlock(&z->lk);
125 056fe1ba 2003-11-23 devnull 
126 056fe1ba 2003-11-23 devnull 	while((p = _vtrecv(z)) != nil)
127 056fe1ba 2003-11-23 devnull 		if(_vtqsend(q, p) < 0){
128 056fe1ba 2003-11-23 devnull 			packetfree(p);
129 056fe1ba 2003-11-23 devnull 			break;
130 056fe1ba 2003-11-23 devnull 		}
131 056fe1ba 2003-11-23 devnull 	qunlock(&z->inlk);
132 056fe1ba 2003-11-23 devnull 	qlock(&z->lk);
133 056fe1ba 2003-11-23 devnull 	_vtqhangup(q);
134 056fe1ba 2003-11-23 devnull 	while((p = _vtnbqrecv(q)) != nil)
135 056fe1ba 2003-11-23 devnull 		packetfree(p);
136 056fe1ba 2003-11-23 devnull 	vtfree(q);
137 056fe1ba 2003-11-23 devnull 	z->readq = nil;
138 056fe1ba 2003-11-23 devnull 	rwakeup(&z->rpcfork);
139 056fe1ba 2003-11-23 devnull 	qunlock(&z->lk);
140 056fe1ba 2003-11-23 devnull 	vthangup(z);
141 056fe1ba 2003-11-23 devnull }
142 056fe1ba 2003-11-23 devnull 
143 056fe1ba 2003-11-23 devnull void
144 056fe1ba 2003-11-23 devnull vtsendproc(void *v)
145 056fe1ba 2003-11-23 devnull {
146 056fe1ba 2003-11-23 devnull 	Queue *q;
147 056fe1ba 2003-11-23 devnull 	Packet *p;
148 056fe1ba 2003-11-23 devnull 	VtConn *z;
149 056fe1ba 2003-11-23 devnull 
150 056fe1ba 2003-11-23 devnull 	z = v;
151 056fe1ba 2003-11-23 devnull 	q = _vtqalloc();
152 056fe1ba 2003-11-23 devnull 
153 056fe1ba 2003-11-23 devnull 	qlock(&z->lk);
154 056fe1ba 2003-11-23 devnull 	z->writeq = q;
155 056fe1ba 2003-11-23 devnull 	qlock(&z->outlk);
156 056fe1ba 2003-11-23 devnull 	rwakeup(&z->rpcfork);
157 056fe1ba 2003-11-23 devnull 	qunlock(&z->lk);
158 056fe1ba 2003-11-23 devnull 
159 056fe1ba 2003-11-23 devnull 	while((p = _vtqrecv(q)) != nil)
160 056fe1ba 2003-11-23 devnull 		if(_vtsend(z, p) < 0)
161 056fe1ba 2003-11-23 devnull 			break;
162 056fe1ba 2003-11-23 devnull 	qunlock(&z->outlk);
163 056fe1ba 2003-11-23 devnull 	qlock(&z->lk);
164 056fe1ba 2003-11-23 devnull 	_vtqhangup(q);
165 056fe1ba 2003-11-23 devnull 	while((p = _vtnbqrecv(q)) != nil)
166 056fe1ba 2003-11-23 devnull 		packetfree(p);
167 056fe1ba 2003-11-23 devnull 	vtfree(q);
168 056fe1ba 2003-11-23 devnull 	z->writeq = nil;
169 056fe1ba 2003-11-23 devnull 	rwakeup(&z->rpcfork);
170 056fe1ba 2003-11-23 devnull 	qunlock(&z->lk);
171 056fe1ba 2003-11-23 devnull 	return;
172 056fe1ba 2003-11-23 devnull }
173 056fe1ba 2003-11-23 devnull 
174 056fe1ba 2003-11-23 devnull Packet*
175 056fe1ba 2003-11-23 devnull vtrecv(VtConn *z)
176 056fe1ba 2003-11-23 devnull {
177 056fe1ba 2003-11-23 devnull 	Packet *p;
178 056fe1ba 2003-11-23 devnull 
179 056fe1ba 2003-11-23 devnull 	qlock(&z->lk);
180 056fe1ba 2003-11-23 devnull 	if(z->state != VtStateConnected){
181 056fe1ba 2003-11-23 devnull 		werrstr("not connected");
182 056fe1ba 2003-11-23 devnull 		qunlock(&z->lk);
183 056fe1ba 2003-11-23 devnull 		return nil;
184 056fe1ba 2003-11-23 devnull 	}
185 056fe1ba 2003-11-23 devnull 	if(z->readq){
186 056fe1ba 2003-11-23 devnull 		qunlock(&z->lk);
187 056fe1ba 2003-11-23 devnull 		return _vtqrecv(z->readq);
188 056fe1ba 2003-11-23 devnull 	}
189 056fe1ba 2003-11-23 devnull 
190 056fe1ba 2003-11-23 devnull 	qlock(&z->inlk);
191 056fe1ba 2003-11-23 devnull 	qunlock(&z->lk);
192 056fe1ba 2003-11-23 devnull 	p = _vtrecv(z);
193 056fe1ba 2003-11-23 devnull 	qunlock(&z->inlk);
194 056fe1ba 2003-11-23 devnull 	if(!p)
195 056fe1ba 2003-11-23 devnull 		vthangup(z);
196 056fe1ba 2003-11-23 devnull 	return p;
197 056fe1ba 2003-11-23 devnull }
198 056fe1ba 2003-11-23 devnull 
199 056fe1ba 2003-11-23 devnull int
200 056fe1ba 2003-11-23 devnull vtsend(VtConn *z, Packet *p)
201 056fe1ba 2003-11-23 devnull {
202 056fe1ba 2003-11-23 devnull 	qlock(&z->lk);
203 056fe1ba 2003-11-23 devnull 	if(z->state != VtStateConnected){
204 056fe1ba 2003-11-23 devnull 		packetfree(p);
205 056fe1ba 2003-11-23 devnull 		werrstr("not connected");
206 056fe1ba 2003-11-23 devnull 		qunlock(&z->lk);
207 056fe1ba 2003-11-23 devnull 		return -1;
208 056fe1ba 2003-11-23 devnull 	}
209 056fe1ba 2003-11-23 devnull 	if(z->writeq){
210 056fe1ba 2003-11-23 devnull 		qunlock(&z->lk);
211 056fe1ba 2003-11-23 devnull 		if(_vtqsend(z->writeq, p) < 0){
212 056fe1ba 2003-11-23 devnull 			packetfree(p);
213 056fe1ba 2003-11-23 devnull 			return -1;
214 056fe1ba 2003-11-23 devnull 		}
215 056fe1ba 2003-11-23 devnull 		return 0;
216 056fe1ba 2003-11-23 devnull 	}
217 056fe1ba 2003-11-23 devnull 
218 056fe1ba 2003-11-23 devnull 	qlock(&z->outlk);
219 056fe1ba 2003-11-23 devnull 	qunlock(&z->lk);
220 056fe1ba 2003-11-23 devnull 	if(_vtsend(z, p) < 0){
221 056fe1ba 2003-11-23 devnull 		qunlock(&z->outlk);
222 056fe1ba 2003-11-23 devnull 		vthangup(z);
223 056fe1ba 2003-11-23 devnull 		return -1;	
224 056fe1ba 2003-11-23 devnull 	}
225 056fe1ba 2003-11-23 devnull 	qunlock(&z->outlk);
226 056fe1ba 2003-11-23 devnull 	return 0;
227 056fe1ba 2003-11-23 devnull }
228 056fe1ba 2003-11-23 devnull