![[BACK]](/cvsweb/icons/back.gif){kind=icon}
Return to devnonet.c CVS log ![[TXT]](/cvsweb/icons/text.gif){kind=icon}
![[DIR]](/cvsweb/icons/dir.gif){kind=icon}
Up to [Research Unix] / researchv10no / sys / io|
Return to devnonet.c CVS log | Up to [Research Unix] / researchv10no / sys / io |
1.1 ! root 1: #include "u.h" ! 2: #include "lib.h" ! 3: #include "mem.h" ! 4: #include "dat.h" ! 5: #include "fns.h" ! 6: #include "io.h" ! 7: #include "errno.h" ! 8: ! 9: #define NOW (MACHP(0)->ticks*MS2HZ) ! 10: ! 11: enum { ! 12: /* ! 13: * configuration parameters ! 14: */ ! 15: Nconv = 32, /* total number of active circuits */ ! 16: Nifc = 2, /* number of interfaces */ ! 17: MSrexmit = 300, /* retranmission interval in ms */ ! 18: MSack = 50, /* ms to sit on an ack */ ! 19: ! 20: /* ! 21: * relative or immutable ! 22: */ ! 23: Ndir = Nconv + 1, /* entries in the nonet directory */ ! 24: Nsubdir = 2, /* entries in the nonet directory */ ! 25: Nmsg = 128, /* max number of outstanding messages */ ! 26: Nmask = Nmsg - 1, /* mask for log(Nmsg) bits */ ! 27: }; ! 28: ! 29: typedef struct Hdr Hdr; ! 30: typedef struct Msg Msg; ! 31: typedef struct Conversation Conversation; ! 32: typedef struct Interface Interface; ! 33: typedef struct Etherhdr Etherhdr; ! 34: ! 35: /* ! 36: * generic nonet header ! 37: */ ! 38: struct Hdr { ! 39: uchar circuit[3]; /* circuit number */ ! 40: uchar flag; ! 41: uchar mid; /* message id */ ! 42: uchar ack; /* piggy back ack */ ! 43: uchar remain[2]; /* count of remaing bytes of data */ ! 44: uchar sum[2]; /* checksum (0 means none) */ ! 45: }; ! 46: #define HDRSIZE 10 ! 47: #define NEWCALL 0x1 /* flag bit marking a new circuit */ ! 48: #define HANGUP 0x2 /* flag bit requesting hangup */ ! 49: #define ACKME 0x4 /* acknowledge this message */ ! 50: ! 51: /* ! 52: * a buffer describing a nonet message ! 53: */ ! 54: struct Msg { ! 55: QLock; ! 56: Blist; ! 57: int mid; /* sequence number */ ! 58: int rem; /* remaining */ ! 59: long time; ! 60: int acked; ! 61: }; ! 62: ! 63: /* ! 64: * Nonet conversation states (for Conversation.state) ! 65: */ ! 66: enum { ! 67: Cclosed, ! 68: Copen, ! 69: Clistening, ! 70: Cconnected, ! 71: Cconnecting, ! 72: Chungup, ! 73: Cclosing, ! 74: }; ! 75: ! 76: /* ! 77: * one exists for each Nonet conversation. ! 78: */ ! 79: struct Conversation { ! 80: QLock; ! 81: ! 82: Queue *rq; /* input queue */ ! 83: int version; /* incremented each time struct is changed */ ! 84: int state; /* true if listening */ ! 85: ! 86: Msg in[Nmsg]; /* messages being received */ ! 87: int rcvcircuit; /* circuit number of incoming packets */ ! 88: ! 89: uchar ack[Nmsg]; /* acknowledgements waiting to be sent */ ! 90: long atime[Nmsg]; ! 91: int afirst; ! 92: int anext; ! 93: ! 94: QLock xlock; /* one trasmitter at a time */ ! 95: Rendez r; /* process waiting for an output mid */ ! 96: Msg ctl; /* for control messages */ ! 97: Msg out[Nmsg]; /* messages being sent */ ! 98: int first; /* first unacknowledged message */ ! 99: int next; /* next message buffer to use */ ! 100: int lastacked; /* last message acked */ ! 101: Block *media; /* prototype media output header */ ! 102: Hdr *hdr; /* nonet header inside of media header */ ! 103: ! 104: Interface *ifc; ! 105: int kstarted; ! 106: char raddr[64]; /* remote address */ ! 107: int rexmit; /* statistics */ ! 108: int retry; ! 109: int bad; ! 110: int sent; ! 111: int rcvd; ! 112: }; ! 113: ! 114: /* ! 115: * a nonet interface. one exists for every stream that a ! 116: * nonet multiplexor is pushed onto. ! 117: */ ! 118: struct Interface { ! 119: QLock; ! 120: int ref; ! 121: char name[64]; /* interface name */ ! 122: int maxtu; /* maximum transfer unit */ ! 123: int mintu; /* minimum transfer unit */ ! 124: int hsize; /* media header size */ ! 125: Queue *wq; /* interface output queue */ ! 126: void (*connect)(Conversation *, char *); ! 127: Conversation conv[Nconv]; ! 128: }; ! 129: static Interface interface[Nifc]; ! 130: ! 131: void nonetkproc(void *); ! 132: int cksum(Block*, int); ! 133: extern Qinfo noetherinfo; ! 134: ! 135: /* ! 136: * start a new conversation. start an ack/retransmit process if ! 137: * none already exists for this circuit. ! 138: */ ! 139: static void ! 140: startconv(Conversation *cp, int circuit, int state) ! 141: { ! 142: int i; ! 143: char name[32]; ! 144: Interface *ifc; ! 145: ! 146: ifc = cp->ifc; ! 147: ! 148: /* ! 149: * allocate the prototype header ! 150: */ ! 151: cp->media = allocb(ifc->hsize + HDRSIZE); ! 152: cp->media->wptr += ifc->hsize + HDRSIZE; ! 153: cp->hdr = (Hdr *)(cp->media->rptr + ifc->hsize); ! 154: ! 155: /* ! 156: * fill in the circuit number ! 157: */ ! 158: cp->hdr->flag = NEWCALL|ACKME; ! 159: cp->hdr->circuit[2] = circuit>>16; ! 160: cp->hdr->circuit[1] = circuit>>8; ! 161: cp->hdr->circuit[0] = circuit; ! 162: ! 163: /* ! 164: * set the state variables ! 165: */ ! 166: cp->state = state; ! 167: cp->retry = 0; ! 168: for(i = 1; i < Nmsg; i++){ ! 169: cp->in[i].mid = i; ! 170: cp->in[i].acked = 0; ! 171: cp->in[i].rem = 0; ! 172: cp->out[i].mid = i | Nmsg; ! 173: cp->out[i].acked = 1; ! 174: cp->out[i].rem = 0; ! 175: } ! 176: cp->in[0].mid = Nmsg; ! 177: cp->in[0].acked = 0; ! 178: cp->in[0].rem = 0; ! 179: cp->out[0].mid = 0; ! 180: cp->out[0].acked = 1; ! 181: cp->out[0].rem = 0; ! 182: cp->first = cp->next = 1; ! 183: cp->rexmit = cp->bad = cp->sent = cp->rcvd = cp->lastacked = 0; ! 184: ! 185: /* ! 186: * used for demultiplexing ! 187: */ ! 188: cp->rcvcircuit = circuit ^ 1; ! 189: ! 190: /* ! 191: * start the ack/rexmit process ! 192: */ ! 193: if(cp->kstarted == 0){ ! 194: cp->kstarted = 1; ! 195: sprint(name, "**nonet%d**", cp - ifc->conv); ! 196: kproc(name, nonetkproc, cp); ! 197: } ! 198: } ! 199: ! 200: /* ! 201: * connect to the destination whose name is pointed to by bp->rptr. ! 202: */ ! 203: void ! 204: connect(Conversation *cp, Block *bp) ! 205: { ! 206: Interface *ifc; ! 207: char *iname; ! 208: int len; ! 209: int circuit; ! 210: ! 211: ifc = cp->ifc; ! 212: qlock(ifc); ! 213: if(waserror()){ ! 214: qunlock(ifc); ! 215: if(cp->media){ ! 216: freeb(cp->media); ! 217: cp->media = 0; ! 218: } ! 219: freeb(bp); ! 220: nexterror(); ! 221: } ! 222: ! 223: /* ! 224: * init ! 225: */ ! 226: startconv(cp, (++(cp->version) * Nconv) + 2*(cp - ifc->conv), Cconnecting); ! 227: (*ifc->connect)(cp, (char *)bp->rptr); ! 228: strncpy(cp->raddr, (char *)bp->rptr, sizeof(cp->raddr)); ! 229: ! 230: qunlock(ifc); ! 231: freeb(bp); ! 232: poperror(); ! 233: } ! 234: ! 235: /* ! 236: * listen for calls from any interface ! 237: */ ! 238: static int ! 239: listendone(void *a) ! 240: { ! 241: Conversation *cp; ! 242: ! 243: cp = (Conversation *)a; ! 244: return cp->state != Clistening; ! 245: } ! 246: static void ! 247: listen(Conversation *cp, Block *bp) ! 248: { ! 249: freeb(bp); ! 250: if(cp->state >= Clistening){ ! 251: print("listen in use %d %ux %ux\n", cp->state, cp->ifc, interface); ! 252: error(0, Einuse); ! 253: } ! 254: cp->state = Clistening; ! 255: sleep(&cp->r, listendone, cp); ! 256: } ! 257: ! 258: /* ! 259: * send a hangup signal up the stream to get all line disciplines ! 260: * to cease and desist ! 261: */ ! 262: static void ! 263: hangup(Conversation *cp) ! 264: { ! 265: Block *bp; ! 266: Queue *q; ! 267: ! 268: cp->state = Chungup; ! 269: bp = allocb(0); ! 270: bp->type = M_HANGUP; ! 271: q = cp->rq; ! 272: PUTNEXT(q, bp); ! 273: wakeup(&cp->r); ! 274: } ! 275: ! 276: /* ! 277: * process a message acknowledgement. if the message ! 278: * has any xmit buffers queued, free them. ! 279: */ ! 280: static void ! 281: rcvack(Conversation *cp, int mid) ! 282: { ! 283: Msg *mp; ! 284: Block *bp; ! 285: int i; ! 286: ! 287: mp = &cp->out[mid & Nmask]; ! 288: ! 289: /* ! 290: * if already acked, ignore ! 291: */ ! 292: if(mp->acked || mp->mid != mid) ! 293: return; ! 294: mp->acked = 1; ! 295: cp->lastacked = mid; ! 296: ! 297: /* ! 298: * free buffers ! 299: */ ! 300: qlock(mp); ! 301: while(bp = getb(mp)) ! 302: freeb(bp); ! 303: qunlock(mp); ! 304: ! 305: /* ! 306: * advance the first pointer and wakeup any processes waiting for a mid ! 307: */ ! 308: if((mid&Nmask) == cp->first){ ! 309: cp->retry = 0; ! 310: for(i = cp->first; i!=cp->next && cp->out[i].acked; i = (i+1)&Nmask) ! 311: ; ! 312: cp->first = i; ! 313: wakeup(&cp->r); ! 314: } ! 315: } ! 316: ! 317: /* ! 318: * queue an acknowledgement to be sent. ignore it if we already have Nmsg ! 319: * acknowledgements queued. ! 320: */ ! 321: static void ! 322: queueack(Conversation *cp, int mid) ! 323: { ! 324: int next; ! 325: ulong now; ! 326: ! 327: now = NOW; ! 328: next = (cp->anext + 1)&Nmask; ! 329: if(next != cp->afirst){ ! 330: cp->ack[cp->anext] = mid; ! 331: cp->atime[cp->anext] = now + MSack; ! 332: cp->anext = next; ! 333: } ! 334: if(now > cp->atime[cp->afirst]) ! 335: wakeup(&cp->rq->r); ! 336: } ! 337: ! 338: /* ! 339: * make a packet header ! 340: */ ! 341: Block * ! 342: mkhdr(Conversation *cp, int rem) ! 343: { ! 344: Block *bp; ! 345: Hdr *hp; ! 346: ! 347: bp = allocb(cp->ifc->hsize + HDRSIZE + cp->ifc->mintu); ! 348: memcpy(bp->wptr, cp->media->rptr, cp->ifc->hsize + HDRSIZE); ! 349: bp->wptr += cp->ifc->hsize + HDRSIZE; ! 350: hp = (Hdr *)(bp->rptr + cp->ifc->hsize); ! 351: hp->remain[1] = rem>>8; ! 352: hp->remain[0] = rem; ! 353: hp->sum[0] = hp->sum[1] = 0; ! 354: return bp; ! 355: } ! 356: ! 357: /* ! 358: * transmit a message. this involves breaking a possibly multi-block message into ! 359: * a train of packets on the media. ! 360: * ! 361: * called by nonetoput() and nonetkproc(). the qlock(mp) synchronizes these two ! 362: * processes. ! 363: */ ! 364: static void ! 365: sendmsg(Conversation *cp, Msg *mp) ! 366: { ! 367: Interface *ifc; ! 368: Queue *wq; ! 369: int msgrem; ! 370: int pktrem; ! 371: int n; ! 372: Block *bp, *pkt, *last; ! 373: uchar *rptr; ! 374: ! 375: ifc = cp->ifc; ! 376: if(ifc == 0) ! 377: return; ! 378: wq = ifc->wq->next; ! 379: ! 380: /* ! 381: * one transmitter at a time ! 382: */ ! 383: qlock(&cp->xlock); ! 384: ! 385: /* ! 386: * synchronize with rcvack, don't want to send while the ! 387: * message is being freed. ! 388: */ ! 389: qlock(mp); ! 390: ! 391: if(waserror() || mp->acked){ ! 392: qunlock(&cp->xlock); ! 393: qunlock(mp); ! 394: return; ! 395: } ! 396: ! 397: /* ! 398: * get the next acknowledge to use if the next queue up ! 399: * is not full. ! 400: */ ! 401: if(cp->afirst != cp->anext && cp->rq->next->len < 16*1024){ ! 402: cp->hdr->ack = cp->ack[cp->afirst]; ! 403: cp->afirst = (cp->afirst+1)&Nmask; ! 404: } ! 405: cp->hdr->mid = mp->mid; ! 406: ! 407: if(ifc->mintu > mp->len) { ! 408: /* ! 409: * short message: ! 410: * copy the whole message into the header block ! 411: */ ! 412: last = pkt = mkhdr(cp, mp->len); ! 413: for(bp = mp->first; bp; bp = bp->next){ ! 414: memcpy(pkt->wptr, bp->rptr, n = BLEN(bp)); ! 415: pkt->wptr += n; ! 416: } ! 417: memset(pkt->wptr, 0, n = ifc->mintu - mp->len); ! 418: pkt->wptr += n; ! 419: } else { ! 420: /* ! 421: * long message: ! 422: * break up the message into interface packets and send them. ! 423: * once around this loop for each non-header block generated. ! 424: */ ! 425: msgrem = mp->len; ! 426: pktrem = msgrem > ifc->maxtu ? ifc->maxtu : msgrem; ! 427: bp = mp->first; ! 428: if(bp) ! 429: rptr = bp->rptr; ! 430: last = pkt = mkhdr(cp, msgrem); ! 431: while(bp){ ! 432: /* ! 433: * if pkt full, send and create new header block ! 434: */ ! 435: if(pktrem == 0){ ! 436: cksum(pkt, ifc->hsize); ! 437: last->flags |= S_DELIM; ! 438: (*wq->put)(wq, pkt); ! 439: last = pkt = mkhdr(cp, -msgrem); ! 440: pktrem = msgrem > ifc->maxtu ? ifc->maxtu : msgrem; ! 441: } ! 442: n = bp->wptr - rptr; ! 443: if(n > pktrem) ! 444: n = pktrem; ! 445: last = last->next = allocb(0); ! 446: last->rptr = rptr; ! 447: last->wptr = rptr = rptr + n; ! 448: msgrem -= n; ! 449: pktrem -= n; ! 450: if(rptr >= bp->wptr){ ! 451: bp = bp->next; ! 452: if(bp) ! 453: rptr = bp->rptr; ! 454: } ! 455: } ! 456: } ! 457: cksum(pkt, ifc->hsize); ! 458: last->flags |= S_DELIM; ! 459: (*wq->put)(wq, pkt); ! 460: mp->time = NOW + MSrexmit; ! 461: qunlock(mp); ! 462: qunlock(&cp->xlock); ! 463: poperror(); ! 464: } ! 465: ! 466: /* ! 467: * send a control message (hangup or acknowledgement). ! 468: */ ! 469: sendctlmsg(Conversation *cp, int flag, int new) ! 470: { ! 471: cp->ctl.len = 0; ! 472: cp->ctl.first = 0; ! 473: cp->ctl.acked = 0; ! 474: if(new) ! 475: cp->ctl.mid = Nmsg^cp->out[cp->next].mid; ! 476: else ! 477: cp->ctl.mid = cp->lastacked; ! 478: cp->hdr->flag |= flag; ! 479: sendmsg(cp, &cp->ctl); ! 480: } ! 481: ! 482: /* ! 483: * receive a message (called by the multiplexor; noetheriput, nofddiiput, ...) ! 484: */ ! 485: static void ! 486: rcvmsg(Conversation *cp, Block *bp) ! 487: { ! 488: Block *nbp; ! 489: Hdr *h; ! 490: short r; ! 491: int c; ! 492: Msg *mp; ! 493: int f; ! 494: Queue *q; ! 495: ! 496: q = cp->rq; ! 497: ! 498: /* ! 499: * grab the packet header, push the pointer past the nonet header ! 500: */ ! 501: h = (Hdr *)bp->rptr; ! 502: bp->rptr += HDRSIZE; ! 503: mp = &cp->in[h->mid & Nmask]; ! 504: r = (h->remain[1]<<8) | h->remain[0]; ! 505: f = h->flag; ! 506: ! 507: /* ! 508: * if a new call request comes in on a connected channel, hang up the call ! 509: */ ! 510: if(h->mid==0 && (f & NEWCALL) && cp->state==Cconnected){ ! 511: freeb(bp); ! 512: hangup(cp); ! 513: return; ! 514: } ! 515: ! 516: /* ! 517: * ignore old messages and process the acknowledgement ! 518: */ ! 519: if(h->mid != mp->mid){ ! 520: if(r == 0){ ! 521: rcvack(cp, h->ack); ! 522: if(f & HANGUP) ! 523: hangup(cp); ! 524: } else { ! 525: if(r>0) ! 526: queueack(cp, h->mid); ! 527: cp->bad++; ! 528: } ! 529: freeb(bp); ! 530: return; ! 531: } ! 532: ! 533: if(r>=0){ ! 534: /* ! 535: * start of message packet ! 536: */ ! 537: if(mp->first){ ! 538: cp->bad++; ! 539: freeb(bp); ! 540: return; ! 541: } ! 542: mp->rem = r; ! 543: } else { ! 544: /* ! 545: * a continuation ! 546: */ ! 547: if(-r != mp->rem) { ! 548: cp->bad++; ! 549: freeb(bp); ! 550: return; ! 551: } ! 552: } ! 553: ! 554: /* ! 555: * take care of packets that were padded up ! 556: */ ! 557: mp->rem -= BLEN(bp); ! 558: if(mp->rem < 0){ ! 559: if(-mp->rem <= BLEN(bp)){ ! 560: bp->wptr += mp->rem; ! 561: mp->rem = 0; ! 562: } else ! 563: panic("rcvmsg: short packet"); ! 564: } ! 565: putb(mp, bp); ! 566: ! 567: /* ! 568: * if the last chunk - pass it up the stream and wake any ! 569: * waiting process. ! 570: * ! 571: * if not, strip off the delimiter. ! 572: */ ! 573: if(mp->rem == 0){ ! 574: rcvack(cp, h->ack); ! 575: if(f & ACKME) ! 576: queueack(cp, h->mid); ! 577: mp->last->flags |= S_DELIM; ! 578: PUTNEXT(q, mp->first); ! 579: mp->first = mp->last = 0; ! 580: mp->len = 0; ! 581: cp->rcvd++; ! 582: ! 583: /* ! 584: * cycle bufffer to next expected mid ! 585: */ ! 586: mp->mid ^= Nmsg; ! 587: ! 588: /* ! 589: * stop xmitting the NEWCALL flag ! 590: */ ! 591: if(cp->state==Cconnecting && !(f & NEWCALL)) ! 592: cp->state = Cconnected; ! 593: } else ! 594: mp->last->flags &= ~S_DELIM; ! 595: ! 596: } ! 597: ! 598: /* ! 599: * the device stream module definition ! 600: */ ! 601: static void nonetstopen(Queue *, Stream *); ! 602: static void nonetstclose(Queue *); ! 603: static void nonetoput(Queue *, Block *); ! 604: static void nonetiput(Queue *, Block *); ! 605: Qinfo nonetinfo = { nonetiput, nonetoput, nonetstopen, nonetstclose, "nonet" } ; ! 606: ! 607: /* ! 608: * store the device end of the stream so that the multiplexor can ! 609: * send blocks upstream. this is called by streamopen() when a ! 610: * nonet device steam is created. ! 611: */ ! 612: static void ! 613: nonetstopen(Queue *q, Stream *s) ! 614: { ! 615: Interface *ifc; ! 616: Conversation *cp; ! 617: ! 618: ifc = &interface[s->dev]; ! 619: cp = &ifc->conv[s->id]; ! 620: cp->ifc = ifc; ! 621: cp->rq = RD(q); ! 622: cp->state = Copen; ! 623: RD(q)->ptr = WR(q)->ptr = (void *)cp; ! 624: } ! 625: ! 626: /* ! 627: * wait until all output has drained or a hangup is received. ! 628: * then send a hangup message (until one is received). ! 629: */ ! 630: static int ! 631: isflushed(void *a) ! 632: { ! 633: Conversation *cp; ! 634: ! 635: cp = (Conversation *)a; ! 636: return cp->first == cp->next || cp->state == Chungup; ! 637: } ! 638: static int ! 639: ishungup(void *a) ! 640: { ! 641: Conversation *cp; ! 642: ! 643: cp = (Conversation *)a; ! 644: return cp->state == Chungup; ! 645: } ! 646: static int ! 647: isdead(void *a) ! 648: { ! 649: Conversation *cp; ! 650: ! 651: cp = (Conversation *)a; ! 652: return cp->kstarted == 0; ! 653: } ! 654: static void ! 655: nonetstclose(Queue *q) ! 656: { ! 657: Conversation *cp; ! 658: Msg *mp; ! 659: int i; ! 660: ! 661: cp = (Conversation *)q->ptr; ! 662: ! 663: /* ! 664: * wait for all messages to drain ! 665: */ ! 666: while(!isflushed(cp)) ! 667: sleep(&cp->r, isflushed, cp); ! 668: ! 669: /* ! 670: * ack all outstanding messages ! 671: */ ! 672: for(; cp->first != cp->next; cp->first = (cp->first+1)&Nmask) { ! 673: mp = &cp->out[cp->first]; ! 674: if(!mp->acked) ! 675: rcvack(cp, mp->mid); ! 676: } ! 677: cp->first = cp->next; ! 678: ! 679: /* ! 680: * send hangup messages to the other side ! 681: * until it hangs up. ! 682: */ ! 683: if(cp->state >= Cconnected){ ! 684: sendctlmsg(cp, HANGUP, 1); ! 685: for(i=0; i<10 && !ishungup(cp); i++){ ! 686: sendctlmsg(cp, HANGUP, 1); ! 687: tsleep(&cp->r, ishungup, cp, MSrexmit); ! 688: } ! 689: } ! 690: ! 691: /* ! 692: * kill off the nonetkproc ! 693: */ ! 694: cp->state = Cclosed; ! 695: wakeup(&cp->rq->r); ! 696: sleep(&cp->r, isdead, cp); ! 697: ! 698: /* ! 699: * close down, synchronizing with interface ! 700: */ ! 701: qlock(cp); ! 702: cp->ifc = 0; ! 703: qunlock(cp); ! 704: } ! 705: ! 706: /* ! 707: * send all messages up stream. this should only be control messages ! 708: */ ! 709: static void ! 710: nonetiput(Queue *q, Block *bp) ! 711: { ! 712: Conversation *cp; ! 713: ! 714: if(bp->type == M_HANGUP){ ! 715: cp = (Conversation *)q->ptr; ! 716: cp->state = Chungup; ! 717: } ! 718: PUTNEXT(q, bp); ! 719: } ! 720: ! 721: /* ! 722: * queue a block ! 723: */ ! 724: static int ! 725: windowopen(void *a) ! 726: { ! 727: Conversation *cp; ! 728: ! 729: cp = (Conversation *)a; ! 730: return (cp->next + 1)&Nmask != cp->first; ! 731: } ! 732: static void ! 733: nonetoput(Queue *q, Block *bp) ! 734: { ! 735: Conversation *cp; ! 736: int next; ! 737: Msg *mp; ! 738: ! 739: cp = (Conversation *)(q->ptr); ! 740: ! 741: /* ! 742: * do all control functions ! 743: */ ! 744: if(bp->type != M_DATA){ ! 745: if(streamparse("connect", bp)) ! 746: connect(cp, bp); ! 747: else if(streamparse("listen", bp)) ! 748: listen(cp, bp); ! 749: else ! 750: freeb(bp); ! 751: return; ! 752: } ! 753: ! 754: /* ! 755: * collect till we see a delim ! 756: */ ! 757: if(!putb(q, bp)) ! 758: return; ! 759: ! 760: /* ! 761: * block if we don't have any free mid's ! 762: */ ! 763: while((next = (cp->next + 1)&Nmask) == cp->first) ! 764: sleep(&cp->r, windowopen, cp); ! 765: ! 766: /* ! 767: * stick the message in a Msg structure ! 768: */ ! 769: mp = &cp->out[cp->next]; ! 770: mp->time = NOW + MSrexmit; ! 771: mp->first = q->first; ! 772: mp->last = q->last; ! 773: mp->len = q->len; ! 774: mp->mid ^= Nmsg; ! 775: mp->acked = 0; ! 776: ! 777: /* ! 778: * init the queue for new messages ! 779: */ ! 780: q->len = 0; ! 781: q->first = q->last = 0; ! 782: ! 783: /* ! 784: * send it ! 785: */ ! 786: cp->next = next; ! 787: sendmsg(cp, mp); ! 788: cp->sent++; ! 789: } ! 790: ! 791: /* ! 792: * wake up every 250 ms to send and ack or resend a message ! 793: */ ! 794: static int ! 795: always(void *a) ! 796: { ! 797: return 0; ! 798: } ! 799: void ! 800: nonetkproc(void *arg) ! 801: { ! 802: Conversation *cp; ! 803: Interface *ifc; ! 804: Queue *wq; ! 805: Msg *mp; ! 806: int i; ! 807: ! 808: cp = (Conversation *)arg; ! 809: ! 810: for(;;){ ! 811: /* ! 812: * die on request ! 813: */ ! 814: if(cp->state == Cclosed){ ! 815: cp->kstarted = 0; ! 816: wakeup(&cp->r); ! 817: return; ! 818: } ! 819: ! 820: /* ! 821: * retransmit first message ! 822: */ ! 823: if(cp->first != cp->next){ ! 824: mp = &cp->out[cp->first]; ! 825: if(!mp->acked && NOW >= mp->time){ ! 826: if(cp->retry++ > 400) ! 827: hangup(cp); ! 828: else { ! 829: cp->rexmit++; ! 830: sendmsg(cp, mp); ! 831: } ! 832: } ! 833: } ! 834: ! 835: /* ! 836: * send any ack whose time is come ! 837: */ ! 838: while(cp->afirst != cp->anext && NOW >= cp->atime[cp->anext] ! 839: && cp->rq->next->len < Streamhi) ! 840: sendctlmsg(cp, 0, 0); ! 841: tsleep(&cp->rq->r, always, 0, MSrexmit/2); ! 842: } ! 843: } ! 844: ! 845: /* ! 846: * nonet directory and subdirectory ! 847: */ ! 848: enum { ! 849: Nraddrqid, ! 850: Nstatsqid, ! 851: Nchanqid, ! 852: Ncloneqid, ! 853: }; ! 854: Dirtab nonetdir[Ndir]; ! 855: Dirtab nosubdir[]={ ! 856: "raddr", Nraddrqid, 0, 0600, ! 857: "stats", Nstatsqid, 0, 0600, ! 858: }; ! 859: ! 860: /* ! 861: * nonet file system. most of the calls use dev.c to access the nonet ! 862: * directory and stream.c to access the nonet devices. ! 863: */ ! 864: void ! 865: nonetreset(void) ! 866: { ! 867: newqinfo(&noetherinfo); ! 868: } ! 869: ! 870: /* ! 871: * create the nonet directory. the files are `clone' and stream ! 872: * directories '1' to '32' (or whatever Nconv is in decimal) ! 873: */ ! 874: void ! 875: nonetinit(void) ! 876: { ! 877: int i; ! 878: ! 879: /* ! 880: * create the directory. ! 881: */ ! 882: /* ! 883: * the circuits ! 884: */ ! 885: for(i = 0; i < Nconv; i++) { ! 886: sprint(nonetdir[i].name, "%d", i); ! 887: nonetdir[i].qid = CHDIR|STREAMQID(i, Nchanqid); ! 888: nonetdir[i].length = 0; ! 889: nonetdir[i].perm = 0600; ! 890: } ! 891: ! 892: /* ! 893: * the clone device ! 894: */ ! 895: strcpy(nonetdir[i].name, "clone"); ! 896: nonetdir[i].qid = Ncloneqid; ! 897: nonetdir[i].length = 0; ! 898: nonetdir[i].perm = 0600; ! 899: ! 900: } ! 901: ! 902: Chan* ! 903: nonetattach(char *spec) ! 904: { ! 905: Interface *ifc; ! 906: Chan *c; ! 907: ! 908: /* ! 909: * find an interface with the same name ! 910: */ ! 911: for(ifc = interface; ifc < &interface[Nifc]; ifc++){ ! 912: qlock(ifc); ! 913: if(strcmp(spec, ifc->name)==0 && ifc->wq) { ! 914: ifc->ref++; ! 915: qunlock(ifc); ! 916: break; ! 917: } ! 918: qunlock(ifc); ! 919: } ! 920: if(ifc == &interface[Nifc]) ! 921: error(0, Enoifc); ! 922: c = devattach('n', spec); ! 923: c->dev = 0; ! 924: return c; ! 925: } ! 926: ! 927: Chan* ! 928: nonetclone(Chan *c, Chan *nc) ! 929: { ! 930: Interface *ifc; ! 931: ! 932: c = devclone(c, nc); ! 933: ifc = &interface[c->dev]; ! 934: qlock(ifc); ! 935: ifc->ref++; ! 936: qunlock(ifc); ! 937: return c; ! 938: } ! 939: ! 940: int ! 941: nonetwalk(Chan *c, char *name) ! 942: { ! 943: if(c->qid == CHDIR) ! 944: return devwalk(c, name, nonetdir, Ndir, devgen); ! 945: else ! 946: return devwalk(c, name, nosubdir, Nsubdir, streamgen); ! 947: } ! 948: ! 949: void ! 950: nonetstat(Chan *c, char *dp) ! 951: { ! 952: if(c->qid == CHDIR) ! 953: devstat(c, dp, nonetdir, Ndir, devgen); ! 954: else ! 955: devstat(c, dp, nosubdir, Nsubdir, streamgen); ! 956: } ! 957: ! 958: /* ! 959: * opening a nonet device allocates a Conversation. Opening the `clone' ! 960: * device is a ``macro'' for finding a free Conversation and opening ! 961: * it's ctl file. ! 962: */ ! 963: Chan* ! 964: nonetopen(Chan *c, int omode) ! 965: { ! 966: extern Qinfo nonetinfo; ! 967: Stream *s; ! 968: Conversation *cp; ! 969: Interface *ifc; ! 970: ! 971: if(c->qid == Ncloneqid){ ! 972: ifc = &interface[c->dev]; ! 973: for(cp = &ifc->conv[0]; cp < &ifc->conv[Nconv]; cp++){ ! 974: if(cp->state == Cclosed && canqlock(cp)){ ! 975: if(cp->state != Cclosed){ ! 976: qunlock(cp); ! 977: continue; ! 978: } ! 979: c->qid = CHDIR|STREAMQID(cp-ifc->conv, Nchanqid); ! 980: devwalk(c, "ctl", 0, 0, streamgen); ! 981: streamopen(c, &nonetinfo); ! 982: qunlock(cp); ! 983: break; ! 984: } ! 985: } ! 986: if(cp == &ifc->conv[Nconv]) ! 987: error(0, Enodev); ! 988: } else if(c->qid != CHDIR) ! 989: streamopen(c, &nonetinfo); ! 990: ! 991: c->mode = openmode(omode); ! 992: c->flag |= COPEN; ! 993: c->offset = 0; ! 994: return c; ! 995: } ! 996: ! 997: void ! 998: nonetcreate(Chan *c, char *name, int omode, ulong perm) ! 999: { ! 1000: error(0, Eperm); ! 1001: } ! 1002: ! 1003: void ! 1004: nonetclose(Chan *c) ! 1005: { ! 1006: Interface *ifc; ! 1007: ! 1008: /* real closing happens in lancestclose */ ! 1009: if(c->qid != CHDIR) ! 1010: streamclose(c); ! 1011: ! 1012: ifc = &interface[c->dev]; ! 1013: qlock(ifc); ! 1014: ifc->ref--; ! 1015: qunlock(ifc); ! 1016: } ! 1017: ! 1018: long ! 1019: nonetread(Chan *c, void *a, long n) ! 1020: { ! 1021: int t; ! 1022: Conversation *cp; ! 1023: char stats[256]; ! 1024: ! 1025: t = STREAMTYPE(c->qid); ! 1026: if(t >= Slowqid) ! 1027: return streamread(c, a, n); ! 1028: ! 1029: if(c->qid == CHDIR) ! 1030: return devdirread(c, a, n, nonetdir, Ndir, devgen); ! 1031: if(c->qid & CHDIR) ! 1032: return devdirread(c, a, n, nosubdir, Nsubdir, streamgen); ! 1033: ! 1034: cp = &interface[c->dev].conv[STREAMID(c->qid)]; ! 1035: switch(t){ ! 1036: case Nraddrqid: ! 1037: return stringread(c, a, n, cp->raddr); ! 1038: case Nstatsqid: ! 1039: sprint(stats, "sent: %d\nrcved: %d\nrexmit: %d\nbad: %d\n", ! 1040: cp->sent, cp->rcvd, cp->rexmit, cp->bad); ! 1041: return stringread(c, a, n, stats); ! 1042: } ! 1043: error(0, Eperm); ! 1044: } ! 1045: ! 1046: long ! 1047: nonetwrite(Chan *c, void *a, long n) ! 1048: { ! 1049: int t; ! 1050: ! 1051: t = STREAMTYPE(c->qid); ! 1052: if(t >= Slowqid) ! 1053: return streamwrite(c, a, n, 0); ! 1054: ! 1055: error(0, Eperm); ! 1056: } ! 1057: ! 1058: void ! 1059: nonetremove(Chan *c) ! 1060: { ! 1061: error(0, Eperm); ! 1062: } ! 1063: ! 1064: void ! 1065: nonetwstat(Chan *c, char *dp) ! 1066: { ! 1067: error(0, Eperm); ! 1068: } ! 1069: ! 1070: void ! 1071: noneterrstr(Error *e, char *buf) ! 1072: { ! 1073: rooterrstr(e, buf); ! 1074: } ! 1075: ! 1076: void ! 1077: nonetuserstr(Error *e, char *buf) ! 1078: { ! 1079: extern consuserstr(Error *, char *); ! 1080: ! 1081: consuserstr(e, buf); ! 1082: } ! 1083: ! 1084: /* ! 1085: * interface ! 1086: */ ! 1087: /* ! 1088: * Create an interface. The qlock on ifc prevents a circuit ! 1089: * from connecting to (nonetconnect) or outputting on (nonetoput) ! 1090: * the interface while it is being created. ! 1091: */ ! 1092: Interface * ! 1093: newifc(Queue *q, Stream *s, int maxtu, int mintu, int hsize, ! 1094: void (*connect)(Conversation *, char *)) ! 1095: { ! 1096: Interface *ifc; ! 1097: ! 1098: for(ifc = interface; ifc < &interface[Nifc]; ifc++){ ! 1099: if(ifc->wq == 0){ ! 1100: qlock(ifc); ! 1101: if(ifc->wq) { ! 1102: /* someone was faster than us */ ! 1103: qunlock(ifc); ! 1104: continue; ! 1105: } ! 1106: RD(q)->ptr = WR(q)->ptr = (void *)ifc; ! 1107: ifc->maxtu = maxtu - hsize - HDRSIZE; ! 1108: ifc->mintu = mintu - hsize - HDRSIZE; ! 1109: ifc->hsize = hsize; ! 1110: ifc->connect = connect; ! 1111: ifc->wq = WR(q); ! 1112: ifc->name[0] = 0; ! 1113: qunlock(ifc); ! 1114: return ifc; ! 1115: } ! 1116: } ! 1117: error(0, Enoifc); ! 1118: } ! 1119: ! 1120: /* ! 1121: * Free an interface. ! 1122: */ ! 1123: void ! 1124: freeifc(Interface *ifc) ! 1125: { ! 1126: qlock(ifc); ! 1127: if(ifc->ref){ ! 1128: qunlock(ifc); ! 1129: print("freeifc in use\n"); ! 1130: error(0, Einuse); ! 1131: } ! 1132: ifc->wq = 0; ! 1133: qunlock(ifc); ! 1134: } ! 1135: /* ! 1136: * ethernet specific multiplexor ! 1137: */ ! 1138: /* ! 1139: * ethernet header of a packet ! 1140: */ ! 1141: struct Etherhdr { ! 1142: uchar d[6]; ! 1143: uchar s[6]; ! 1144: uchar type[2]; ! 1145: uchar circuit[3]; /* circuit number */ ! 1146: uchar flag; ! 1147: uchar mid; /* message id */ ! 1148: uchar ack; /* piggy back ack */ ! 1149: uchar remain[2]; /* count of remaing bytes of data */ ! 1150: uchar sum[2]; /* checksum (0 means none) */ ! 1151: }; ! 1152: #define EHDRSIZE 24 ! 1153: #define EMAXBODY (1514-HDRSIZE) /* maximum ethernet packet body */ ! 1154: #define ETHER_TYPE 0x900 /* most significant byte last */ ! 1155: ! 1156: /* ! 1157: * the ethernet multiplexor stream module definition ! 1158: */ ! 1159: static void noetheropen(Queue *, Stream *); ! 1160: static void noetherclose(Queue *); ! 1161: static void noetheroput(Queue *, Block *); ! 1162: static void noetheriput(Queue *, Block *); ! 1163: Qinfo noetherinfo = { noetheriput, nullput, noetheropen, noetherclose, "noether" }; ! 1164: ! 1165: /* ! 1166: * parse an ethernet address (assumed to be 12 ascii hex digits) ! 1167: */ ! 1168: void ! 1169: etherparse(uchar *to, char *from) ! 1170: { ! 1171: int tdig; ! 1172: int fdig; ! 1173: int i; ! 1174: ! 1175: if(strlen(from) != 12) ! 1176: error(0, Ebadnet); ! 1177: ! 1178: for(i = 0; i < 6; i++){ ! 1179: fdig = *from++; ! 1180: tdig = fdig > 'a' ? (fdig - 'a' + 10) ! 1181: : (fdig > 'A' ? (fdig - 'A' + 10) : (fdig - '0')); ! 1182: fdig = *from++; ! 1183: tdig <<= 4; ! 1184: tdig |= fdig > 'a' ? (fdig - 'a' + 10) ! 1185: : (fdig > 'A' ? (fdig - 'A' + 10) : (fdig - '0')); ! 1186: *to++ = tdig; ! 1187: } ! 1188: } ! 1189: ! 1190: /* ! 1191: * perfrorm the ether specific part of nonetconnect. just stick ! 1192: * the address into the prototype header. ! 1193: */ ! 1194: void ! 1195: noetherconnect(Conversation *cp, char *ea) ! 1196: { ! 1197: Etherhdr *eh; ! 1198: ! 1199: /* ! 1200: * special hack for ross ! 1201: */ ! 1202: if(strcmp(ea, "020701005eff")==0) ! 1203: cp->hdr->flag &= ~ACKME; ! 1204: ! 1205: eh = (Etherhdr *)cp->media->rptr; ! 1206: etherparse(eh->d, ea); ! 1207: eh->type[0] = ETHER_TYPE>>8; ! 1208: eh->type[1] = ETHER_TYPE; ! 1209: } ! 1210: ! 1211: /* ! 1212: * set up an ether interface ! 1213: */ ! 1214: static void ! 1215: noetheropen(Queue *q, Stream *s) ! 1216: { ! 1217: newifc(q, s, 1514, 60, 14, noetherconnect); ! 1218: } ! 1219: ! 1220: /* ! 1221: * tear down an ether interface ! 1222: */ ! 1223: static void ! 1224: noetherclose(Queue *q) ! 1225: { ! 1226: Interface *ifc; ! 1227: ! 1228: ifc = (Interface *)(q->ptr); ! 1229: freeifc(ifc); ! 1230: } ! 1231: ! 1232: /* ! 1233: * Input a packet and use the ether address to select the correct ! 1234: * nonet device to pass it to. ! 1235: * ! 1236: * Simplifying assumption: one put == one packet && the complete header ! 1237: * is in the first block. If this isn't true, demultiplexing will not work. ! 1238: */ ! 1239: static void ! 1240: noetheriput(Queue *q, Block *bp) ! 1241: { ! 1242: Interface *ifc; ! 1243: int circuit; ! 1244: Conversation *cp; ! 1245: Etherhdr *h; ! 1246: Etherhdr *ph; ! 1247: ulong s; ! 1248: Block *end; ! 1249: ! 1250: if(bp->type != M_DATA){ ! 1251: PUTNEXT(q, bp); ! 1252: return; ! 1253: } ! 1254: ! 1255: ifc = (Interface *)(q->ptr); ! 1256: h = (Etherhdr *)(bp->rptr); ! 1257: circuit = (h->circuit[2]<<16) | (h->circuit[1]<<8) | h->circuit[0]; ! 1258: s = (h->sum[1]<<8) | h->sum[0]; ! 1259: if(s && s != cksum(bp, 14)){ ! 1260: print("checksum error %ux %ux\n", s, (h->sum[1]<<8) | h->sum[0]); ! 1261: freeb(bp); ! 1262: return; ! 1263: } ! 1264: ! 1265: /* ! 1266: * look for an existing circuit. ! 1267: */ ! 1268: for(cp = &ifc->conv[0]; cp < &ifc->conv[Nconv]; cp++){ ! 1269: if(cp->state > Clistening && circuit == cp->rcvcircuit && canqlock(cp)){ ! 1270: ph = (Etherhdr *)(cp->media->rptr); ! 1271: if(ifc == cp->ifc ! 1272: && circuit == cp->rcvcircuit ! 1273: && cp->state > Clistening ! 1274: && memcmp(ph->d, h->s, sizeof(h->s)) == 0){ ! 1275: cp->hdr->flag &= ~NEWCALL; ! 1276: bp->rptr += ifc->hsize; ! 1277: rcvmsg(cp, bp); ! 1278: qunlock(cp); ! 1279: return; ! 1280: } ! 1281: qunlock(cp); ! 1282: } ! 1283: } ! 1284: ! 1285: /* ! 1286: * ... or one just listening ! 1287: */ ! 1288: if((h->flag & NEWCALL) == 0) { ! 1289: freeb(bp); ! 1290: return; ! 1291: } ! 1292: for(cp = &ifc->conv[0]; cp < &ifc->conv[Nconv]; cp++){ ! 1293: if(cp->state == Clistening && canqlock(cp)) { ! 1294: /* ! 1295: * initialize the conversation ! 1296: */ ! 1297: startconv(cp, circuit^1, Cconnecting); ! 1298: wakeup(&cp->r); ! 1299: sprint(cp->raddr, "%.2ux%.2ux%.2ux%.2ux%.2ux%.2ux", h->s[0], ! 1300: h->s[1], h->s[2], h->s[3], h->s[4], h->s[5]); ! 1301: ! 1302: /* ! 1303: * fill in media dependent prototype header ! 1304: */ ! 1305: ph = (Etherhdr *)(cp->media->rptr); ! 1306: memcpy(ph->d, h->s, sizeof(h->s)); ! 1307: ph->type[0] = ETHER_TYPE>>8; ! 1308: ph->type[1] = ETHER_TYPE; ! 1309: ! 1310: /* ! 1311: * pass on the packet ! 1312: */ ! 1313: bp->rptr += ifc->hsize; ! 1314: rcvmsg(cp, bp); ! 1315: qunlock(cp); ! 1316: return; ! 1317: } ! 1318: } ! 1319: ! 1320: /* ! 1321: * not found ! 1322: */ ! 1323: freeb(bp); ! 1324: } ! 1325: ! 1326: /* ! 1327: * calculate the checksum of a list of blocks. ignore the first `offset' bytes. ! 1328: */ ! 1329: int ! 1330: cksum(Block *bp, int offset) ! 1331: { ! 1332: Block *nbp = bp; ! 1333: uchar *ep, *p; ! 1334: int n; ! 1335: ulong s; ! 1336: Hdr *hp; ! 1337: ! 1338: s = 0; ! 1339: p = bp->rptr + offset; ! 1340: n = bp->wptr - p; ! 1341: hp = (Hdr *)p; ! 1342: hp->sum[0] = hp->sum[1] = 0; ! 1343: for(;;){ ! 1344: ep = p+(n&~0x7); ! 1345: while(p < ep) { ! 1346: s = s + s + p[0]; ! 1347: s = s + s + p[1]; ! 1348: s = s + s + p[2]; ! 1349: s = s + s + p[3]; ! 1350: s = s + s + p[4]; ! 1351: s = s + s + p[5]; ! 1352: s = s + s + p[6]; ! 1353: s = s + s + p[7]; ! 1354: s = (s&0xffff) + (s>>16); ! 1355: p += 8; ! 1356: } ! 1357: ep = p+(n&0x7); ! 1358: while(p < ep) { ! 1359: s = s + s + *p; ! 1360: p++; ! 1361: } ! 1362: s = (s&0xffff) + (s>>16); ! 1363: bp = bp->next; ! 1364: if(bp == 0) ! 1365: break; ! 1366: p = bp->rptr; ! 1367: n = BLEN(bp); ! 1368: } ! 1369: s = (s&0xffff) + (s>>16); ! 1370: hp->sum[1] = s>>8; ! 1371: hp->sum[0] = s; ! 1372: return s & 0xffff; ! 1373: }
This archive runs on limited infrastructure. Preserving old code on modern bandwidth. Automated agents are requested to crawl responsibly.