![[BACK]](/cvsweb/icons/back.gif){kind=icon}
Return to sturp.c CVS log ![[TXT]](/cvsweb/icons/text.gif){kind=icon}
![[DIR]](/cvsweb/icons/dir.gif){kind=icon}
Up to [Plan 9 NeXT] / lucent / sys / src / 9 / port|
Return to sturp.c CVS log | Up to [Plan 9 NeXT] / lucent / sys / src / 9 / port |
1.1 root 1: #include "u.h"
2: #include "../port/lib.h"
3: #include "mem.h"
4: #include "dat.h"
5: #include "fns.h"
6: #include "io.h"
7: #include "../port/error.h"
8:
9: enum {
10: MSrexmit= 1000,
11: Nmask= 0x7,
12: };
13:
14: #define DPRINT if(q->flag&QDEBUG)kprint
15:
16: typedef struct Urp Urp;
17:
18: #define NOW (MACHP(0)->ticks*MS2HZ)
19:
20: /*
21: * URP status
22: */
23: struct urpstat {
24: ulong input; /* bytes read from urp */
25: ulong output; /* bytes output to urp */
26: ulong rexmit; /* retransmit rejected urp msg */
27: ulong rjtrs; /* reject, trailer size */
28: ulong rjpks; /* reject, packet size */
29: ulong rjseq; /* reject, sequence number */
30: ulong levelb; /* unknown level b */
31: ulong enqsx; /* enqs sent */
32: ulong enqsr; /* enqs rcved */
33: } urpstat;
34:
35: struct Urp {
36: QLock;
37: Urp *list; /* list of all urp structures */
38: short state; /* flags */
39: Rendez r; /* process waiting for output to finish */
40:
41: /* input */
42: QLock ack; /* ack lock */
43: Queue *rq; /* input queue */
44: uchar iseq; /* last good input sequence number */
45: uchar lastecho; /* last echo/rej sent */
46: uchar trbuf[3]; /* trailer being collected */
47: short trx; /* # bytes in trailer being collected */
48: int blocks;
49:
50: /* output */
51: QLock xmit; /* output lock, only one process at a time */
52: Queue *wq; /* output queue */
53: int maxout; /* maximum outstanding unacked blocks */
54: int maxblock; /* max block size */
55: int next; /* next block to send */
56: int unechoed; /* first unechoed block */
57: int unacked; /* first unacked block */
58: int nxb; /* next xb to use */
59: Block *xb[8]; /* the xmit window buffer */
60: QLock xl[8];
61: ulong timer; /* timeout for xmit */
62: int rexmit;
63: };
64:
65: /* list of allocated urp structures (never freed) */
66: struct
67: {
68: Lock;
69: Urp *urp;
70: } urpalloc;
71:
72: Rendez urpkr;
73: QLock urpkl;
74: int urpkstarted;
75:
76: #define WINDOW(u) ((u)->unechoed>(u)->next ? (u)->unechoed+(u)->maxout-(u)->next-8 :\
77: (u)->unechoed+(u)->maxout-(u)->next)
78: #define IN(x, f, n) (f<=n ? (x>=f && x<n) : (x<n || x>=f))
79: #define NEXT(x) (((x)+1)&Nmask)
80:
81: /*
82: * Protocol control bytes
83: */
84: #define SEQ 0010 /* sequence number, ends trailers */
85: #undef ECHO
86: #define ECHO 0020 /* echos, data given to next queue */
87: #define REJ 0030 /* rejections, transmission error */
88: #define ACK 0040 /* acknowledgments */
89: #define BOT 0050 /* beginning of trailer */
90: #define BOTM 0051 /* beginning of trailer, more data follows */
91: #define BOTS 0052 /* seq update algorithm on this trailer */
92: #define SOU 0053 /* start of unsequenced trailer */
93: #define EOU 0054 /* end of unsequenced trailer */
94: #define ENQ 0055 /* xmitter requests flow/error status */
95: #define CHECK 0056 /* xmitter requests error status */
96: #define INITREQ 0057 /* request initialization */
97: #define INIT0 0060 /* disable trailer processing */
98: #define INIT1 0061 /* enable trailer procesing */
99: #define AINIT 0062 /* response to INIT0/INIT1 */
100: #undef DELAY
101: #define DELAY 0100 /* real-time printing delay */
102: #define BREAK 0110 /* Send/receive break (new style) */
103:
104: #define REJECTING 0x1
105: #define INITING 0x2
106: #define HUNGUP 0x4
107: #define OPEN 0x8
108: #define CLOSING 0x10
109:
110: /*
111: * predeclared
112: */
113: static void urpreset(void);
114: static void urpciput(Queue*, Block*);
115: static void urpiput(Queue*, Block*);
116: static void urpoput(Queue*, Block*);
117: static void urpopen(Queue*, Stream*);
118: static void urpclose(Queue *);
119: static void output(Urp*);
120: static void sendblock(Urp*, int);
121: static void rcvack(Urp*, int);
122: static void flushinput(Urp*);
123: static void sendctl(Urp*, int);
124: static void sendack(Urp*);
125: static void sendrej(Urp*);
126: static void initoutput(Urp*, int);
127: static void initinput(Urp*);
128: static void urpkproc(void *arg);
129: static void urpvomit(char*, Urp*);
130: static void tryoutput(Urp*);
131:
132: Qinfo urpinfo =
133: {
134: urpciput,
135: urpoput,
136: urpopen,
137: urpclose,
138: "urp",
139: urpreset
140: };
141:
142: void
143: sturplink(void)
144: {
145: newqinfo(&urpinfo);
146: }
147:
148: static void
149: urpreset(void)
150: {
151: }
152:
153: static void
154: urpopen(Queue *q, Stream *s)
155: {
156: Urp *up;
157:
158: USED(s);
159: if(!urpkstarted){
160: qlock(&urpkl);
161: if(!urpkstarted){
162: urpkstarted = 1;
163: kproc("urpkproc", urpkproc, 0);
164: }
165: qunlock(&urpkl);
166: }
167:
168: /*
169: * find an unused urp structure
170: */
171: for(up = urpalloc.urp; up; up = up->list){
172: if(up->state == 0){
173: qlock(up);
174: if(up->state == 0)
175: break;
176: qunlock(up);
177: }
178: }
179: if(up == 0){
180: /*
181: * none available, create a new one, they are never freed
182: */
183: up = smalloc(sizeof(Urp));
184: qlock(up);
185: lock(&urpalloc);
186: up->list = urpalloc.urp;
187: urpalloc.urp = up;
188: unlock(&urpalloc);
189: }
190: q->ptr = q->other->ptr = up;
191: q->rp = &urpkr;
192: up->rq = q;
193: up->wq = q->other;
194: up->state = OPEN;
195: qunlock(up);
196: initinput(up);
197: initoutput(up, 0);
198: }
199:
200: /*
201: * Shut down the connection and kill off the kernel process
202: */
203: static int
204: isflushed(void *a)
205: {
206: Urp *up;
207:
208: up = (Urp *)a;
209: return (up->state&HUNGUP) || (up->unechoed==up->nxb && up->wq->len==0);
210: }
211: static void
212: urpclose(Queue *q)
213: {
214: Urp *up;
215: int i;
216:
217: up = (Urp *)q->ptr;
218: if(up == 0)
219: return;
220:
221: /*
222: * wait for all outstanding messages to drain, tell kernel
223: * process we're closing.
224: *
225: * if 2 minutes elapse, give it up
226: */
227: up->state |= CLOSING;
228: if(!waserror()){
229: tsleep(&up->r, isflushed, up, 2*60*1000);
230: poperror();
231: }
232: up->state |= HUNGUP;
233:
234: qlock(&up->xmit);
235: /*
236: * ack all outstanding messages
237: */
238: i = up->next - 1;
239: if(i < 0)
240: i = 7;
241: rcvack(up, ECHO+i);
242:
243: /*
244: * free all staged but unsent messages
245: */
246: for(i = 0; i < 7; i++){
247: qlock(&up->xl[i]);
248: if(up->xb[i]){
249: freeb(up->xb[i]);
250: up->xb[i] = 0;
251: }
252: qunlock(&up->xl[i]);
253: }
254: qunlock(&up->xmit);
255:
256: qlock(up);
257: up->state = 0;
258: qunlock(up);
259: }
260:
261: /*
262: * upstream control messages
263: */
264: static void
265: urpctliput(Urp *up, Queue *q, Block *bp)
266: {
267: switch(bp->type){
268: case M_HANGUP:
269: up->state |= HUNGUP;
270: wakeup(&up->r);
271: break;
272: }
273: PUTNEXT(q, bp);
274: }
275:
276: /*
277: * character mode input.
278: *
279: * the first byte in every message is a ctl byte (which belongs at the end).
280: */
281: void
282: urpciput(Queue *q, Block *bp)
283: {
284: Urp *up;
285: int i;
286: int ctl;
287:
288: up = (Urp *)q->ptr;
289: if(up == 0)
290: return;
291: if(bp->type != M_DATA){
292: urpctliput(up, q, bp);
293: return;
294: }
295:
296: /*
297: * get the control character
298: */
299: ctl = *bp->rptr++;
300: if(ctl < 0)
301: return;
302:
303: /*
304: * take care of any data
305: */
306: if(BLEN(bp)>0 && q->next->len<2*Streamhi && q->next->nb<2*Streambhi){
307: bp->flags |= S_DELIM;
308: urpstat.input += BLEN(bp);
309: PUTNEXT(q, bp);
310: } else
311: freeb(bp);
312:
313: /*
314: * handle the control character
315: */
316: switch(ctl){
317: case 0:
318: break;
319: case ENQ:
320: DPRINT("rENQ(c)\n");
321: urpstat.enqsr++;
322: sendctl(up, up->lastecho);
323: sendctl(up, ACK+up->iseq);
324: break;
325:
326: case CHECK:
327: DPRINT("rCHECK(c)\n");
328: sendctl(up, ACK+up->iseq);
329: break;
330:
331: case AINIT:
332: DPRINT("rAINIT(c)\n");
333: up->state &= ~INITING;
334: flushinput(up);
335: tryoutput(up);
336: break;
337:
338: case INIT0:
339: case INIT1:
340: DPRINT("rINIT%d(c)\n", ctl-INIT0);
341: sendctl(up, AINIT);
342: if(ctl == INIT1)
343: q->put = urpiput;
344: initinput(up);
345: break;
346:
347: case INITREQ:
348: DPRINT("rINITREQ(c)\n");
349: initoutput(up, 0);
350: break;
351:
352: case BREAK:
353: break;
354:
355: case REJ+0: case REJ+1: case REJ+2: case REJ+3:
356: case REJ+4: case REJ+5: case REJ+6: case REJ+7:
357: DPRINT("rREJ%d(c)\n", ctl-REJ);
358: rcvack(up, ctl);
359: break;
360:
361: case ACK+0: case ACK+1: case ACK+2: case ACK+3:
362: case ACK+4: case ACK+5: case ACK+6: case ACK+7:
363: case ECHO+0: case ECHO+1: case ECHO+2: case ECHO+3:
364: case ECHO+4: case ECHO+5: case ECHO+6: case ECHO+7:
365: DPRINT("%s%d(c)\n", (ctl&ECHO)?"rECHO":"rACK", ctl&7);
366: rcvack(up, ctl);
367: break;
368:
369: case SEQ+0: case SEQ+1: case SEQ+2: case SEQ+3:
370: case SEQ+4: case SEQ+5: case SEQ+6: case SEQ+7:
371: DPRINT("rSEQ%d(c)\n", ctl-SEQ);
372: qlock(&up->ack);
373: i = ctl & Nmask;
374: if(!QFULL(q->next))
375: sendctl(up, up->lastecho = ECHO+i);
376: up->iseq = i;
377: qunlock(&up->ack);
378: break;
379: }
380: }
381:
382: /*
383: * block mode input.
384: *
385: * the first byte in every message is a ctl byte (which belongs at the end).
386: *
387: * Simplifying assumption: one put == one message && the control byte
388: * is in the first block. If this isn't true, strange bytes will be
389: * used as control bytes.
390: *
391: * There's no input lock. The channel could be closed while we're
392: * processing a message.
393: */
394: void
395: urpiput(Queue *q, Block *bp)
396: {
397: Urp *up;
398: int i, len;
399: int ctl;
400:
401: up = (Urp *)q->ptr;
402: if(up == 0)
403: return;
404: if(bp->type != M_DATA){
405: urpctliput(up, q, bp);
406: return;
407: }
408:
409: /*
410: * get the control character
411: */
412: ctl = *bp->rptr++;
413:
414: /*
415: * take care of any block count(trx)
416: */
417: while(up->trx){
418: if(BLEN(bp)<=0)
419: break;
420: switch (up->trx) {
421: case 1:
422: case 2:
423: up->trbuf[up->trx++] = *bp->rptr++;
424: continue;
425: default:
426: up->trx = 0;
427: break;
428: }
429: }
430:
431: /*
432: * queue the block(s)
433: */
434: if(BLEN(bp) > 0){
435: bp->flags &= ~S_DELIM;
436: putq(q, bp);
437: if(q->len > 4*1024){
438: flushinput(up);
439: return;
440: }
441: } else
442: freeb(bp);
443:
444: /*
445: * handle the control character
446: */
447: switch(ctl){
448: case 0:
449: break;
450: case ENQ:
451: DPRINT("rENQ %d %uo %uo\n", up->blocks, up->lastecho, ACK+up->iseq);
452: up->blocks = 0;
453: urpstat.enqsr++;
454: sendctl(up, up->lastecho);
455: sendctl(up, ACK+up->iseq);
456: flushinput(up);
457: break;
458:
459: case CHECK:
460: DPRINT("rCHECK\n");
461: sendctl(up, ACK+up->iseq);
462: break;
463:
464: case AINIT:
465: DPRINT("rAINIT\n");
466: up->state &= ~INITING;
467: flushinput(up);
468: tryoutput(up);
469: break;
470:
471: case INIT0:
472: case INIT1:
473: DPRINT("rINIT%d\n", ctl-INIT0);
474: sendctl(up, AINIT);
475: if(ctl == INIT0)
476: q->put = urpciput;
477: initinput(up);
478: break;
479:
480: case INITREQ:
481: DPRINT("rINITREQ\n");
482: initoutput(up, 0);
483: break;
484:
485: case BREAK:
486: break;
487:
488: case BOT:
489: case BOTM:
490: case BOTS:
491: DPRINT("rBOT%c...", " MS"[ctl-BOT]);
492: up->trx = 1;
493: up->trbuf[0] = ctl;
494: break;
495:
496: case REJ+0: case REJ+1: case REJ+2: case REJ+3:
497: case REJ+4: case REJ+5: case REJ+6: case REJ+7:
498: DPRINT("rREJ%d\n", ctl-REJ);
499: rcvack(up, ctl);
500: break;
501:
502: case ACK+0: case ACK+1: case ACK+2: case ACK+3:
503: case ACK+4: case ACK+5: case ACK+6: case ACK+7:
504: case ECHO+0: case ECHO+1: case ECHO+2: case ECHO+3:
505: case ECHO+4: case ECHO+5: case ECHO+6: case ECHO+7:
506: DPRINT("%s%d\n", (ctl&ECHO)?"rECHO":"rACK", ctl&7);
507: rcvack(up, ctl);
508: break;
509:
510: /*
511: * if the sequence number is the next expected
512: * and the trailer length == 3
513: * and the block count matches the bytes received
514: * then send the bytes upstream.
515: */
516: case SEQ+0: case SEQ+1: case SEQ+2: case SEQ+3:
517: case SEQ+4: case SEQ+5: case SEQ+6: case SEQ+7:
518: len = up->trbuf[1] + (up->trbuf[2]<<8);
519: DPRINT("rSEQ%d(%d,%d,%d)...", ctl-SEQ, up->trx, len, q->len);
520: i = ctl & Nmask;
521: if(up->trx != 3){
522: urpstat.rjtrs++;
523: sendrej(up);
524: break;
525: }else if(q->len != len){
526: urpstat.rjpks++;
527: sendrej(up);
528: break;
529: }else if(i != ((up->iseq+1)&Nmask)){
530: urpstat.rjseq++;
531: sendrej(up);
532: break;
533: }else if(q->next->len > (3*Streamhi)/2
534: || q->next->nb > (3*Streambhi)/2){
535: DPRINT("next->len=%d, next->nb=%d\n",
536: q->next->len, q->next->nb);
537: flushinput(up);
538: break;
539: }
540: DPRINT("accept %d\n", q->len);
541:
542: /*
543: * send data upstream
544: */
545: if(q->first) {
546: if(up->trbuf[0] != BOTM)
547: q->last->flags |= S_DELIM;
548: while(bp = getq(q)){
549: urpstat.input += BLEN(bp);
550: PUTNEXT(q, bp);
551: }
552: } else {
553: bp = allocb(0);
554: if(up->trbuf[0] != BOTM)
555: bp->flags |= S_DELIM;
556: PUTNEXT(q, bp);
557: }
558: up->trx = 0;
559:
560: /*
561: * acknowledge receipt
562: */
563: qlock(&up->ack);
564: up->iseq = i;
565: if(!QFULL(q->next))
566: sendctl(up, up->lastecho = ECHO|i);
567: qunlock(&up->ack);
568: break;
569: }
570: }
571:
572: /*
573: * downstream control
574: */
575: Queue *trapq;
576: static void
577: urpctloput(Urp *up, Queue *q, Block *bp)
578: {
579: char *fields[2];
580: int outwin;
581:
582: switch(bp->type){
583: case M_CTL:
584: if(streamparse("break", bp)){
585: /*
586: * send a break as part of the data stream
587: */
588: urpstat.output++;
589: bp->wptr = bp->lim;
590: bp->rptr = bp->wptr - 1;
591: *bp->rptr = BREAK;
592: putq(q, bp);
593: output(up);
594: return;
595: }
596: if(streamparse("init", bp)){
597: outwin = strtoul((char*)bp->rptr, 0, 0);
598: initoutput(up, outwin);
599: freeb(bp);
600: return;
601: }
602: if(streamparse("debug", bp)){
603: switch(getfields((char *)bp->rptr, fields, 2, " ")){
604: case 1:
605: if (strcmp(fields[0], "on") == 0) {
606: q->flag |= QDEBUG;
607: q->other->flag |= QDEBUG;
608: }
609: if (strcmp(fields[0], "off") == 0) {
610: q->flag &= ~QDEBUG;
611: q->other->flag &= ~QDEBUG;
612: }
613: }
614: freeb(bp);
615: return;
616: }
617: if(streamparse("trap", bp)){
618: trapq = q;
619: return;
620: }
621: }
622: PUTNEXT(q, bp);
623: }
624:
625: /*
626: * accept data from a writer
627: */
628: static void
629: urpoput(Queue *q, Block *bp)
630: {
631: Urp *up;
632:
633: up = (Urp *)q->ptr;
634:
635: if(bp->type != M_DATA){
636: urpctloput(up, q, bp);
637: return;
638: }
639:
640: urpstat.output += BLEN(bp);
641: putq(q, bp);
642: output(up);
643: }
644:
645: /*
646: * start output
647: */
648: static void
649: output(Urp *up)
650: {
651: Block *bp, *nbp;
652: ulong now;
653: Queue *q;
654: int i;
655:
656: if(!canqlock(&up->xmit))
657: return;
658:
659: if(waserror()){
660: print("urp output error\n");
661: qunlock(&up->xmit);
662: nexterror();
663: }
664:
665: /*
666: * if still initing and it's time to rexmit, send an INIT1
667: */
668: now = NOW;
669: if(up->state & INITING){
670: if(now > up->timer){
671: q = up->wq;
672: DPRINT("INITING timer (%d, %d): ", now, up->timer);
673: sendctl(up, INIT1);
674: up->timer = now + MSrexmit;
675: }
676: goto out;
677: }
678:
679: /*
680: * fill the transmit buffers, `nxb' can never overtake `unechoed'
681: */
682: q = up->wq;
683: i = NEXT(up->nxb);
684: if(i != up->unechoed) {
685: for(bp = getq(q); bp && i!=up->unechoed; i = NEXT(i)){
686: if(up->xb[up->nxb] != 0)
687: urpvomit("output", up);
688: if(BLEN(bp) > up->maxblock){
689: nbp = up->xb[up->nxb] = allocb(0);
690: nbp->rptr = bp->rptr;
691: nbp->wptr = bp->rptr = bp->rptr + up->maxblock;
692: } else {
693: up->xb[up->nxb] = bp;
694: bp = getq(q);
695: }
696: up->nxb = i;
697: }
698: if(bp)
699: putbq(q, bp);
700: }
701:
702: /*
703: * retransmit cruft
704: */
705: if(up->rexmit){
706: /*
707: * if a retransmit is requested, move next back to
708: * the unacked blocks
709: */
710: urpstat.rexmit++;
711: up->rexmit = 0;
712: up->next = up->unacked;
713: } else if(up->unechoed!=up->next && NOW>up->timer){
714: /*
715: * if a retransmit time has elapsed since a transmit,
716: * send an ENQ
717: */
718: DPRINT("OUTPUT timer (%d, %d): ", NOW, up->timer);
719: up->timer = NOW + MSrexmit;
720: up->state &= ~REJECTING;
721: urpstat.enqsx++;
722: sendctl(up, ENQ);
723: goto out;
724: }
725:
726: /*
727: * if there's a window open, push some blocks out
728: *
729: * the lock is to synchronize with acknowledges that free
730: * blocks.
731: */
732: while(WINDOW(up)>0 && up->next!=up->nxb){
733: i = up->next;
734: qlock(&up->xl[i]);
735: if(waserror()){
736: qunlock(&up->xl[i]);
737: nexterror();
738: }
739: sendblock(up, i);
740: qunlock(&up->xl[i]);
741: up->next = NEXT(up->next);
742: poperror();
743: }
744: out:
745: qunlock(&up->xmit);
746: poperror();
747: }
748:
749: /*
750: * try output, this is called by an input process
751: */
752: void
753: tryoutput(Urp *up)
754: {
755: if(!waserror()){
756: output(up);
757: poperror();
758: }
759: }
760:
761: /*
762: * send a control byte, put the byte at the end of the allocated
763: * space in case a lower layer needs header room.
764: */
765: static void
766: sendctl(Urp *up, int ctl)
767: {
768: Block *bp;
769: Queue *q;
770:
771: q = up->wq;
772: if(QFULL(q->next))
773: return;
774: bp = allocb(1);
775: bp->wptr = bp->lim;
776: bp->rptr = bp->lim-1;
777: *bp->rptr = ctl;
778: bp->flags |= S_DELIM;
779: DPRINT("sCTL %ulx\n", ctl);
780: PUTNEXT(q, bp);
781: }
782:
783: /*
784: * send a reject
785: */
786: static void
787: sendrej(Urp *up)
788: {
789: Queue *q = up->wq;
790: flushinput(up);
791: qlock(&up->ack);
792: if((up->lastecho&~Nmask) == ECHO){
793: DPRINT("REJ %d\n", up->iseq);
794: sendctl(up, up->lastecho = REJ|up->iseq);
795: }
796: qunlock(&up->ack);
797: }
798:
799: /*
800: * send an acknowledge
801: */
802: static void
803: sendack(Urp *up)
804: {
805: /*
806: * check the precondition for acking
807: */
808: if(QFULL(up->rq->next) || (up->lastecho&Nmask)==up->iseq)
809: return;
810:
811: if(!canqlock(&up->ack))
812: return;
813:
814: /*
815: * check again now that we've locked
816: */
817: if(QFULL(up->rq->next) || (up->lastecho&Nmask)==up->iseq){
818: qunlock(&up->ack);
819: return;
820: }
821:
822: /*
823: * send the ack
824: */
825: { Queue *q = up->wq; DPRINT("sendack: "); }
826: sendctl(up, up->lastecho = ECHO|up->iseq);
827: qunlock(&up->ack);
828: }
829:
830: /*
831: * send a block.
832: */
833: static void
834: sendblock(Urp *up, int bn)
835: {
836: Block *bp, *m, *nbp;
837: int n;
838: Queue *q;
839:
840: q = up->wq;
841: up->timer = NOW + MSrexmit;
842: if(QFULL(q->next))
843: return;
844:
845: /*
846: * message 1, the BOT and the data
847: */
848: bp = up->xb[bn];
849: if(bp == 0)
850: return;
851: m = allocb(1);
852: m->rptr = m->lim - 1;
853: m->wptr = m->lim;
854: *m->rptr = (bp->flags & S_DELIM) ? BOT : BOTM;
855: nbp = allocb(0);
856: nbp->rptr = bp->rptr;
857: nbp->wptr = bp->wptr;
858: nbp->base = bp->base;
859: nbp->lim = bp->lim;
860: nbp->flags |= S_DELIM;
861: if(bp->type == M_CTL){
862: PUTNEXT(q, nbp);
863: m->flags |= S_DELIM;
864: PUTNEXT(q, m);
865: } else {
866: m->next = nbp;
867: PUTNEXT(q, m);
868: }
869:
870: /*
871: * message 2, the block length and the SEQ
872: */
873: m = allocb(3);
874: m->rptr = m->lim - 3;
875: m->wptr = m->lim;
876: n = BLEN(bp);
877: m->rptr[0] = SEQ | bn;
878: m->rptr[1] = n;
879: m->rptr[2] = n>>8;
880: m->flags |= S_DELIM;
881: PUTNEXT(q, m);
882: DPRINT("sb %d (%d)\n", bn, up->timer);
883: }
884:
885: /*
886: * receive an acknowledgement
887: */
888: static void
889: rcvack(Urp *up, int msg)
890: {
891: int seqno;
892: int next;
893: int i;
894:
895: seqno = msg&Nmask;
896: next = NEXT(seqno);
897:
898: /*
899: * release any acknowledged blocks
900: */
901: if(IN(seqno, up->unacked, up->next)){
902: for(; up->unacked != next; up->unacked = NEXT(up->unacked)){
903: i = up->unacked;
904: qlock(&up->xl[i]);
905: if(up->xb[i])
906: freeb(up->xb[i]);
907: up->xb[i] = 0;
908: qunlock(&up->xl[i]);
909: }
910: }
911:
912: switch(msg & 0370){
913: case ECHO:
914: if(IN(seqno, up->unechoed, up->next)) {
915: up->unechoed = next;
916: }
917: /*
918: * the next reject at the start of a window starts a
919: * retransmission.
920: */
921: up->state &= ~REJECTING;
922: break;
923: case REJ:
924: if(IN(seqno, up->unechoed, up->next))
925: up->unechoed = next;
926: /*
927: * ... FALL THROUGH ...
928: */
929: case ACK:
930: /*
931: * start a retransmission if we aren't retransmitting
932: * and this is the start of a window.
933: */
934: if(up->unechoed==next && !(up->state & REJECTING)){
935: up->state |= REJECTING;
936: up->rexmit = 1;
937: }
938: break;
939: }
940:
941: tryoutput(up);
942: if(up->state & CLOSING)
943: wakeup(&up->r);
944: }
945:
946: /*
947: * throw away any partially collected input
948: */
949: static void
950: flushinput(Urp *up)
951: {
952: Block *bp;
953:
954: while (bp = getq(up->rq))
955: freeb(bp);
956: up->trx = 0;
957: }
958:
959: /*
960: * initialize output
961: */
962: static void
963: initoutput(Urp *up, int window)
964: {
965: int i;
966:
967: /*
968: * set output window
969: */
970: up->maxblock = window/4;
971: if(up->maxblock < 64)
972: up->maxblock = 64;
973: up->maxblock -= 4;
974: up->maxout = 4;
975:
976: /*
977: * set sequence varialbles
978: */
979: up->unechoed = 1;
980: up->unacked = 1;
981: up->next = 1;
982: up->nxb = 1;
983: up->rexmit = 0;
984:
985: /*
986: * free any outstanding blocks
987: */
988: for(i = 0; i < 8; i++){
989: qlock(&up->xl[i]);
990: if(up->xb[i])
991: freeb(up->xb[i]);
992: up->xb[i] = 0;
993: qunlock(&up->xl[i]);
994: }
995:
996: /*
997: * tell the other side we've inited
998: */
999: up->state |= INITING;
1000: up->timer = NOW + MSrexmit;
1001: { Queue *q = up->wq; DPRINT("initoutput (%d): ", up->timer); }
1002: sendctl(up, INIT1);
1003: }
1004:
1005: /*
1006: * initialize input
1007: */
1008: static void
1009: initinput(Urp *up)
1010: {
1011: /*
1012: * restart all sequence parameters
1013: */
1014: up->blocks = 0;
1015: up->trx = 0;
1016: up->iseq = 0;
1017: up->lastecho = ECHO+0;
1018: flushinput(up);
1019: }
1020:
1021: static void
1022: urpkproc(void *arg)
1023: {
1024: Urp *up;
1025:
1026: USED(arg);
1027:
1028: if(waserror())
1029: ;
1030:
1031: for(;;){
1032: for(up = urpalloc.urp; up; up = up->list){
1033: if(up->state==0 || (up->state&HUNGUP))
1034: continue;
1035: if(!canqlock(up))
1036: continue;
1037: if(waserror()){
1038: qunlock(up);
1039: continue;
1040: }
1041: if(up->state==0 || (up->state&HUNGUP)){
1042: qunlock(up);
1043: poperror();
1044: continue;
1045: }
1046: if(up->iseq!=(up->lastecho&7) && !QFULL(up->rq->next))
1047: sendack(up);
1048: output(up);
1049: qunlock(up);
1050: poperror();
1051: }
1052: tsleep(&urpkr, return0, 0, 500);
1053: }
1054: }
1055:
1056: /*
1057: * urp got very confused, complain
1058: */
1059: static void
1060: urpvomit(char *msg, Urp* up)
1061: {
1062: print("urpvomit: %s %ux next %d unechoed %d unacked %d nxb %d\n",
1063: msg, up, up->next, up->unechoed, up->unacked, up->nxb);
1064: print("\txb: %ux %ux %ux %ux %ux %ux %ux %ux\n",
1065: up->xb[0], up->xb[1], up->xb[2], up->xb[3], up->xb[4],
1066: up->xb[5], up->xb[6], up->xb[7]);
1067: print("\tiseq: %uo lastecho: %uo trx: %d trbuf: %uo %uo %uo\n",
1068: up->iseq, up->lastecho, up->trx, up->trbuf[0], up->trbuf[1],
1069: up->trbuf[2]);
1070: print("\tupq: %ux %d %d\n", &up->rq->next->r, up->rq->next->nb,
1071: up->rq->next->len);
1072: }
1073:
1074: void
1075: urpfillstats(Chan *c, char *buf, int len)
1076: {
1077: char b[256];
1078:
1079: USED(c);
1080: sprint(b, "in: %d\nout: %d\nrexmit: %d\nrjtrs: %d\nrjpks: %d\nrjseq: %d\nenqsx: %d\nenqsr: %d\n",
1081: urpstat.input, urpstat.output, urpstat.rexmit, urpstat.rjtrs,
1082: urpstat.rjpks, urpstat.rjseq, urpstat.enqsx, urpstat.enqsr);
1083: strncpy(buf, b, len);
1084: }
This archive runs on limited infrastructure. Preserving old code on modern bandwidth. Automated agents are requested to crawl responsibly.