libgo/runtime/chan.c

   1 // Copyright 2009 The Go Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style
   3 // license that can be found in the LICENSE file.
   4
   5 #include "runtime.h"
   6 #include "arch.h"
   7 #include "go-type.h"
   8 #include "race.h"
   9 #include "malloc.h"
  10
  11 #define NOSELGEN        1
  12
  13 static  int32   debug   = 0;
  14
  15 typedef struct  WaitQ   WaitQ;
  16 typedef struct  SudoG   SudoG;
  17 typedef struct  Select  Select;
  18 typedef struct  Scase   Scase;
  19
  20 typedef struct  __go_type_descriptor    Type;
  21 typedef struct  __go_channel_type       ChanType;
  22
  23 struct  SudoG
  24 {
  25         G*      g;              // g and selgen constitute
  26         uint32  selgen;         // a weak pointer to g
  27         SudoG*  link;
  28         int64   releasetime;
  29         byte*   elem;           // data element
  30 };
  31
  32 struct  WaitQ
  33 {
  34         SudoG*  first;
  35         SudoG*  last;
  36 };
  37
  38 struct  Hchan
  39 {
  40         uintgo  qcount;                 // total data in the q
  41         uintgo  dataqsiz;               // size of the circular q
  42         uint16  elemsize;
  43         bool    closed;
  44         uint8   elemalign;
  45         uintgo  sendx;                  // send index
  46         uintgo  recvx;                  // receive index
  47         WaitQ   recvq;                  // list of recv waiters
  48         WaitQ   sendq;                  // list of send waiters
  49         Lock;
  50 };
  51
  52 // Buffer follows Hchan immediately in memory.
  53 // chanbuf(c, i) is pointer to the i'th slot in the buffer.
  54 #define chanbuf(c, i) ((byte*)((c)+1)+(uintptr)(c)->elemsize*(i))
  55
  56 enum
  57 {
  58         // Scase.kind
  59         CaseRecv,
  60         CaseSend,
  61         CaseDefault,
  62 };
  63
  64 struct  Scase
  65 {
  66         SudoG   sg;                     // must be first member (cast to Scase)
  67         Hchan*  chan;                   // chan
  68         uint16  kind;
  69         uint16  index;                  // index to return
  70         bool*   receivedp;              // pointer to received bool (recv2)
  71 };
  72
  73 struct  Select
  74 {
  75         uint16  tcase;                  // total count of scase[]
  76         uint16  ncase;                  // currently filled scase[]
  77         uint16* pollorder;              // case poll order
  78         Hchan** lockorder;              // channel lock order
  79         Scase   scase[1];               // one per case (in order of appearance)
  80 };
  81
  82 static  void    dequeueg(WaitQ*);
  83 static  SudoG*  dequeue(WaitQ*);
  84 static  void    enqueue(WaitQ*, SudoG*);
  85 static  void    racesync(Hchan*, SudoG*);
  86
  87 Hchan*
  88 runtime_makechan_c(ChanType *t, int64 hint)
  89 {
  90         Hchan *c;
  91         uintptr n;
  92         const Type *elem;
  93
  94         elem = t->__element_type;
  95
  96         // compiler checks this but be safe.
  97         if(elem->__size >= (1<<16))
  98                 runtime_throw("makechan: invalid channel element type");
  99
 100         if(hint < 0 || (intgo)hint != hint || (elem->__size > 0 && (uintptr)hint > MaxMem / elem->__size))
 101                 runtime_panicstring("makechan: size out of range");
 102
 103         n = sizeof(*c);
 104
 105         // allocate memory in one call
 106         c = (Hchan*)runtime_mal(n + hint*elem->__size);
 107         c->elemsize = elem->__size;
 108         c->elemalign = elem->__align;
 109         c->dataqsiz = hint;
 110
 111         if(debug)
 112                 runtime_printf("makechan: chan=%p; elemsize=%D; elemalign=%d; dataqsiz=%D\n",
 113                         c, (int64)elem->__size, elem->__align, (int64)c->dataqsiz);
 114
 115         return c;
 116 }
 117
 118 // For reflect
 119 //      func makechan(typ *ChanType, size uint64) (chan)
 120 uintptr reflect_makechan(ChanType *, uint64)
 121   asm ("reflect.makechan");
 122
 123 uintptr
 124 reflect_makechan(ChanType *t, uint64 size)
 125 {
 126         void *ret;
 127         Hchan *c;
 128
 129         c = runtime_makechan_c(t, size);
 130         ret = runtime_mal(sizeof(void*));
 131         __builtin_memcpy(ret, &c, sizeof(void*));
 132         return (uintptr)ret;
 133 }
 134
 135 // makechan(t *ChanType, hint int64) (hchan *chan any);
 136 Hchan*
 137 __go_new_channel(ChanType *t, uintptr hint)
 138 {
 139         return runtime_makechan_c(t, hint);
 140 }
 141
 142 Hchan*
 143 __go_new_channel_big(ChanType *t, uint64 hint)
 144 {
 145         return runtime_makechan_c(t, hint);
 146 }
 147
 148 /*
 149  * generic single channel send/recv
 150  * if the bool pointer is nil,
 151  * then the full exchange will
 152  * occur. if pres is not nil,
 153  * then the protocol will not
 154  * sleep but return if it could
 155  * not complete.
 156  *
 157  * sleep can wake up with g->param == nil
 158  * when a channel involved in the sleep has
 159  * been closed.  it is easiest to loop and re-run
 160  * the operation; we'll see that it's now closed.
 161  */
 162 void
 163 runtime_chansend(ChanType *t, Hchan *c, byte *ep, bool *pres, void *pc)
 164 {
 165         SudoG *sg;
 166         SudoG mysg;
 167         G* gp;
 168         int64 t0;
 169         G* g;
 170
 171         g = runtime_g();
 172
 173         if(c == nil) {
 174                 USED(t);
 175                 if(pres != nil) {
 176                         *pres = false;
 177                         return;
 178                 }
 179                 runtime_park(nil, nil, "chan send (nil chan)");
 180                 return;  // not reached
 181         }
 182
 183         if(runtime_gcwaiting)
 184                 runtime_gosched();
 185
 186         if(debug) {
 187                 runtime_printf("chansend: chan=%p\n", c);
 188         }
 189
 190         t0 = 0;
 191         mysg.releasetime = 0;
 192         if(runtime_blockprofilerate > 0) {
 193                 t0 = runtime_cputicks();
 194                 mysg.releasetime = -1;
 195         }
 196
 197         runtime_lock(c);
 198         // TODO(dvyukov): add similar instrumentation to select.
 199         if(raceenabled)
 200                 runtime_racereadpc(c, pc, runtime_chansend);
 201         if(c->closed)
 202                 goto closed;
 203
 204         if(c->dataqsiz > 0)
 205                 goto asynch;
 206
 207         sg = dequeue(&c->recvq);
 208         if(sg != nil) {
 209                 if(raceenabled)
 210                         racesync(c, sg);
 211                 runtime_unlock(c);
 212
 213                 gp = sg->g;
 214                 gp->param = sg;
 215                 if(sg->elem != nil)
 216                         runtime_memmove(sg->elem, ep, c->elemsize);
 217                 if(sg->releasetime)
 218                         sg->releasetime = runtime_cputicks();
 219                 runtime_ready(gp);
 220
 221                 if(pres != nil)
 222                         *pres = true;
 223                 return;
 224         }
 225
 226         if(pres != nil) {
 227                 runtime_unlock(c);
 228                 *pres = false;
 229                 return;
 230         }
 231
 232         mysg.elem = ep;
 233         mysg.g = g;
 234         mysg.selgen = NOSELGEN;
 235         g->param = nil;
 236         enqueue(&c->sendq, &mysg);
 237         runtime_park(runtime_unlock, c, "chan send");
 238
 239         if(g->param == nil) {
 240                 runtime_lock(c);
 241                 if(!c->closed)
 242                         runtime_throw("chansend: spurious wakeup");
 243                 goto closed;
 244         }
 245
 246         if(mysg.releasetime > 0)
 247                 runtime_blockevent(mysg.releasetime - t0, 2);
 248
 249         return;
 250
 251 asynch:
 252         if(c->closed)
 253                 goto closed;
 254
 255         if(c->qcount >= c->dataqsiz) {
 256                 if(pres != nil) {
 257                         runtime_unlock(c);
 258                         *pres = false;
 259                         return;
 260                 }
 261                 mysg.g = g;
 262                 mysg.elem = nil;
 263                 mysg.selgen = NOSELGEN;
 264                 enqueue(&c->sendq, &mysg);
 265                 runtime_park(runtime_unlock, c, "chan send");
 266
 267                 runtime_lock(c);
 268                 goto asynch;
 269         }
 270
 271         if(raceenabled)
 272                 runtime_racerelease(chanbuf(c, c->sendx));
 273
 274         runtime_memmove(chanbuf(c, c->sendx), ep, c->elemsize);
 275         if(++c->sendx == c->dataqsiz)
 276                 c->sendx = 0;
 277         c->qcount++;
 278
 279         sg = dequeue(&c->recvq);
 280         if(sg != nil) {
 281                 gp = sg->g;
 282                 runtime_unlock(c);
 283                 if(sg->releasetime)
 284                         sg->releasetime = runtime_cputicks();
 285                 runtime_ready(gp);
 286         } else
 287                 runtime_unlock(c);
 288         if(pres != nil)
 289                 *pres = true;
 290         if(mysg.releasetime > 0)
 291                 runtime_blockevent(mysg.releasetime - t0, 2);
 292         return;
 293
 294 closed:
 295         runtime_unlock(c);
 296         runtime_panicstring("send on closed channel");
 297 }
 298
 299
 300 void
 301 runtime_chanrecv(ChanType *t, Hchan* c, byte *ep, bool *selected, bool *received)
 302 {
 303         SudoG *sg;
 304         SudoG mysg;
 305         G *gp;
 306         int64 t0;
 307         G *g;
 308
 309         if(runtime_gcwaiting)
 310                 runtime_gosched();
 311
 312         if(debug)
 313                 runtime_printf("chanrecv: chan=%p\n", c);
 314
 315         g = runtime_g();
 316
 317         if(c == nil) {
 318                 USED(t);
 319                 if(selected != nil) {
 320                         *selected = false;
 321                         return;
 322                 }
 323                 runtime_park(nil, nil, "chan receive (nil chan)");
 324                 return;  // not reached
 325         }
 326
 327         t0 = 0;
 328         mysg.releasetime = 0;
 329         if(runtime_blockprofilerate > 0) {
 330                 t0 = runtime_cputicks();
 331                 mysg.releasetime = -1;
 332         }
 333
 334         runtime_lock(c);
 335         if(c->dataqsiz > 0)
 336                 goto asynch;
 337
 338         if(c->closed)
 339                 goto closed;
 340
 341         sg = dequeue(&c->sendq);
 342         if(sg != nil) {
 343                 if(raceenabled)
 344                         racesync(c, sg);
 345                 runtime_unlock(c);
 346
 347                 if(ep != nil)
 348                         runtime_memmove(ep, sg->elem, c->elemsize);
 349                 gp = sg->g;
 350                 gp->param = sg;
 351                 if(sg->releasetime)
 352                         sg->releasetime = runtime_cputicks();
 353                 runtime_ready(gp);
 354
 355                 if(selected != nil)
 356                         *selected = true;
 357                 if(received != nil)
 358                         *received = true;
 359                 return;
 360         }
 361
 362         if(selected != nil) {
 363                 runtime_unlock(c);
 364                 *selected = false;
 365                 return;
 366         }
 367
 368         mysg.elem = ep;
 369         mysg.g = g;
 370         mysg.selgen = NOSELGEN;
 371         g->param = nil;
 372         enqueue(&c->recvq, &mysg);
 373         runtime_park(runtime_unlock, c, "chan receive");
 374
 375         if(g->param == nil) {
 376                 runtime_lock(c);
 377                 if(!c->closed)
 378                         runtime_throw("chanrecv: spurious wakeup");
 379                 goto closed;
 380         }
 381
 382         if(received != nil)
 383                 *received = true;
 384         if(mysg.releasetime > 0)
 385                 runtime_blockevent(mysg.releasetime - t0, 2);
 386         return;
 387
 388 asynch:
 389         if(c->qcount <= 0) {
 390                 if(c->closed)
 391                         goto closed;
 392
 393                 if(selected != nil) {
 394                         runtime_unlock(c);
 395                         *selected = false;
 396                         if(received != nil)
 397                                 *received = false;
 398                         return;
 399                 }
 400                 mysg.g = g;
 401                 mysg.elem = nil;
 402                 mysg.selgen = NOSELGEN;
 403                 enqueue(&c->recvq, &mysg);
 404                 runtime_park(runtime_unlock, c, "chan receive");
 405
 406                 runtime_lock(c);
 407                 goto asynch;
 408         }
 409
 410         if(raceenabled)
 411                 runtime_raceacquire(chanbuf(c, c->recvx));
 412
 413         if(ep != nil)
 414                 runtime_memmove(ep, chanbuf(c, c->recvx), c->elemsize);
 415         runtime_memclr(chanbuf(c, c->recvx), c->elemsize);
 416         if(++c->recvx == c->dataqsiz)
 417                 c->recvx = 0;
 418         c->qcount--;
 419
 420         sg = dequeue(&c->sendq);
 421         if(sg != nil) {
 422                 gp = sg->g;
 423                 runtime_unlock(c);
 424                 if(sg->releasetime)
 425                         sg->releasetime = runtime_cputicks();
 426                 runtime_ready(gp);
 427         } else
 428                 runtime_unlock(c);
 429
 430         if(selected != nil)
 431                 *selected = true;
 432         if(received != nil)
 433                 *received = true;
 434         if(mysg.releasetime > 0)
 435                 runtime_blockevent(mysg.releasetime - t0, 2);
 436         return;
 437
 438 closed:
 439         if(ep != nil)
 440                 runtime_memclr(ep, c->elemsize);
 441         if(selected != nil)
 442                 *selected = true;
 443         if(received != nil)
 444                 *received = false;
 445         if(raceenabled)
 446                 runtime_raceacquire(c);
 447         runtime_unlock(c);
 448         if(mysg.releasetime > 0)
 449                 runtime_blockevent(mysg.releasetime - t0, 2);
 450 }
 451
 452 // The compiler generates a call to __go_send_small to send a value 8
 453 // bytes or smaller.
 454 void
 455 __go_send_small(ChanType *t, Hchan* c, uint64 val)
 456 {
 457         union
 458         {
 459                 byte b[sizeof(uint64)];
 460                 uint64 v;
 461         } u;
 462         byte *p;
 463
 464         u.v = val;
 465 #ifndef WORDS_BIGENDIAN
 466         p = u.b;
 467 #else
 468         p = u.b + sizeof(uint64) - t->__element_type->__size;
 469 #endif
 470         runtime_chansend(t, c, p, nil, runtime_getcallerpc(&t));
 471 }
 472
 473 // The compiler generates a call to __go_send_big to send a value
 474 // larger than 8 bytes or smaller.
 475 void
 476 __go_send_big(ChanType *t, Hchan* c, byte* p)
 477 {
 478         runtime_chansend(t, c, p, nil, runtime_getcallerpc(&t));
 479 }
 480
 481 // The compiler generates a call to __go_receive_small to receive a
 482 // value 8 bytes or smaller.
 483 uint64
 484 __go_receive_small(ChanType *t, Hchan* c)
 485 {
 486         union {
 487                 byte b[sizeof(uint64)];
 488                 uint64 v;
 489         } u;
 490         byte *p;
 491
 492         u.v = 0;
 493 #ifndef WORDS_BIGENDIAN
 494         p = u.b;
 495 #else
 496         p = u.b + sizeof(uint64) - t->__element_type->__size;
 497 #endif
 498         runtime_chanrecv(t, c, p, nil, nil);
 499         return u.v;
 500 }
 501
 502 // The compiler generates a call to __go_receive_big to receive a
 503 // value larger than 8 bytes.
 504 void
 505 __go_receive_big(ChanType *t, Hchan* c, byte* p)
 506 {
 507         runtime_chanrecv(t, c, p, nil, nil);
 508 }
 509
 510 _Bool runtime_chanrecv2(ChanType *t, Hchan* c, byte* p)
 511   __asm__("runtime.chanrecv2");
 512
 513 _Bool
 514 runtime_chanrecv2(ChanType *t, Hchan* c, byte* p)
 515 {
 516         bool received;
 517
 518         runtime_chanrecv(t, c, p, nil, &received);
 519         return received;
 520 }
 521
 522 // func selectnbsend(c chan any, elem any) bool
 523 //
 524 // compiler implements
 525 //
 526 //      select {
 527 //      case c <- v:
 528 //              ... foo
 529 //      default:
 530 //              ... bar
 531 //      }
 532 //
 533 // as
 534 //
 535 //      if selectnbsend(c, v) {
 536 //              ... foo
 537 //      } else {
 538 //              ... bar
 539 //      }
 540 //
 541 _Bool
 542 runtime_selectnbsend(ChanType *t, Hchan *c, byte *p)
 543 {
 544         bool res;
 545
 546         runtime_chansend(t, c, p, &res, runtime_getcallerpc(&t));
 547         return res;
 548 }
 549
 550 // func selectnbrecv(elem *any, c chan any) bool
 551 //
 552 // compiler implements
 553 //
 554 //      select {
 555 //      case v = <-c:
 556 //              ... foo
 557 //      default:
 558 //              ... bar
 559 //      }
 560 //
 561 // as
 562 //
 563 //      if selectnbrecv(&v, c) {
 564 //              ... foo
 565 //      } else {
 566 //              ... bar
 567 //      }
 568 //
 569 _Bool
 570 runtime_selectnbrecv(ChanType *t, byte *v, Hchan *c)
 571 {
 572         bool selected;
 573
 574         runtime_chanrecv(t, c, v, &selected, nil);
 575         return selected;
 576 }
 577
 578 // func selectnbrecv2(elem *any, ok *bool, c chan any) bool
 579 //
 580 // compiler implements
 581 //
 582 //      select {
 583 //      case v, ok = <-c:
 584 //              ... foo
 585 //      default:
 586 //              ... bar
 587 //      }
 588 //
 589 // as
 590 //
 591 //      if c != nil && selectnbrecv2(&v, &ok, c) {
 592 //              ... foo
 593 //      } else {
 594 //              ... bar
 595 //      }
 596 //
 597 _Bool
 598 runtime_selectnbrecv2(ChanType *t, byte *v, _Bool *received, Hchan *c)
 599 {
 600         bool selected;
 601         bool r;
 602
 603         r = false;
 604         runtime_chanrecv(t, c, v, &selected, received == nil ? nil : &r);
 605         if(received != nil)
 606                 *received = r;
 607         return selected;
 608 }
 609
 610 // For reflect:
 611 //      func chansend(c chan, val iword, nb bool) (selected bool)
 612 // where an iword is the same word an interface value would use:
 613 // the actual data if it fits, or else a pointer to the data.
 614
 615 _Bool reflect_chansend(ChanType *, Hchan *, uintptr, _Bool)
 616   __asm__("reflect.chansend");
 617
 618 _Bool
 619 reflect_chansend(ChanType *t, Hchan *c, uintptr val, _Bool nb)
 620 {
 621         bool selected;
 622         bool *sp;
 623         byte *vp;
 624
 625         if(nb) {
 626                 selected = false;
 627                 sp = (bool*)&selected;
 628         } else {
 629                 selected = true;
 630                 sp = nil;
 631         }
 632         if(__go_is_pointer_type(t->__element_type))
 633                 vp = (byte*)&val;
 634         else
 635                 vp = (byte*)val;
 636         runtime_chansend(t, c, vp, sp, runtime_getcallerpc(&t));
 637         return selected;
 638 }
 639
 640 // For reflect:
 641 //      func chanrecv(c chan, nb bool) (val iword, selected, received bool)
 642 // where an iword is the same word an interface value would use:
 643 // the actual data if it fits, or else a pointer to the data.
 644
 645 struct chanrecv_ret
 646 {
 647         uintptr val;
 648         _Bool selected;
 649         _Bool received;
 650 };
 651
 652 struct chanrecv_ret reflect_chanrecv(ChanType *, Hchan *, _Bool)
 653   __asm__("reflect.chanrecv");
 654
 655 struct chanrecv_ret
 656 reflect_chanrecv(ChanType *t, Hchan *c, _Bool nb)
 657 {
 658         struct chanrecv_ret ret;
 659         byte *vp;
 660         bool *sp;
 661         bool selected;
 662         bool received;
 663
 664         if(nb) {
 665                 selected = false;
 666                 sp = &selected;
 667         } else {
 668                 ret.selected = true;
 669                 sp = nil;
 670         }
 671         received = false;
 672         if(__go_is_pointer_type(t->__element_type)) {
 673                 vp = (byte*)&ret.val;
 674         } else {
 675                 vp = runtime_mal(t->__element_type->__size);
 676                 ret.val = (uintptr)vp;
 677         }
 678         runtime_chanrecv(t, c, vp, sp, &received);
 679         if(nb)
 680                 ret.selected = selected;
 681         ret.received = received;
 682         return ret;
 683 }
 684
 685 static void newselect(int32, Select**);
 686
 687 // newselect(size uint32) (sel *byte);
 688
 689 void* runtime_newselect(int32) __asm__("runtime.newselect");
 690
 691 void*
 692 runtime_newselect(int32 size)
 693 {
 694         Select *sel;
 695
 696         newselect(size, &sel);
 697         return (void*)sel;
 698 }
 699
 700 static void
 701 newselect(int32 size, Select **selp)
 702 {
 703         int32 n;
 704         Select *sel;
 705
 706         n = 0;
 707         if(size > 1)
 708                 n = size-1;
 709
 710         // allocate all the memory we need in a single allocation
 711         // start with Select with size cases
 712         // then lockorder with size entries
 713         // then pollorder with size entries
 714         sel = runtime_mal(sizeof(*sel) +
 715                 n*sizeof(sel->scase[0]) +
 716                 size*sizeof(sel->lockorder[0]) +
 717                 size*sizeof(sel->pollorder[0]));
 718
 719         sel->tcase = size;
 720         sel->ncase = 0;
 721         sel->lockorder = (void*)(sel->scase + size);
 722         sel->pollorder = (void*)(sel->lockorder + size);
 723         *selp = sel;
 724
 725         if(debug)
 726                 runtime_printf("newselect s=%p size=%d\n", sel, size);
 727 }
 728
 729 // cut in half to give stack a chance to split
 730 static void selectsend(Select *sel, Hchan *c, int index, void *elem);
 731
 732 // selectsend(sel *byte, hchan *chan any, elem *any) (selected bool);
 733
 734 void runtime_selectsend(Select *, Hchan *, void *, int32)
 735   __asm__("runtime.selectsend");
 736
 737 void
 738 runtime_selectsend(Select *sel, Hchan *c, void *elem, int32 index)
 739 {
 740         // nil cases do not compete
 741         if(c == nil)
 742                 return;
 743
 744         selectsend(sel, c, index, elem);
 745 }
 746
 747 static void
 748 selectsend(Select *sel, Hchan *c, int index, void *elem)
 749 {
 750         int32 i;
 751         Scase *cas;
 752
 753         i = sel->ncase;
 754         if(i >= sel->tcase)
 755                 runtime_throw("selectsend: too many cases");
 756         sel->ncase = i+1;
 757         cas = &sel->scase[i];
 758
 759         cas->index = index;
 760         cas->chan = c;
 761         cas->kind = CaseSend;
 762         cas->sg.elem = elem;
 763
 764         if(debug)
 765                 runtime_printf("selectsend s=%p index=%d chan=%p\n",
 766                         sel, cas->index, cas->chan);
 767 }
 768
 769 // cut in half to give stack a chance to split
 770 static void selectrecv(Select *sel, Hchan *c, int index, void *elem, bool*);
 771
 772 // selectrecv(sel *byte, hchan *chan any, elem *any) (selected bool);
 773
 774 void runtime_selectrecv(Select *, Hchan *, void *, int32)
 775   __asm__("runtime.selectrecv");
 776
 777 void
 778 runtime_selectrecv(Select *sel, Hchan *c, void *elem, int32 index)
 779 {
 780         // nil cases do not compete
 781         if(c == nil)
 782                 return;
 783
 784         selectrecv(sel, c, index, elem, nil);
 785 }
 786
 787 // selectrecv2(sel *byte, hchan *chan any, elem *any, received *bool) (selected bool);
 788
 789 void runtime_selectrecv2(Select *, Hchan *, void *, bool *, int32)
 790   __asm__("runtime.selectrecv2");
 791
 792 void
 793 runtime_selectrecv2(Select *sel, Hchan *c, void *elem, bool *received, int32 index)
 794 {
 795         // nil cases do not compete
 796         if(c == nil)
 797                 return;
 798
 799         selectrecv(sel, c, index, elem, received);
 800 }
 801
 802 static void
 803 selectrecv(Select *sel, Hchan *c, int index, void *elem, bool *received)
 804 {
 805         int32 i;
 806         Scase *cas;
 807
 808         i = sel->ncase;
 809         if(i >= sel->tcase)
 810                 runtime_throw("selectrecv: too many cases");
 811         sel->ncase = i+1;
 812         cas = &sel->scase[i];
 813         cas->index = index;
 814         cas->chan = c;
 815
 816         cas->kind = CaseRecv;
 817         cas->sg.elem = elem;
 818         cas->receivedp = received;
 819
 820         if(debug)
 821                 runtime_printf("selectrecv s=%p index=%d chan=%p\n",
 822                         sel, cas->index, cas->chan);
 823 }
 824
 825 // cut in half to give stack a chance to split
 826 static void selectdefault(Select*, int);
 827
 828 // selectdefault(sel *byte) (selected bool);
 829
 830 void runtime_selectdefault(Select *, int32) __asm__("runtime.selectdefault");
 831
 832 void
 833 runtime_selectdefault(Select *sel, int32 index)
 834 {
 835         selectdefault(sel, index);
 836 }
 837
 838 static void
 839 selectdefault(Select *sel, int32 index)
 840 {
 841         int32 i;
 842         Scase *cas;
 843
 844         i = sel->ncase;
 845         if(i >= sel->tcase)
 846                 runtime_throw("selectdefault: too many cases");
 847         sel->ncase = i+1;
 848         cas = &sel->scase[i];
 849         cas->index = index;
 850         cas->chan = nil;
 851
 852         cas->kind = CaseDefault;
 853
 854         if(debug)
 855                 runtime_printf("selectdefault s=%p index=%d\n",
 856                         sel, cas->index);
 857 }
 858
 859 static void
 860 sellock(Select *sel)
 861 {
 862         uint32 i;
 863         Hchan *c, *c0;
 864
 865         c = nil;
 866         for(i=0; i<sel->ncase; i++) {
 867                 c0 = sel->lockorder[i];
 868                 if(c0 && c0 != c) {
 869                         c = sel->lockorder[i];
 870                         runtime_lock(c);
 871                 }
 872         }
 873 }
 874
 875 static void
 876 selunlock(Select *sel)
 877 {
 878         uint32 i;
 879         Hchan *c, *c0;
 880
 881         c = nil;
 882         for(i=sel->ncase; i-->0;) {
 883                 c0 = sel->lockorder[i];
 884                 if(c0 && c0 != c) {
 885                         c = c0;
 886                         runtime_unlock(c);
 887                 }
 888         }
 889 }
 890
 891 void
 892 runtime_block(void)
 893 {
 894         runtime_park(nil, nil, "select (no cases)");    // forever
 895 }
 896
 897 static int selectgo(Select**);
 898
 899 // selectgo(sel *byte);
 900
 901 int runtime_selectgo(Select *) __asm__("runtime.selectgo");
 902
 903 int
 904 runtime_selectgo(Select *sel)
 905 {
 906         return selectgo(&sel);
 907 }
 908
 909 static int
 910 selectgo(Select **selp)
 911 {
 912         Select *sel;
 913         uint32 o, i, j;
 914         Scase *cas, *dfl;
 915         Hchan *c;
 916         SudoG *sg;
 917         G *gp;
 918         int index;
 919         G *g;
 920
 921         sel = *selp;
 922         if(runtime_gcwaiting)
 923                 runtime_gosched();
 924
 925         if(debug)
 926                 runtime_printf("select: sel=%p\n", sel);
 927
 928         g = runtime_g();
 929
 930         // The compiler rewrites selects that statically have
 931         // only 0 or 1 cases plus default into simpler constructs.
 932         // The only way we can end up with such small sel->ncase
 933         // values here is for a larger select in which most channels
 934         // have been nilled out.  The general code handles those
 935         // cases correctly, and they are rare enough not to bother
 936         // optimizing (and needing to test).
 937
 938         // generate permuted order
 939         for(i=0; i<sel->ncase; i++)
 940                 sel->pollorder[i] = i;
 941         for(i=1; i<sel->ncase; i++) {
 942                 o = sel->pollorder[i];
 943                 j = runtime_fastrand1()%(i+1);
 944                 sel->pollorder[i] = sel->pollorder[j];
 945                 sel->pollorder[j] = o;
 946         }
 947
 948         // sort the cases by Hchan address to get the locking order.
 949         for(i=0; i<sel->ncase; i++) {
 950                 c = sel->scase[i].chan;
 951                 for(j=i; j>0 && sel->lockorder[j-1] >= c; j--)
 952                         sel->lockorder[j] = sel->lockorder[j-1];
 953                 sel->lockorder[j] = c;
 954         }
 955         sellock(sel);
 956
 957 loop:
 958         // pass 1 - look for something already waiting
 959         dfl = nil;
 960         for(i=0; i<sel->ncase; i++) {
 961                 o = sel->pollorder[i];
 962                 cas = &sel->scase[o];
 963                 c = cas->chan;
 964
 965                 switch(cas->kind) {
 966                 case CaseRecv:
 967                         if(c->dataqsiz > 0) {
 968                                 if(c->qcount > 0)
 969                                         goto asyncrecv;
 970                         } else {
 971                                 sg = dequeue(&c->sendq);
 972                                 if(sg != nil)
 973                                         goto syncrecv;
 974                         }
 975                         if(c->closed)
 976                                 goto rclose;
 977                         break;
 978
 979                 case CaseSend:
 980                         if(c->closed)
 981                                 goto sclose;
 982                         if(c->dataqsiz > 0) {
 983                                 if(c->qcount < c->dataqsiz)
 984                                         goto asyncsend;
 985                         } else {
 986                                 sg = dequeue(&c->recvq);
 987                                 if(sg != nil)
 988                                         goto syncsend;
 989                         }
 990                         break;
 991
 992                 case CaseDefault:
 993                         dfl = cas;
 994                         break;
 995                 }
 996         }
 997
 998         if(dfl != nil) {
 999                 selunlock(sel);
1000                 cas = dfl;
1001                 goto retc;
1002         }
1003
1004
1005         // pass 2 - enqueue on all chans
1006         for(i=0; i<sel->ncase; i++) {
1007                 o = sel->pollorder[i];
1008                 cas = &sel->scase[o];
1009                 c = cas->chan;
1010                 sg = &cas->sg;
1011                 sg->g = g;
1012                 sg->selgen = g->selgen;
1013
1014                 switch(cas->kind) {
1015                 case CaseRecv:
1016                         enqueue(&c->recvq, sg);
1017                         break;
1018
1019                 case CaseSend:
1020                         enqueue(&c->sendq, sg);
1021                         break;
1022                 }
1023         }
1024
1025         g->param = nil;
1026         runtime_park((void(*)(Lock*))selunlock, (Lock*)sel, "select");
1027
1028         sellock(sel);
1029         sg = g->param;
1030
1031         // pass 3 - dequeue from unsuccessful chans
1032         // otherwise they stack up on quiet channels
1033         for(i=0; i<sel->ncase; i++) {
1034                 cas = &sel->scase[i];
1035                 if(cas != (Scase*)sg) {
1036                         c = cas->chan;
1037                         if(cas->kind == CaseSend)
1038                                 dequeueg(&c->sendq);
1039                         else
1040                                 dequeueg(&c->recvq);
1041                 }
1042         }
1043
1044         if(sg == nil)
1045                 goto loop;
1046
1047         cas = (Scase*)sg;
1048         c = cas->chan;
1049
1050         if(c->dataqsiz > 0)
1051                 runtime_throw("selectgo: shouldnt happen");
1052
1053         if(debug)
1054                 runtime_printf("wait-return: sel=%p c=%p cas=%p kind=%d\n",
1055                         sel, c, cas, cas->kind);
1056
1057         if(cas->kind == CaseRecv) {
1058                 if(cas->receivedp != nil)
1059                         *cas->receivedp = true;
1060         }
1061
1062         selunlock(sel);
1063         goto retc;
1064
1065 asyncrecv:
1066         // can receive from buffer
1067         if(raceenabled)
1068                 runtime_raceacquire(chanbuf(c, c->recvx));
1069         if(cas->receivedp != nil)
1070                 *cas->receivedp = true;
1071         if(cas->sg.elem != nil)
1072                 runtime_memmove(cas->sg.elem, chanbuf(c, c->recvx), c->elemsize);
1073         runtime_memclr(chanbuf(c, c->recvx), c->elemsize);
1074         if(++c->recvx == c->dataqsiz)
1075                 c->recvx = 0;
1076         c->qcount--;
1077         sg = dequeue(&c->sendq);
1078         if(sg != nil) {
1079                 gp = sg->g;
1080                 selunlock(sel);
1081                 runtime_ready(gp);
1082         } else {
1083                 selunlock(sel);
1084         }
1085         goto retc;
1086
1087 asyncsend:
1088         // can send to buffer
1089         if(raceenabled)
1090                 runtime_racerelease(chanbuf(c, c->sendx));
1091         runtime_memmove(chanbuf(c, c->sendx), cas->sg.elem, c->elemsize);
1092         if(++c->sendx == c->dataqsiz)
1093                 c->sendx = 0;
1094         c->qcount++;
1095         sg = dequeue(&c->recvq);
1096         if(sg != nil) {
1097                 gp = sg->g;
1098                 selunlock(sel);
1099                 runtime_ready(gp);
1100         } else {
1101                 selunlock(sel);
1102         }
1103         goto retc;
1104
1105 syncrecv:
1106         // can receive from sleeping sender (sg)
1107         if(raceenabled)
1108                 racesync(c, sg);
1109         selunlock(sel);
1110         if(debug)
1111                 runtime_printf("syncrecv: sel=%p c=%p o=%d\n", sel, c, o);
1112         if(cas->receivedp != nil)
1113                 *cas->receivedp = true;
1114         if(cas->sg.elem != nil)
1115                 runtime_memmove(cas->sg.elem, sg->elem, c->elemsize);
1116         gp = sg->g;
1117         gp->param = sg;
1118         runtime_ready(gp);
1119         goto retc;
1120
1121 rclose:
1122         // read at end of closed channel
1123         selunlock(sel);
1124         if(cas->receivedp != nil)
1125                 *cas->receivedp = false;
1126         if(cas->sg.elem != nil)
1127                 runtime_memclr(cas->sg.elem, c->elemsize);
1128         if(raceenabled)
1129                 runtime_raceacquire(c);
1130         goto retc;
1131
1132 syncsend:
1133         // can send to sleeping receiver (sg)
1134         if(raceenabled)
1135                 racesync(c, sg);
1136         selunlock(sel);
1137         if(debug)
1138                 runtime_printf("syncsend: sel=%p c=%p o=%d\n", sel, c, o);
1139         if(sg->elem != nil)
1140                 runtime_memmove(sg->elem, cas->sg.elem, c->elemsize);
1141         gp = sg->g;
1142         gp->param = sg;
1143         runtime_ready(gp);
1144
1145 retc:
1146         // return index corresponding to chosen case
1147         index = cas->index;
1148         runtime_free(sel);
1149         return index;
1150
1151 sclose:
1152         // send on closed channel
1153         selunlock(sel);
1154         runtime_panicstring("send on closed channel");
1155         return 0;  // not reached
1156 }
1157
1158 // This struct must match ../reflect/value.go:/runtimeSelect.
1159 typedef struct runtimeSelect runtimeSelect;
1160 struct runtimeSelect
1161 {
1162         uintptr dir;
1163         ChanType *typ;
1164         Hchan *ch;
1165         uintptr val;
1166 };
1167
1168 // This enum must match ../reflect/value.go:/SelectDir.
1169 enum SelectDir {
1170         SelectSend = 1,
1171         SelectRecv,
1172         SelectDefault,
1173 };
1174
1175 struct rselect_ret {
1176         intgo chosen;
1177         uintptr word;
1178         bool recvOK;
1179 };
1180
1181 // func rselect(cases []runtimeSelect) (chosen int, word uintptr, recvOK bool)
1182
1183 struct rselect_ret reflect_rselect(Slice)
1184      asm("reflect.rselect");
1185
1186 struct rselect_ret
1187 reflect_rselect(Slice cases)
1188 {
1189         struct rselect_ret ret;
1190         int32 i;
1191         Select *sel;
1192         runtimeSelect* rcase, *rc;
1193         void *elem;
1194         void *recvptr;
1195         uintptr maxsize;
1196         bool onlyptr;
1197
1198         ret.chosen = -1;
1199         ret.word = 0;
1200         ret.recvOK = false;
1201
1202         maxsize = 0;
1203         onlyptr = true;
1204         rcase = (runtimeSelect*)cases.__values;
1205         for(i=0; i<cases.__count; i++) {
1206                 rc = &rcase[i];
1207                 if(rc->dir == SelectRecv && rc->ch != nil) {
1208                         if(maxsize < rc->typ->__element_type->__size)
1209                                 maxsize = rc->typ->__element_type->__size;
1210                         if(!__go_is_pointer_type(rc->typ->__element_type))
1211                                 onlyptr = false;
1212                 }
1213         }
1214
1215         recvptr = nil;
1216         if(!onlyptr)
1217                 recvptr = runtime_mal(maxsize);
1218
1219         newselect(cases.__count, &sel);
1220         for(i=0; i<cases.__count; i++) {
1221                 rc = &rcase[i];
1222                 switch(rc->dir) {
1223                 case SelectDefault:
1224                         selectdefault(sel, i);
1225                         break;
1226                 case SelectSend:
1227                         if(rc->ch == nil)
1228                                 break;
1229                         if(!__go_is_pointer_type(rc->typ->__element_type))
1230                                 elem = (void*)rc->val;
1231                         else
1232                                 elem = (void*)&rc->val;
1233                         selectsend(sel, rc->ch, i, elem);
1234                         break;
1235                 case SelectRecv:
1236                         if(rc->ch == nil)
1237                                 break;
1238                         if(!__go_is_pointer_type(rc->typ->__element_type))
1239                                 elem = recvptr;
1240                         else
1241                                 elem = &ret.word;
1242                         selectrecv(sel, rc->ch, i, elem, &ret.recvOK);
1243                         break;
1244                 }
1245         }
1246
1247         ret.chosen = (intgo)(uintptr)selectgo(&sel);
1248         if(rcase[ret.chosen].dir == SelectRecv && !__go_is_pointer_type(rcase[ret.chosen].typ->__element_type))
1249                 ret.word = (uintptr)recvptr;
1250
1251         return ret;
1252 }
1253
1254 // closechan(sel *byte);
1255 void
1256 runtime_closechan(Hchan *c)
1257 {
1258         SudoG *sg;
1259         G* gp;
1260
1261         if(c == nil)
1262                 runtime_panicstring("close of nil channel");
1263
1264         if(runtime_gcwaiting)
1265                 runtime_gosched();
1266
1267         runtime_lock(c);
1268         if(c->closed) {
1269                 runtime_unlock(c);
1270                 runtime_panicstring("close of closed channel");
1271         }
1272
1273         if(raceenabled) {
1274                 runtime_racewritepc(c, runtime_getcallerpc(&c), runtime_closechan);
1275                 runtime_racerelease(c);
1276         }
1277
1278         c->closed = true;
1279
1280         // release all readers
1281         for(;;) {
1282                 sg = dequeue(&c->recvq);
1283                 if(sg == nil)
1284                         break;
1285                 gp = sg->g;
1286                 gp->param = nil;
1287                 runtime_ready(gp);
1288         }
1289
1290         // release all writers
1291         for(;;) {
1292                 sg = dequeue(&c->sendq);
1293                 if(sg == nil)
1294                         break;
1295                 gp = sg->g;
1296                 gp->param = nil;
1297                 runtime_ready(gp);
1298         }
1299
1300         runtime_unlock(c);
1301 }
1302
1303 void
1304 __go_builtin_close(Hchan *c)
1305 {
1306         runtime_closechan(c);
1307 }
1308
1309 // For reflect
1310 //      func chanclose(c chan)
1311
1312 void reflect_chanclose(uintptr) __asm__("reflect.chanclose");
1313
1314 void
1315 reflect_chanclose(uintptr c)
1316 {
1317         runtime_closechan((Hchan*)c);
1318 }
1319
1320 // For reflect
1321 //      func chanlen(c chan) (len int)
1322
1323 intgo reflect_chanlen(uintptr) __asm__("reflect.chanlen");
1324
1325 intgo
1326 reflect_chanlen(uintptr ca)
1327 {
1328         Hchan *c;
1329         intgo len;
1330
1331         c = (Hchan*)ca;
1332         if(c == nil)
1333                 len = 0;
1334         else
1335                 len = c->qcount;
1336         return len;
1337 }
1338
1339 intgo
1340 __go_chan_len(Hchan *c)
1341 {
1342         return reflect_chanlen((uintptr)c);
1343 }
1344
1345 // For reflect
1346 //      func chancap(c chan) (cap intgo)
1347
1348 intgo reflect_chancap(uintptr) __asm__("reflect.chancap");
1349
1350 intgo
1351 reflect_chancap(uintptr ca)
1352 {
1353         Hchan *c;
1354         intgo cap;
1355
1356         c = (Hchan*)ca;
1357         if(c == nil)
1358                 cap = 0;
1359         else
1360                 cap = c->dataqsiz;
1361         return cap;
1362 }
1363
1364 intgo
1365 __go_chan_cap(Hchan *c)
1366 {
1367         return reflect_chancap((uintptr)c);
1368 }
1369
1370 static SudoG*
1371 dequeue(WaitQ *q)
1372 {
1373         SudoG *sgp;
1374
1375 loop:
1376         sgp = q->first;
1377         if(sgp == nil)
1378                 return nil;
1379         q->first = sgp->link;
1380
1381         // if sgp is stale, ignore it
1382         if(sgp->selgen != NOSELGEN &&
1383                 (sgp->selgen != sgp->g->selgen ||
1384                 !runtime_cas(&sgp->g->selgen, sgp->selgen, sgp->selgen + 2))) {
1385                 //prints("INVALID PSEUDOG POINTER\n");
1386                 goto loop;
1387         }
1388
1389         return sgp;
1390 }
1391
1392 static void
1393 dequeueg(WaitQ *q)
1394 {
1395         SudoG **l, *sgp, *prevsgp;
1396         G *g;
1397
1398         g = runtime_g();
1399         prevsgp = nil;
1400         for(l=&q->first; (sgp=*l) != nil; l=&sgp->link, prevsgp=sgp) {
1401                 if(sgp->g == g) {
1402                         *l = sgp->link;
1403                         if(q->last == sgp)
1404                                 q->last = prevsgp;
1405                         break;
1406                 }
1407         }
1408 }
1409
1410 static void
1411 enqueue(WaitQ *q, SudoG *sgp)
1412 {
1413         sgp->link = nil;
1414         if(q->first == nil) {
1415                 q->first = sgp;
1416                 q->last = sgp;
1417                 return;
1418         }
1419         q->last->link = sgp;
1420         q->last = sgp;
1421 }
1422
1423 static void
1424 racesync(Hchan *c, SudoG *sg)
1425 {
1426         runtime_racerelease(chanbuf(c, 0));
1427         runtime_raceacquireg(sg->g, chanbuf(c, 0));
1428         runtime_racereleaseg(sg->g, chanbuf(c, 0));
1429         runtime_raceacquire(chanbuf(c, 0));
1430 }