sys/src/9/ip/rudp.c

   1 /*
   2  *  Reliable User Datagram Protocol, currently only for IPv4.
   3  *  This protocol is compatible with UDP's packet format.
   4  *  It could be done over UDP if need be.
   5  */
   6 #include        "u.h"
   7 #include        "../port/lib.h"
   8 #include        "mem.h"
   9 #include        "dat.h"
  10 #include        "fns.h"
  11 #include        "../port/error.h"
  12
  13 #include        "ip.h"
  14
  15 #define DEBUG   0
  16 #define DPRINT if(DEBUG)print
  17
  18 #define SEQDIFF(a,b) ( (a)>=(b)?\
  19                         (a)-(b):\
  20                         0xffffffffUL-((b)-(a)) )
  21 #define INSEQ(a,start,end) ( (start)<=(end)?\
  22                                 ((a)>(start)&&(a)<=(end)):\
  23                                 ((a)>(start)||(a)<=(end)) )
  24 #define UNACKED(r) SEQDIFF(r->sndseq, r->ackrcvd)
  25 #define NEXTSEQ(a) ( (a)+1 == 0 ? 1 : (a)+1 )
  26
  27 enum
  28 {
  29         UDP_PHDRSIZE    = 12,   /* pseudo header */
  30 //      UDP_HDRSIZE     = 20,   /* pseudo header + udp header */
  31         UDP_RHDRSIZE    = 36,   /* pseudo header + udp header + rudp header */
  32         UDP_IPHDR       = 8,    /* ip header */
  33         IP_UDPPROTO     = 254,
  34         UDP_USEAD7      = 52,   /* size of new ipv6 headers struct */
  35
  36         Rudprxms        = 200,
  37         Rudptickms      = 50,
  38         Rudpmaxxmit     = 10,
  39         Maxunacked      = 100,
  40 };
  41
  42 #define Hangupgen       0xffffffff      /* used only in hangup messages */
  43
  44 typedef struct Udphdr Udphdr;
  45 struct Udphdr
  46 {
  47         /* ip header */
  48         uchar   vihl;           /* Version and header length */
  49         uchar   tos;            /* Type of service */
  50         uchar   length[2];      /* packet length */
  51         uchar   id[2];          /* Identification */
  52         uchar   frag[2];        /* Fragment information */
  53
  54         /* pseudo header starts here */
  55         uchar   Unused;
  56         uchar   udpproto;       /* Protocol */
  57         uchar   udpplen[2];     /* Header plus data length */
  58         uchar   udpsrc[4];      /* Ip source */
  59         uchar   udpdst[4];      /* Ip destination */
  60
  61         /* udp header */
  62         uchar   udpsport[2];    /* Source port */
  63         uchar   udpdport[2];    /* Destination port */
  64         uchar   udplen[2];      /* data length */
  65         uchar   udpcksum[2];    /* Checksum */
  66 };
  67
  68 typedef struct Rudphdr Rudphdr;
  69 struct Rudphdr
  70 {
  71         /* ip header */
  72         uchar   vihl;           /* Version and header length */
  73         uchar   tos;            /* Type of service */
  74         uchar   length[2];      /* packet length */
  75         uchar   id[2];          /* Identification */
  76         uchar   frag[2];        /* Fragment information */
  77
  78         /* pseudo header starts here */
  79         uchar   Unused;
  80         uchar   udpproto;       /* Protocol */
  81         uchar   udpplen[2];     /* Header plus data length */
  82         uchar   udpsrc[4];      /* Ip source */
  83         uchar   udpdst[4];      /* Ip destination */
  84
  85         /* udp header */
  86         uchar   udpsport[2];    /* Source port */
  87         uchar   udpdport[2];    /* Destination port */
  88         uchar   udplen[2];      /* data length (includes rudp header) */
  89         uchar   udpcksum[2];    /* Checksum */
  90
  91         /* rudp header */
  92         uchar   relseq[4];      /* id of this packet (or 0) */
  93         uchar   relsgen[4];     /* generation/time stamp */
  94         uchar   relack[4];      /* packet being acked (or 0) */
  95         uchar   relagen[4];     /* generation/time stamp */
  96 };
  97
  98
  99 /*
 100  *  one state structure per destination
 101  */
 102 typedef struct Reliable Reliable;
 103 struct Reliable
 104 {
 105         Ref;
 106
 107         Reliable *next;
 108
 109         uchar   addr[IPaddrlen];        /* always V6 when put here */
 110         ushort  port;
 111
 112         Block   *unacked;       /* unacked msg list */
 113         Block   *unackedtail;   /*  and its tail */
 114
 115         int     timeout;        /* time since first unacked msg sent */
 116         int     xmits;          /* number of times first unacked msg sent */
 117
 118         ulong   sndseq;         /* next packet to be sent */
 119         ulong   sndgen;         /*  and its generation */
 120
 121         ulong   rcvseq;         /* last packet received */
 122         ulong   rcvgen;         /*  and its generation */
 123
 124         ulong   acksent;        /* last ack sent */
 125         ulong   ackrcvd;        /* last msg for which ack was rcvd */
 126
 127         /* flow control */
 128         QLock   lock;
 129         Rendez  vous;
 130         int     blocked;
 131 };
 132
 133
 134
 135 /* MIB II counters */
 136 typedef struct Rudpstats Rudpstats;
 137 struct Rudpstats
 138 {
 139         ulong   rudpInDatagrams;
 140         ulong   rudpNoPorts;
 141         ulong   rudpInErrors;
 142         ulong   rudpOutDatagrams;
 143 };
 144
 145 typedef struct Rudppriv Rudppriv;
 146 struct Rudppriv
 147 {
 148         Ipht    ht;
 149
 150         /* MIB counters */
 151         Rudpstats       ustats;
 152
 153         /* non-MIB stats */
 154         ulong   csumerr;                /* checksum errors */
 155         ulong   lenerr;                 /* short packet */
 156         ulong   rxmits;                 /* # of retransmissions */
 157         ulong   orders;                 /* # of out of order pkts */
 158
 159         /* keeping track of the ack kproc */
 160         int     ackprocstarted;
 161         QLock   apl;
 162 };
 163
 164
 165 static ulong generation = 0;
 166 static Rendez rend;
 167
 168 /*
 169  *  protocol specific part of Conv
 170  */
 171 typedef struct Rudpcb Rudpcb;
 172 struct Rudpcb
 173 {
 174         QLock;
 175         uchar   headers;
 176         uchar   randdrop;
 177         Reliable *r;
 178 };
 179
 180 /*
 181  * local functions
 182  */
 183 void    relsendack(Conv*, Reliable*, int);
 184 int     reliput(Conv*, Block*, uchar*, ushort);
 185 Reliable *relstate(Rudpcb*, uchar*, ushort, char*);
 186 void    relput(Reliable*);
 187 void    relforget(Conv *, uchar*, int, int);
 188 void    relackproc(void *);
 189 void    relackq(Reliable *, Block*);
 190 void    relhangup(Conv *, Reliable*);
 191 void    relrexmit(Conv *, Reliable*);
 192 void    relput(Reliable*);
 193 void    rudpkick(void *x);
 194
 195 static void
 196 rudpstartackproc(Proto *rudp)
 197 {
 198         Rudppriv *rpriv;
 199         char kpname[KNAMELEN];
 200
 201         rpriv = rudp->priv;
 202         if(rpriv->ackprocstarted == 0){
 203                 qlock(&rpriv->apl);
 204                 if(rpriv->ackprocstarted == 0){
 205                         sprint(kpname, "#I%drudpack", rudp->f->dev);
 206                         kproc(kpname, relackproc, rudp);
 207                         rpriv->ackprocstarted = 1;
 208                 }
 209                 qunlock(&rpriv->apl);
 210         }
 211 }
 212
 213 static char*
 214 rudpconnect(Conv *c, char **argv, int argc)
 215 {
 216         char *e;
 217         Rudppriv *upriv;
 218
 219         upriv = c->p->priv;
 220         rudpstartackproc(c->p);
 221         e = Fsstdconnect(c, argv, argc);
 222         Fsconnected(c, e);
 223         iphtadd(&upriv->ht, c);
 224
 225         return e;
 226 }
 227
 228
 229 static int
 230 rudpstate(Conv *c, char *state, int n)
 231 {
 232         Rudpcb *ucb;
 233         Reliable *r;
 234         int m;
 235
 236         m = snprint(state, n, "%s", c->inuse?"Open":"Closed");
 237         ucb = (Rudpcb*)c->ptcl;
 238         qlock(ucb);
 239         for(r = ucb->r; r; r = r->next)
 240                 m += snprint(state+m, n-m, " %I/%ld", r->addr, UNACKED(r));
 241         m += snprint(state+m, n-m, "\n");
 242         qunlock(ucb);
 243         return m;
 244 }
 245
 246 static char*
 247 rudpannounce(Conv *c, char** argv, int argc)
 248 {
 249         char *e;
 250         Rudppriv *upriv;
 251
 252         upriv = c->p->priv;
 253         rudpstartackproc(c->p);
 254         e = Fsstdannounce(c, argv, argc);
 255         if(e != nil)
 256                 return e;
 257         Fsconnected(c, nil);
 258         iphtadd(&upriv->ht, c);
 259
 260         return nil;
 261 }
 262
 263 static void
 264 rudpcreate(Conv *c)
 265 {
 266         c->rq = qopen(64*1024, Qmsg, 0, 0);
 267         c->wq = qopen(64*1024, Qkick, rudpkick, c);
 268 }
 269
 270 static void
 271 rudpclose(Conv *c)
 272 {
 273         Rudpcb *ucb;
 274         Reliable *r, *nr;
 275         Rudppriv *upriv;
 276
 277         upriv = c->p->priv;
 278         iphtrem(&upriv->ht, c);
 279
 280         /* force out any delayed acks */
 281         ucb = (Rudpcb*)c->ptcl;
 282         qlock(ucb);
 283         for(r = ucb->r; r; r = r->next){
 284                 if(r->acksent != r->rcvseq)
 285                         relsendack(c, r, 0);
 286         }
 287         qunlock(ucb);
 288
 289         qclose(c->rq);
 290         qclose(c->wq);
 291         qclose(c->eq);
 292         ipmove(c->laddr, IPnoaddr);
 293         ipmove(c->raddr, IPnoaddr);
 294         c->lport = 0;
 295         c->rport = 0;
 296
 297         ucb->headers = 0;
 298         ucb->randdrop = 0;
 299         qlock(ucb);
 300         for(r = ucb->r; r; r = nr){
 301                 if(r->acksent != r->rcvseq)
 302                         relsendack(c, r, 0);
 303                 nr = r->next;
 304                 relhangup(c, r);
 305                 relput(r);
 306         }
 307         ucb->r = 0;
 308
 309         qunlock(ucb);
 310 }
 311
 312 /*
 313  *  randomly don't send packets
 314  */
 315 static void
 316 doipoput(Conv *c, Fs *f, Block *bp, int x, int ttl, int tos)
 317 {
 318         Rudpcb *ucb;
 319
 320         ucb = (Rudpcb*)c->ptcl;
 321         if(ucb->randdrop && nrand(100) < ucb->randdrop)
 322                 freeblist(bp);
 323         else
 324                 ipoput4(f, bp, x, ttl, tos, nil);
 325 }
 326
 327 int
 328 flow(void *v)
 329 {
 330         Reliable *r = v;
 331
 332         return UNACKED(r) <= Maxunacked;
 333 }
 334
 335 void
 336 rudpkick(void *x)
 337 {
 338         Conv *c = x;
 339         Udphdr *uh;
 340         ushort rport;
 341         uchar laddr[IPaddrlen], raddr[IPaddrlen];
 342         Block *bp;
 343         Rudpcb *ucb;
 344         Rudphdr *rh;
 345         Reliable *r;
 346         int dlen, ptcllen;
 347         Rudppriv *upriv;
 348         Fs *f;
 349
 350         upriv = c->p->priv;
 351         f = c->p->f;
 352
 353         netlog(c->p->f, Logrudp, "rudp: kick\n");
 354         bp = qget(c->wq);
 355         if(bp == nil)
 356                 return;
 357
 358         ucb = (Rudpcb*)c->ptcl;
 359         switch(ucb->headers) {
 360         case 7:
 361                 /* get user specified addresses */
 362                 bp = pullupblock(bp, UDP_USEAD7);
 363                 if(bp == nil)
 364                         return;
 365                 ipmove(raddr, bp->rp);
 366                 bp->rp += IPaddrlen;
 367                 ipmove(laddr, bp->rp);
 368                 bp->rp += IPaddrlen;
 369                 /* pick interface closest to dest */
 370                 if(ipforme(f, laddr) != Runi)
 371                         findlocalip(f, laddr, raddr);
 372                 bp->rp += IPaddrlen;            /* Ignore ifc address */
 373                 rport = nhgets(bp->rp);
 374                 bp->rp += 2+2;                  /* Ignore local port */
 375                 break;
 376         default:
 377                 ipmove(raddr, c->raddr);
 378                 ipmove(laddr, c->laddr);
 379                 rport = c->rport;
 380                 break;
 381         }
 382
 383         dlen = blocklen(bp);
 384
 385         /* Make space to fit rudp & ip header */
 386         bp = padblock(bp, UDP_IPHDR+UDP_RHDRSIZE);
 387         uh = (Udphdr *)(bp->rp);
 388         uh->vihl = IP_VER4;
 389
 390         rh = (Rudphdr*)uh;
 391
 392         ptcllen = dlen + (UDP_RHDRSIZE-UDP_PHDRSIZE);
 393         uh->Unused = 0;
 394         uh->udpproto = IP_UDPPROTO;
 395         uh->frag[0] = 0;
 396         uh->frag[1] = 0;
 397         hnputs(uh->udpplen, ptcllen);
 398         switch(ucb->headers){
 399         case 7:
 400                 v6tov4(uh->udpdst, raddr);
 401                 hnputs(uh->udpdport, rport);
 402                 v6tov4(uh->udpsrc, laddr);
 403                 break;
 404         default:
 405                 v6tov4(uh->udpdst, c->raddr);
 406                 hnputs(uh->udpdport, c->rport);
 407                 if(ipcmp(c->laddr, IPnoaddr) == 0)
 408                         findlocalip(f, c->laddr, c->raddr);
 409                 v6tov4(uh->udpsrc, c->laddr);
 410                 break;
 411         }
 412         hnputs(uh->udpsport, c->lport);
 413         hnputs(uh->udplen, ptcllen);
 414         uh->udpcksum[0] = 0;
 415         uh->udpcksum[1] = 0;
 416
 417         qlock(ucb);
 418         r = relstate(ucb, raddr, rport, "kick");
 419         r->sndseq = NEXTSEQ(r->sndseq);
 420         hnputl(rh->relseq, r->sndseq);
 421         hnputl(rh->relsgen, r->sndgen);
 422
 423         hnputl(rh->relack, r->rcvseq);  /* ACK last rcvd packet */
 424         hnputl(rh->relagen, r->rcvgen);
 425
 426         if(r->rcvseq != r->acksent)
 427                 r->acksent = r->rcvseq;
 428
 429         hnputs(uh->udpcksum, ptclcsum(bp, UDP_IPHDR, dlen+UDP_RHDRSIZE));
 430
 431         relackq(r, bp);
 432         qunlock(ucb);
 433
 434         upriv->ustats.rudpOutDatagrams++;
 435
 436         DPRINT("sent: %lud/%lud, %lud/%lud\n",
 437                 r->sndseq, r->sndgen, r->rcvseq, r->rcvgen);
 438
 439         doipoput(c, f, bp, 0, c->ttl, c->tos);
 440
 441         if(waserror()) {
 442                 relput(r);
 443                 qunlock(&r->lock);
 444                 nexterror();
 445         }
 446
 447         /* flow control of sorts */
 448         qlock(&r->lock);
 449         if(UNACKED(r) > Maxunacked){
 450                 r->blocked = 1;
 451                 sleep(&r->vous, flow, r);
 452                 r->blocked = 0;
 453         }
 454
 455         qunlock(&r->lock);
 456         relput(r);
 457         poperror();
 458 }
 459
 460 void
 461 rudpiput(Proto *rudp, Ipifc *ifc, Block *bp)
 462 {
 463         int len, olen, ottl;
 464         Udphdr *uh;
 465         Conv *c;
 466         Rudpcb *ucb;
 467         uchar raddr[IPaddrlen], laddr[IPaddrlen];
 468         ushort rport, lport;
 469         Rudppriv *upriv;
 470         Fs *f;
 471         uchar *p;
 472
 473         upriv = rudp->priv;
 474         f = rudp->f;
 475
 476         upriv->ustats.rudpInDatagrams++;
 477
 478         uh = (Udphdr*)(bp->rp);
 479
 480         /* Put back pseudo header for checksum
 481          * (remember old values for icmpnoconv())
 482          */
 483         ottl = uh->Unused;
 484         uh->Unused = 0;
 485         len = nhgets(uh->udplen);
 486         olen = nhgets(uh->udpplen);
 487         hnputs(uh->udpplen, len);
 488
 489         v4tov6(raddr, uh->udpsrc);
 490         v4tov6(laddr, uh->udpdst);
 491         lport = nhgets(uh->udpdport);
 492         rport = nhgets(uh->udpsport);
 493
 494         if(nhgets(uh->udpcksum)) {
 495                 if(ptclcsum(bp, UDP_IPHDR, len+UDP_PHDRSIZE)) {
 496                         upriv->ustats.rudpInErrors++;
 497                         upriv->csumerr++;
 498                         netlog(f, Logrudp, "rudp: checksum error %I\n", raddr);
 499                         DPRINT("rudp: checksum error %I\n", raddr);
 500                         freeblist(bp);
 501                         return;
 502                 }
 503         }
 504
 505         qlock(rudp);
 506
 507         c = iphtlook(&upriv->ht, raddr, rport, laddr, lport);
 508         if(c == nil){
 509                 /* no conversation found */
 510                 upriv->ustats.rudpNoPorts++;
 511                 qunlock(rudp);
 512                 netlog(f, Logudp, "udp: no conv %I!%d -> %I!%d\n", raddr, rport,
 513                         laddr, lport);
 514                 uh->Unused = ottl;
 515                 hnputs(uh->udpplen, olen);
 516                 icmpnoconv(f, bp);
 517                 freeblist(bp);
 518                 return;
 519         }
 520         ucb = (Rudpcb*)c->ptcl;
 521         qlock(ucb);
 522         qunlock(rudp);
 523
 524         if(reliput(c, bp, raddr, rport) < 0){
 525                 qunlock(ucb);
 526                 freeb(bp);
 527                 return;
 528         }
 529
 530         /*
 531          * Trim the packet down to data size
 532          */
 533
 534         len -= (UDP_RHDRSIZE-UDP_PHDRSIZE);
 535         bp = trimblock(bp, UDP_IPHDR+UDP_RHDRSIZE, len);
 536         if(bp == nil) {
 537                 netlog(f, Logrudp, "rudp: len err %I.%d -> %I.%d\n",
 538                         raddr, rport, laddr, lport);
 539                 DPRINT("rudp: len err %I.%d -> %I.%d\n",
 540                         raddr, rport, laddr, lport);
 541                 upriv->lenerr++;
 542                 return;
 543         }
 544
 545         netlog(f, Logrudpmsg, "rudp: %I.%d -> %I.%d l %d\n",
 546                 raddr, rport, laddr, lport, len);
 547
 548         switch(ucb->headers){
 549         case 7:
 550                 /* pass the src address */
 551                 bp = padblock(bp, UDP_USEAD7);
 552                 p = bp->rp;
 553                 ipmove(p, raddr); p += IPaddrlen;
 554                 ipmove(p, laddr); p += IPaddrlen;
 555                 ipmove(p, ifc->lifc->local); p += IPaddrlen;
 556                 hnputs(p, rport); p += 2;
 557                 hnputs(p, lport);
 558                 break;
 559         default:
 560                 /* connection oriented rudp */
 561                 if(ipcmp(c->raddr, IPnoaddr) == 0){
 562                         /* save the src address in the conversation */
 563                         ipmove(c->raddr, raddr);
 564                         c->rport = rport;
 565
 566                         /* reply with the same ip address (if not broadcast) */
 567                         if(ipforme(f, laddr) == Runi)
 568                                 ipmove(c->laddr, laddr);
 569                         else
 570                                 v4tov6(c->laddr, ifc->lifc->local);
 571                 }
 572                 break;
 573         }
 574
 575         if(qfull(c->rq)) {
 576                 netlog(f, Logrudp, "rudp: qfull %I.%d -> %I.%d\n", raddr, rport,
 577                         laddr, lport);
 578                 freeblist(bp);
 579         } else {
 580                 if(bp->next)
 581                         bp = concatblock(bp);
 582                 qpass(c->rq, bp);
 583         }
 584         qunlock(ucb);
 585 }
 586
 587 static char *rudpunknown = "unknown rudp ctl request";
 588
 589 char*
 590 rudpctl(Conv *c, char **f, int n)
 591 {
 592         Rudpcb *ucb;
 593         uchar ip[IPaddrlen];
 594         int x;
 595
 596         ucb = (Rudpcb*)c->ptcl;
 597         if(n < 1)
 598                 return rudpunknown;
 599
 600         if(strcmp(f[0], "headers") == 0){
 601                 ucb->headers = 7;               /* new headers format */
 602                 return nil;
 603         } else if(strcmp(f[0], "hangup") == 0){
 604                 if(n < 3)
 605                         return "bad syntax";
 606                 if (parseip(ip, f[1]) == -1)
 607                         return Ebadip;
 608                 x = atoi(f[2]);
 609                 qlock(ucb);
 610                 relforget(c, ip, x, 1);
 611                 qunlock(ucb);
 612                 return nil;
 613         } else if(strcmp(f[0], "randdrop") == 0){
 614                 x = 10;                 /* default is 10% */
 615                 if(n > 1)
 616                         x = atoi(f[1]);
 617                 if(x > 100 || x < 0)
 618                         return "illegal rudp drop rate";
 619                 ucb->randdrop = x;
 620                 return nil;
 621         }
 622         return rudpunknown;
 623 }
 624
 625 void
 626 rudpadvise(Proto *rudp, Block *bp, char *msg)
 627 {
 628         Udphdr *h;
 629         uchar source[IPaddrlen], dest[IPaddrlen];
 630         ushort psource, pdest;
 631         Conv *s, **p;
 632
 633         h = (Udphdr*)(bp->rp);
 634
 635         v4tov6(dest, h->udpdst);
 636         v4tov6(source, h->udpsrc);
 637         psource = nhgets(h->udpsport);
 638         pdest = nhgets(h->udpdport);
 639
 640         /* Look for a connection */
 641         for(p = rudp->conv; *p; p++) {
 642                 s = *p;
 643                 if(s->rport == pdest)
 644                 if(s->lport == psource)
 645                 if(ipcmp(s->raddr, dest) == 0)
 646                 if(ipcmp(s->laddr, source) == 0){
 647                         if(s->ignoreadvice)
 648                                 break;
 649                         qhangup(s->rq, msg);
 650                         qhangup(s->wq, msg);
 651                         break;
 652                 }
 653         }
 654         freeblist(bp);
 655 }
 656
 657 int
 658 rudpstats(Proto *rudp, char *buf, int len)
 659 {
 660         Rudppriv *upriv;
 661
 662         upriv = rudp->priv;
 663         return snprint(buf, len, "%lud %lud %lud %lud %lud %lud\n",
 664                 upriv->ustats.rudpInDatagrams,
 665                 upriv->ustats.rudpNoPorts,
 666                 upriv->ustats.rudpInErrors,
 667                 upriv->ustats.rudpOutDatagrams,
 668                 upriv->rxmits,
 669                 upriv->orders);
 670 }
 671
 672 void
 673 rudpinit(Fs *fs)
 674 {
 675
 676         Proto *rudp;
 677
 678         rudp = smalloc(sizeof(Proto));
 679         rudp->priv = smalloc(sizeof(Rudppriv));
 680         rudp->name = "rudp";
 681         rudp->connect = rudpconnect;
 682         rudp->announce = rudpannounce;
 683         rudp->ctl = rudpctl;
 684         rudp->state = rudpstate;
 685         rudp->create = rudpcreate;
 686         rudp->close = rudpclose;
 687         rudp->rcv = rudpiput;
 688         rudp->advise = rudpadvise;
 689         rudp->stats = rudpstats;
 690         rudp->ipproto = IP_UDPPROTO;
 691         rudp->nc = 32;
 692         rudp->ptclsize = sizeof(Rudpcb);
 693
 694         Fsproto(fs, rudp);
 695 }
 696
 697 /*********************************************/
 698 /* Here starts the reliable helper functions */
 699 /*********************************************/
 700 /*
 701  *  Enqueue a copy of an unacked block for possible retransmissions
 702  */
 703 void
 704 relackq(Reliable *r, Block *bp)
 705 {
 706         Block *np;
 707
 708         np = copyblock(bp, blocklen(bp));
 709         if(r->unacked)
 710                 r->unackedtail->list = np;
 711         else {
 712                 /* restart timer */
 713                 r->timeout = 0;
 714                 r->xmits = 1;
 715                 r->unacked = np;
 716         }
 717         r->unackedtail = np;
 718         np->list = nil;
 719 }
 720
 721 /*
 722  *  retransmit unacked blocks
 723  */
 724 void
 725 relackproc(void *a)
 726 {
 727         Rudpcb *ucb;
 728         Proto *rudp;
 729         Reliable *r;
 730         Conv **s, *c;
 731
 732         rudp = (Proto *)a;
 733
 734         while(waserror())
 735                 ;
 736 loop:
 737         tsleep(&up->sleep, return0, 0, Rudptickms);
 738
 739         for(s = rudp->conv; *s; s++) {
 740                 c = *s;
 741                 ucb = (Rudpcb*)c->ptcl;
 742                 qlock(ucb);
 743
 744                 for(r = ucb->r; r; r = r->next) {
 745                         if(r->unacked != nil){
 746                                 r->timeout += Rudptickms;
 747                                 if(r->timeout > Rudprxms*r->xmits)
 748                                         relrexmit(c, r);
 749                         }
 750                         if(r->acksent != r->rcvseq)
 751                                 relsendack(c, r, 0);
 752                 }
 753                 qunlock(ucb);
 754         }
 755         goto loop;
 756 }
 757
 758 /*
 759  *  get the state record for a conversation
 760  */
 761 Reliable*
 762 relstate(Rudpcb *ucb, uchar *addr, ushort port, char *from)
 763 {
 764         Reliable *r, **l;
 765
 766         l = &ucb->r;
 767         for(r = *l; r; r = *l){
 768                 if(memcmp(addr, r->addr, IPaddrlen) == 0 &&
 769                     port == r->port)
 770                         break;
 771                 l = &r->next;
 772         }
 773
 774         /* no state for this addr/port, create some */
 775         if(r == nil){
 776                 while(generation == 0)
 777                         generation = rand();
 778
 779                 DPRINT("from %s new state %lud for %I!%ud\n",
 780                         from, generation, addr, port);
 781
 782                 r = smalloc(sizeof(Reliable));
 783                 memmove(r->addr, addr, IPaddrlen);
 784                 r->port = port;
 785                 r->unacked = 0;
 786                 if(generation == Hangupgen)
 787                         generation++;
 788                 r->sndgen = generation++;
 789                 r->sndseq = 0;
 790                 r->ackrcvd = 0;
 791                 r->rcvgen = 0;
 792                 r->rcvseq = 0;
 793                 r->acksent = 0;
 794                 r->xmits = 0;
 795                 r->timeout = 0;
 796                 r->ref = 0;
 797                 incref(r);      /* one reference for being in the list */
 798
 799                 *l = r;
 800         }
 801
 802         incref(r);
 803         return r;
 804 }
 805
 806 void
 807 relput(Reliable *r)
 808 {
 809         if(decref(r) == 0)
 810                 free(r);
 811 }
 812
 813 /*
 814  *  forget a Reliable state
 815  */
 816 void
 817 relforget(Conv *c, uchar *ip, int port, int originator)
 818 {
 819         Rudpcb *ucb;
 820         Reliable *r, **l;
 821
 822         ucb = (Rudpcb*)c->ptcl;
 823
 824         l = &ucb->r;
 825         for(r = *l; r; r = *l){
 826                 if(ipcmp(ip, r->addr) == 0 && port == r->port){
 827                         *l = r->next;
 828                         if(originator)
 829                                 relsendack(c, r, 1);
 830                         relhangup(c, r);
 831                         relput(r);      /* remove from the list */
 832                         break;
 833                 }
 834                 l = &r->next;
 835         }
 836 }
 837
 838 /*
 839  *  process a rcvd reliable packet. return -1 if not to be passed to user process,
 840  *  0 therwise.
 841  *
 842  *  called with ucb locked.
 843  */
 844 int
 845 reliput(Conv *c, Block *bp, uchar *addr, ushort port)
 846 {
 847         Block *nbp;
 848         Rudpcb *ucb;
 849         Rudppriv *upriv;
 850         Udphdr *uh;
 851         Reliable *r;
 852         Rudphdr *rh;
 853         ulong seq, ack, sgen, agen, ackreal;
 854         int rv = -1;
 855
 856         /* get fields */
 857         uh = (Udphdr*)(bp->rp);
 858         rh = (Rudphdr*)uh;
 859         seq = nhgetl(rh->relseq);
 860         sgen = nhgetl(rh->relsgen);
 861         ack = nhgetl(rh->relack);
 862         agen = nhgetl(rh->relagen);
 863
 864         upriv = c->p->priv;
 865         ucb = (Rudpcb*)c->ptcl;
 866         r = relstate(ucb, addr, port, "input");
 867
 868         DPRINT("rcvd %lud/%lud, %lud/%lud, r->sndgen = %lud\n",
 869                 seq, sgen, ack, agen, r->sndgen);
 870
 871         /* if acking an incorrect generation, ignore */
 872         if(ack && agen != r->sndgen)
 873                 goto out;
 874
 875         /* Look for a hangup */
 876         if(sgen == Hangupgen) {
 877                 if(agen == r->sndgen)
 878                         relforget(c, addr, port, 0);
 879                 goto out;
 880         }
 881
 882         /* make sure we're not talking to a new remote side */
 883         if(r->rcvgen != sgen){
 884                 if(seq != 0 && seq != 1)
 885                         goto out;
 886
 887                 /* new connection */
 888                 if(r->rcvgen != 0){
 889                         DPRINT("new con r->rcvgen = %lud, sgen = %lud\n", r->rcvgen, sgen);
 890                         relhangup(c, r);
 891                 }
 892                 r->rcvgen = sgen;
 893         }
 894
 895         /* dequeue acked packets */
 896         if(ack && agen == r->sndgen){
 897                 ackreal = 0;
 898                 while(r->unacked != nil && INSEQ(ack, r->ackrcvd, r->sndseq)){
 899                         nbp = r->unacked;
 900                         r->unacked = nbp->list;
 901                         DPRINT("%lud/%lud acked, r->sndgen = %lud\n",
 902                                ack, agen, r->sndgen);
 903                         freeb(nbp);
 904                         r->ackrcvd = NEXTSEQ(r->ackrcvd);
 905                         ackreal = 1;
 906                 }
 907
 908                 /* flow control */
 909                 if(UNACKED(r) < Maxunacked/8 && r->blocked)
 910                         wakeup(&r->vous);
 911
 912                 /*
 913                  *  retransmit next packet if the acked packet
 914                  *  was transmitted more than once
 915                  */
 916                 if(ackreal && r->unacked != nil){
 917                         r->timeout = 0;
 918                         if(r->xmits > 1){
 919                                 r->xmits = 1;
 920                                 relrexmit(c, r);
 921                         }
 922                 }
 923
 924         }
 925
 926         /* no message or input queue full */
 927         if(seq == 0 || qfull(c->rq))
 928                 goto out;
 929
 930         /* refuse out of order delivery */
 931         if(seq != NEXTSEQ(r->rcvseq)){
 932                 relsendack(c, r, 0);    /* tell him we got it already */
 933                 upriv->orders++;
 934                 DPRINT("out of sequence %lud not %lud\n", seq, NEXTSEQ(r->rcvseq));
 935                 goto out;
 936         }
 937         r->rcvseq = seq;
 938
 939         rv = 0;
 940 out:
 941         relput(r);
 942         return rv;
 943 }
 944
 945 void
 946 relsendack(Conv *c, Reliable *r, int hangup)
 947 {
 948         Udphdr *uh;
 949         Block *bp;
 950         Rudphdr *rh;
 951         int ptcllen;
 952         Fs *f;
 953
 954         bp = allocb(UDP_IPHDR + UDP_RHDRSIZE);
 955         bp->wp += UDP_IPHDR + UDP_RHDRSIZE;
 956         f = c->p->f;
 957         uh = (Udphdr *)(bp->rp);
 958         uh->vihl = IP_VER4;
 959         rh = (Rudphdr*)uh;
 960
 961         ptcllen = (UDP_RHDRSIZE-UDP_PHDRSIZE);
 962         uh->Unused = 0;
 963         uh->udpproto = IP_UDPPROTO;
 964         uh->frag[0] = 0;
 965         uh->frag[1] = 0;
 966         hnputs(uh->udpplen, ptcllen);
 967
 968         v6tov4(uh->udpdst, r->addr);
 969         hnputs(uh->udpdport, r->port);
 970         hnputs(uh->udpsport, c->lport);
 971         if(ipcmp(c->laddr, IPnoaddr) == 0)
 972                 findlocalip(f, c->laddr, c->raddr);
 973         v6tov4(uh->udpsrc, c->laddr);
 974         hnputs(uh->udplen, ptcllen);
 975
 976         if(hangup)
 977                 hnputl(rh->relsgen, Hangupgen);
 978         else
 979                 hnputl(rh->relsgen, r->sndgen);
 980         hnputl(rh->relseq, 0);
 981         hnputl(rh->relagen, r->rcvgen);
 982         hnputl(rh->relack, r->rcvseq);
 983
 984         if(r->acksent < r->rcvseq)
 985                 r->acksent = r->rcvseq;
 986
 987         uh->udpcksum[0] = 0;
 988         uh->udpcksum[1] = 0;
 989         hnputs(uh->udpcksum, ptclcsum(bp, UDP_IPHDR, UDP_RHDRSIZE));
 990
 991         DPRINT("sendack: %lud/%lud, %lud/%lud\n", 0L, r->sndgen, r->rcvseq, r->rcvgen);
 992         doipoput(c, f, bp, 0, c->ttl, c->tos);
 993 }
 994
 995
 996 /*
 997  *  called with ucb locked (and c locked if user initiated close)
 998  */
 999 void
1000 relhangup(Conv *c, Reliable *r)
1001 {
1002         int n;
1003         Block *bp;
1004         char hup[ERRMAX];
1005
1006         n = snprint(hup, sizeof(hup), "hangup %I!%d", r->addr, r->port);
1007         qproduce(c->eq, hup, n);
1008
1009         /*
1010          *  dump any unacked outgoing messages
1011          */
1012         for(bp = r->unacked; bp != nil; bp = r->unacked){
1013                 r->unacked = bp->list;
1014                 bp->list = nil;
1015                 freeb(bp);
1016         }
1017
1018         r->rcvgen = 0;
1019         r->rcvseq = 0;
1020         r->acksent = 0;
1021         if(generation == Hangupgen)
1022                 generation++;
1023         r->sndgen = generation++;
1024         r->sndseq = 0;
1025         r->ackrcvd = 0;
1026         r->xmits = 0;
1027         r->timeout = 0;
1028         wakeup(&r->vous);
1029 }
1030
1031 /*
1032  *  called with ucb locked
1033  */
1034 void
1035 relrexmit(Conv *c, Reliable *r)
1036 {
1037         Rudppriv *upriv;
1038         Block *np;
1039         Fs *f;
1040
1041         upriv = c->p->priv;
1042         f = c->p->f;
1043         r->timeout = 0;
1044         if(r->xmits++ > Rudpmaxxmit){
1045                 relhangup(c, r);
1046                 return;
1047         }
1048
1049         upriv->rxmits++;
1050         np = copyblock(r->unacked, blocklen(r->unacked));
1051         DPRINT("rxmit r->ackrvcd+1 = %lud\n", r->ackrcvd+1);
1052         doipoput(c, f, np, 0, c->ttl, c->tos);
1053 }