libgo/runtime/chan.goc

   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 package runtime
   6 #include "runtime.h"
   7 #include "arch.h"
   8 #include "go-type.h"
   9 #include "race.h"
  10 #include "malloc.h"
  11 #include "chan.h"
  12
  13 uint32 runtime_Hchansize = sizeof(Hchan);
  14
  15 static  void    dequeueg(WaitQ*);
  16 static  SudoG*  dequeue(WaitQ*);
  17 static  void    enqueue(WaitQ*, SudoG*);
  18 static  void    racesync(Hchan*, SudoG*);
  19
  20 static Hchan*
  21 makechan(ChanType *t, int64 hint)
  22 {
  23         Hchan *c;
  24         uintptr n;
  25         const Type *elem;
  26
  27         elem = t->__element_type;
  28
  29         // compiler checks this but be safe.
  30         if(elem->__size >= (1<<16))
  31                 runtime_throw("makechan: invalid channel element type");
  32
  33         if(hint < 0 || (intgo)hint != hint || (elem->__size > 0 && (uintptr)hint > (MaxMem - sizeof(*c)) / elem->__size))
  34                 runtime_panicstring("makechan: size out of range");
  35
  36         n = sizeof(*c);
  37         n = ROUND(n, elem->__align);
  38
  39         // allocate memory in one call
  40         c = (Hchan*)runtime_mallocgc(sizeof(*c) + hint*elem->__size, (uintptr)t | TypeInfo_Chan, 0);
  41         c->elemsize = elem->__size;
  42         c->elemtype = elem;
  43         c->dataqsiz = hint;
  44
  45         if(debug)
  46                 runtime_printf("makechan: chan=%p; elemsize=%D; dataqsiz=%D\n",
  47                         c, (int64)elem->__size, (int64)c->dataqsiz);
  48
  49         return c;
  50 }
  51
  52 func reflect.makechan(t *ChanType, size uint64) (c *Hchan) {
  53         c = makechan(t, size);
  54 }
  55
  56 Hchan*
  57 __go_new_channel(ChanType *t, uintptr hint)
  58 {
  59         return makechan(t, hint);
  60 }
  61
  62 Hchan*
  63 __go_new_channel_big(ChanType *t, uint64 hint)
  64 {
  65         return makechan(t, hint);
  66 }
  67
  68 /*
  69  * generic single channel send/recv
  70  * if the bool pointer is nil,
  71  * then the full exchange will
  72  * occur. if pres is not nil,
  73  * then the protocol will not
  74  * sleep but return if it could
  75  * not complete.
  76  *
  77  * sleep can wake up with g->param == nil
  78  * when a channel involved in the sleep has
  79  * been closed.  it is easiest to loop and re-run
  80  * the operation; we'll see that it's now closed.
  81  */
  82 static bool
  83 chansend(ChanType *t, Hchan *c, byte *ep, bool block, void *pc)
  84 {
  85         SudoG *sg;
  86         SudoG mysg;
  87         G* gp;
  88         int64 t0;
  89         G* g;
  90
  91         g = runtime_g();
  92
  93         if(raceenabled)
  94                 runtime_racereadobjectpc(ep, t->__element_type, runtime_getcallerpc(&t), chansend);
  95
  96         if(c == nil) {
  97                 USED(t);
  98                 if(!block)
  99                         return false;
 100                 runtime_park(nil, nil, "chan send (nil chan)");
 101                 return false;  // not reached
 102         }
 103
 104         if(runtime_gcwaiting())
 105                 runtime_gosched();
 106
 107         if(debug) {
 108                 runtime_printf("chansend: chan=%p\n", c);
 109         }
 110
 111         t0 = 0;
 112         mysg.releasetime = 0;
 113         if(runtime_blockprofilerate > 0) {
 114                 t0 = runtime_cputicks();
 115                 mysg.releasetime = -1;
 116         }
 117
 118         runtime_lock(c);
 119         if(raceenabled)
 120                 runtime_racereadpc(c, pc, chansend);
 121         if(c->closed)
 122                 goto closed;
 123
 124         if(c->dataqsiz > 0)
 125                 goto asynch;
 126
 127         sg = dequeue(&c->recvq);
 128         if(sg != nil) {
 129                 if(raceenabled)
 130                         racesync(c, sg);
 131                 runtime_unlock(c);
 132
 133                 gp = sg->g;
 134                 gp->param = sg;
 135                 if(sg->elem != nil)
 136                         runtime_memmove(sg->elem, ep, c->elemsize);
 137                 if(sg->releasetime)
 138                         sg->releasetime = runtime_cputicks();
 139                 runtime_ready(gp);
 140                 return true;
 141         }
 142
 143         if(!block) {
 144                 runtime_unlock(c);
 145                 return false;
 146         }
 147
 148         mysg.elem = ep;
 149         mysg.g = g;
 150         mysg.selectdone = nil;
 151         g->param = nil;
 152         enqueue(&c->sendq, &mysg);
 153         runtime_parkunlock(c, "chan send");
 154
 155         if(g->param == nil) {
 156                 runtime_lock(c);
 157                 if(!c->closed)
 158                         runtime_throw("chansend: spurious wakeup");
 159                 goto closed;
 160         }
 161
 162         if(mysg.releasetime > 0)
 163                 runtime_blockevent(mysg.releasetime - t0, 2);
 164
 165         return true;
 166
 167 asynch:
 168         if(c->closed)
 169                 goto closed;
 170
 171         if(c->qcount >= c->dataqsiz) {
 172                 if(!block) {
 173                         runtime_unlock(c);
 174                         return false;
 175                 }
 176                 mysg.g = g;
 177                 mysg.elem = nil;
 178                 mysg.selectdone = nil;
 179                 enqueue(&c->sendq, &mysg);
 180                 runtime_parkunlock(c, "chan send");
 181
 182                 runtime_lock(c);
 183                 goto asynch;
 184         }
 185
 186         if(raceenabled)
 187                 runtime_racerelease(chanbuf(c, c->sendx));
 188
 189         runtime_memmove(chanbuf(c, c->sendx), ep, c->elemsize);
 190         if(++c->sendx == c->dataqsiz)
 191                 c->sendx = 0;
 192         c->qcount++;
 193
 194         sg = dequeue(&c->recvq);
 195         if(sg != nil) {
 196                 gp = sg->g;
 197                 runtime_unlock(c);
 198                 if(sg->releasetime)
 199                         sg->releasetime = runtime_cputicks();
 200                 runtime_ready(gp);
 201         } else
 202                 runtime_unlock(c);
 203         if(mysg.releasetime > 0)
 204                 runtime_blockevent(mysg.releasetime - t0, 2);
 205         return true;
 206
 207 closed:
 208         runtime_unlock(c);
 209         runtime_panicstring("send on closed channel");
 210         return false;  // not reached
 211 }
 212
 213
 214 static bool
 215 chanrecv(ChanType *t, Hchan* c, byte *ep, bool block, bool *received)
 216 {
 217         SudoG *sg;
 218         SudoG mysg;
 219         G *gp;
 220         int64 t0;
 221         G *g;
 222
 223         if(runtime_gcwaiting())
 224                 runtime_gosched();
 225
 226         // raceenabled: don't need to check ep, as it is always on the stack.
 227
 228         if(debug)
 229                 runtime_printf("chanrecv: chan=%p\n", c);
 230
 231         g = runtime_g();
 232
 233         if(c == nil) {
 234                 USED(t);
 235                 if(!block)
 236                         return false;
 237                 runtime_park(nil, nil, "chan receive (nil chan)");
 238                 return false;  // not reached
 239         }
 240
 241         t0 = 0;
 242         mysg.releasetime = 0;
 243         if(runtime_blockprofilerate > 0) {
 244                 t0 = runtime_cputicks();
 245                 mysg.releasetime = -1;
 246         }
 247
 248         runtime_lock(c);
 249         if(c->dataqsiz > 0)
 250                 goto asynch;
 251
 252         if(c->closed)
 253                 goto closed;
 254
 255         sg = dequeue(&c->sendq);
 256         if(sg != nil) {
 257                 if(raceenabled)
 258                         racesync(c, sg);
 259                 runtime_unlock(c);
 260
 261                 if(ep != nil)
 262                         runtime_memmove(ep, sg->elem, c->elemsize);
 263                 gp = sg->g;
 264                 gp->param = sg;
 265                 if(sg->releasetime)
 266                         sg->releasetime = runtime_cputicks();
 267                 runtime_ready(gp);
 268
 269                 if(received != nil)
 270                         *received = true;
 271                 return true;
 272         }
 273
 274         if(!block) {
 275                 runtime_unlock(c);
 276                 return false;
 277         }
 278
 279         mysg.elem = ep;
 280         mysg.g = g;
 281         mysg.selectdone = nil;
 282         g->param = nil;
 283         enqueue(&c->recvq, &mysg);
 284         runtime_parkunlock(c, "chan receive");
 285
 286         if(g->param == nil) {
 287                 runtime_lock(c);
 288                 if(!c->closed)
 289                         runtime_throw("chanrecv: spurious wakeup");
 290                 goto closed;
 291         }
 292
 293         if(received != nil)
 294                 *received = true;
 295         if(mysg.releasetime > 0)
 296                 runtime_blockevent(mysg.releasetime - t0, 2);
 297         return true;
 298
 299 asynch:
 300         if(c->qcount <= 0) {
 301                 if(c->closed)
 302                         goto closed;
 303
 304                 if(!block) {
 305                         runtime_unlock(c);
 306                         if(received != nil)
 307                                 *received = false;
 308                         return false;
 309                 }
 310                 mysg.g = g;
 311                 mysg.elem = nil;
 312                 mysg.selectdone = nil;
 313                 enqueue(&c->recvq, &mysg);
 314                 runtime_parkunlock(c, "chan receive");
 315
 316                 runtime_lock(c);
 317                 goto asynch;
 318         }
 319
 320         if(raceenabled)
 321                 runtime_raceacquire(chanbuf(c, c->recvx));
 322
 323         if(ep != nil)
 324                 runtime_memmove(ep, chanbuf(c, c->recvx), c->elemsize);
 325         runtime_memclr(chanbuf(c, c->recvx), c->elemsize);
 326         if(++c->recvx == c->dataqsiz)
 327                 c->recvx = 0;
 328         c->qcount--;
 329
 330         sg = dequeue(&c->sendq);
 331         if(sg != nil) {
 332                 gp = sg->g;
 333                 runtime_unlock(c);
 334                 if(sg->releasetime)
 335                         sg->releasetime = runtime_cputicks();
 336                 runtime_ready(gp);
 337         } else
 338                 runtime_unlock(c);
 339
 340         if(received != nil)
 341                 *received = true;
 342         if(mysg.releasetime > 0)
 343                 runtime_blockevent(mysg.releasetime - t0, 2);
 344         return true;
 345
 346 closed:
 347         if(ep != nil)
 348                 runtime_memclr(ep, c->elemsize);
 349         if(received != nil)
 350                 *received = false;
 351         if(raceenabled)
 352                 runtime_raceacquire(c);
 353         runtime_unlock(c);
 354         if(mysg.releasetime > 0)
 355                 runtime_blockevent(mysg.releasetime - t0, 2);
 356         return true;
 357 }
 358
 359 // The compiler generates a call to __go_send_small to send a value 8
 360 // bytes or smaller.
 361 void
 362 __go_send_small(ChanType *t, Hchan* c, uint64 val)
 363 {
 364         union
 365         {
 366                 byte b[sizeof(uint64)];
 367                 uint64 v;
 368         } u;
 369         byte *v;
 370
 371         u.v = val;
 372 #ifndef WORDS_BIGENDIAN
 373         v = u.b;
 374 #else
 375         v = u.b + sizeof(uint64) - t->__element_type->__size;
 376 #endif
 377         chansend(t, c, v, true, runtime_getcallerpc(&t));
 378 }
 379
 380 // The compiler generates a call to __go_send_big to send a value
 381 // larger than 8 bytes or smaller.
 382 void
 383 __go_send_big(ChanType *t, Hchan* c, byte* v)
 384 {
 385         chansend(t, c, v, true, runtime_getcallerpc(&t));
 386 }
 387
 388 // The compiler generates a call to __go_receive to receive a
 389 // value from a channel.
 390 void
 391 __go_receive(ChanType *t, Hchan* c, byte* v)
 392 {
 393         chanrecv(t, c, v, true, nil);
 394 }
 395
 396 _Bool runtime_chanrecv2(ChanType *t, Hchan* c, byte* v)
 397   __asm__ (GOSYM_PREFIX "runtime.chanrecv2");
 398
 399 _Bool
 400 runtime_chanrecv2(ChanType *t, Hchan* c, byte* v)
 401 {
 402         bool received = false;
 403
 404         chanrecv(t, c, v, true, &received);
 405         return received;
 406 }
 407
 408 // compiler implements
 409 //
 410 //      select {
 411 //      case c <- v:
 412 //              ... foo
 413 //      default:
 414 //              ... bar
 415 //      }
 416 //
 417 // as
 418 //
 419 //      if selectnbsend(c, v) {
 420 //              ... foo
 421 //      } else {
 422 //              ... bar
 423 //      }
 424 //
 425 func selectnbsend(t *ChanType, c *Hchan, elem *byte) (selected bool) {
 426         selected = chansend(t, c, elem, false, runtime_getcallerpc(&t));
 427 }
 428
 429 // compiler implements
 430 //
 431 //      select {
 432 //      case v = <-c:
 433 //              ... foo
 434 //      default:
 435 //              ... bar
 436 //      }
 437 //
 438 // as
 439 //
 440 //      if selectnbrecv(&v, c) {
 441 //              ... foo
 442 //      } else {
 443 //              ... bar
 444 //      }
 445 //
 446 func selectnbrecv(t *ChanType, elem *byte, c *Hchan) (selected bool) {
 447         selected = chanrecv(t, c, elem, false, nil);
 448 }
 449
 450 // compiler implements
 451 //
 452 //      select {
 453 //      case v, ok = <-c:
 454 //              ... foo
 455 //      default:
 456 //              ... bar
 457 //      }
 458 //
 459 // as
 460 //
 461 //      if c != nil && selectnbrecv2(&v, &ok, c) {
 462 //              ... foo
 463 //      } else {
 464 //              ... bar
 465 //      }
 466 //
 467 func selectnbrecv2(t *ChanType, elem *byte, received *bool, c *Hchan) (selected bool) {
 468         bool r;
 469
 470         selected = chanrecv(t, c, elem, false, received == nil ? nil : &r);
 471         if(received != nil)
 472                 *received = r;
 473 }
 474
 475 func reflect.chansend(t *ChanType, c *Hchan, elem *byte, nb bool) (selected bool) {
 476         selected = chansend(t, c, elem, !nb, runtime_getcallerpc(&t));
 477 }
 478
 479 func reflect.chanrecv(t *ChanType, c *Hchan, nb bool, elem *byte) (selected bool, received bool) {
 480         received = false;
 481         selected = chanrecv(t, c, elem, !nb, &received);
 482 }
 483
 484 static Select* newselect(int32);
 485
 486 func newselect(size int32) (sel *byte) {
 487         sel = (byte*)newselect(size);
 488 }
 489
 490 static Select*
 491 newselect(int32 size)
 492 {
 493         int32 n;
 494         Select *sel;
 495
 496         n = 0;
 497         if(size > 1)
 498                 n = size-1;
 499
 500         // allocate all the memory we need in a single allocation
 501         // start with Select with size cases
 502         // then lockorder with size entries
 503         // then pollorder with size entries
 504         sel = runtime_mal(sizeof(*sel) +
 505                 n*sizeof(sel->scase[0]) +
 506                 size*sizeof(sel->lockorder[0]) +
 507                 size*sizeof(sel->pollorder[0]));
 508
 509         sel->tcase = size;
 510         sel->ncase = 0;
 511         sel->lockorder = (void*)(sel->scase + size);
 512         sel->pollorder = (void*)(sel->lockorder + size);
 513
 514         if(debug)
 515                 runtime_printf("newselect s=%p size=%d\n", sel, size);
 516         return sel;
 517 }
 518
 519 // cut in half to give stack a chance to split
 520 static void selectsend(Select *sel, Hchan *c, int index, void *elem);
 521
 522 func selectsend(sel *Select, c *Hchan, elem *byte, index int32) {
 523         // nil cases do not compete
 524         if(c != nil)
 525                 selectsend(sel, c, index, elem);
 526 }
 527
 528 static void
 529 selectsend(Select *sel, Hchan *c, int index, void *elem)
 530 {
 531         int32 i;
 532         Scase *cas;
 533
 534         i = sel->ncase;
 535         if(i >= sel->tcase)
 536                 runtime_throw("selectsend: too many cases");
 537         sel->ncase = i+1;
 538         cas = &sel->scase[i];
 539
 540         cas->index = index;
 541         cas->chan = c;
 542         cas->kind = CaseSend;
 543         cas->sg.elem = elem;
 544
 545         if(debug)
 546                 runtime_printf("selectsend s=%p index=%d chan=%p\n",
 547                         sel, cas->index, cas->chan);
 548 }
 549
 550 // cut in half to give stack a chance to split
 551 static void selectrecv(Select *sel, Hchan *c, int index, void *elem, bool*);
 552
 553 func selectrecv(sel *Select, c *Hchan, elem *byte, index int32) {
 554         // nil cases do not compete
 555         if(c != nil)
 556                 selectrecv(sel, c, index, elem, nil);
 557 }
 558
 559 func selectrecv2(sel *Select, c *Hchan, elem *byte, received *bool, index int32) {
 560         // nil cases do not compete
 561         if(c != nil)
 562                 selectrecv(sel, c, index, elem, received);
 563 }
 564
 565 static void
 566 selectrecv(Select *sel, Hchan *c, int index, void *elem, bool *received)
 567 {
 568         int32 i;
 569         Scase *cas;
 570
 571         i = sel->ncase;
 572         if(i >= sel->tcase)
 573                 runtime_throw("selectrecv: too many cases");
 574         sel->ncase = i+1;
 575         cas = &sel->scase[i];
 576         cas->index = index;
 577         cas->chan = c;
 578
 579         cas->kind = CaseRecv;
 580         cas->sg.elem = elem;
 581         cas->receivedp = received;
 582
 583         if(debug)
 584                 runtime_printf("selectrecv s=%p index=%d chan=%p\n",
 585                         sel, cas->index, cas->chan);
 586 }
 587
 588 // cut in half to give stack a chance to split
 589 static void selectdefault(Select*, int);
 590
 591 func selectdefault(sel *Select, index int32) {
 592         selectdefault(sel, index);
 593 }
 594
 595 static void
 596 selectdefault(Select *sel, int32 index)
 597 {
 598         int32 i;
 599         Scase *cas;
 600
 601         i = sel->ncase;
 602         if(i >= sel->tcase)
 603                 runtime_throw("selectdefault: too many cases");
 604         sel->ncase = i+1;
 605         cas = &sel->scase[i];
 606         cas->index = index;
 607         cas->chan = nil;
 608
 609         cas->kind = CaseDefault;
 610
 611         if(debug)
 612                 runtime_printf("selectdefault s=%p index=%d\n",
 613                         sel, cas->index);
 614 }
 615
 616 static void
 617 sellock(Select *sel)
 618 {
 619         uint32 i;
 620         Hchan *c, *c0;
 621
 622         c = nil;
 623         for(i=0; i<sel->ncase; i++) {
 624                 c0 = sel->lockorder[i];
 625                 if(c0 && c0 != c) {
 626                         c = sel->lockorder[i];
 627                         runtime_lock(c);
 628                 }
 629         }
 630 }
 631
 632 static void
 633 selunlock(Select *sel)
 634 {
 635         int32 i, n, r;
 636         Hchan *c;
 637
 638         // We must be very careful here to not touch sel after we have unlocked
 639         // the last lock, because sel can be freed right after the last unlock.
 640         // Consider the following situation.
 641         // First M calls runtime_park() in runtime_selectgo() passing the sel.
 642         // Once runtime_park() has unlocked the last lock, another M makes
 643         // the G that calls select runnable again and schedules it for execution.
 644         // When the G runs on another M, it locks all the locks and frees sel.
 645         // Now if the first M touches sel, it will access freed memory.
 646         n = (int32)sel->ncase;
 647         r = 0;
 648         // skip the default case
 649         if(n>0 && sel->lockorder[0] == nil)
 650                 r = 1;
 651         for(i = n-1; i >= r; i--) {
 652                 c = sel->lockorder[i];
 653                 if(i>0 && sel->lockorder[i-1] == c)
 654                         continue;  // will unlock it on the next iteration
 655                 runtime_unlock(c);
 656         }
 657 }
 658
 659 static bool
 660 selparkcommit(G *gp, void *sel)
 661 {
 662         USED(gp);
 663         selunlock(sel);
 664         return true;
 665 }
 666
 667 func block() {
 668         runtime_park(nil, nil, "select (no cases)");    // forever
 669 }
 670
 671 static int selectgo(Select**);
 672
 673 // selectgo(sel *byte);
 674
 675 func selectgo(sel *Select) (ret int32) {
 676         return selectgo(&sel);
 677 }
 678
 679 static int
 680 selectgo(Select **selp)
 681 {
 682         Select *sel;
 683         uint32 o, i, j, k, done;
 684         int64 t0;
 685         Scase *cas, *dfl;
 686         Hchan *c;
 687         SudoG *sg;
 688         G *gp;
 689         int index;
 690         G *g;
 691
 692         sel = *selp;
 693         if(runtime_gcwaiting())
 694                 runtime_gosched();
 695
 696         if(debug)
 697                 runtime_printf("select: sel=%p\n", sel);
 698
 699         g = runtime_g();
 700
 701         t0 = 0;
 702         if(runtime_blockprofilerate > 0) {
 703                 t0 = runtime_cputicks();
 704                 for(i=0; i<sel->ncase; i++)
 705                         sel->scase[i].sg.releasetime = -1;
 706         }
 707
 708         // The compiler rewrites selects that statically have
 709         // only 0 or 1 cases plus default into simpler constructs.
 710         // The only way we can end up with such small sel->ncase
 711         // values here is for a larger select in which most channels
 712         // have been nilled out.  The general code handles those
 713         // cases correctly, and they are rare enough not to bother
 714         // optimizing (and needing to test).
 715
 716         // generate permuted order
 717         for(i=0; i<sel->ncase; i++)
 718                 sel->pollorder[i] = i;
 719         for(i=1; i<sel->ncase; i++) {
 720                 o = sel->pollorder[i];
 721                 j = runtime_fastrand1()%(i+1);
 722                 sel->pollorder[i] = sel->pollorder[j];
 723                 sel->pollorder[j] = o;
 724         }
 725
 726         // sort the cases by Hchan address to get the locking order.
 727         // simple heap sort, to guarantee n log n time and constant stack footprint.
 728         for(i=0; i<sel->ncase; i++) {
 729                 j = i;
 730                 c = sel->scase[j].chan;
 731                 while(j > 0 && sel->lockorder[k=(j-1)/2] < c) {
 732                         sel->lockorder[j] = sel->lockorder[k];
 733                         j = k;
 734                 }
 735                 sel->lockorder[j] = c;
 736         }
 737         for(i=sel->ncase; i-->0; ) {
 738                 c = sel->lockorder[i];
 739                 sel->lockorder[i] = sel->lockorder[0];
 740                 j = 0;
 741                 for(;;) {
 742                         k = j*2+1;
 743                         if(k >= i)
 744                                 break;
 745                         if(k+1 < i && sel->lockorder[k] < sel->lockorder[k+1])
 746                                 k++;
 747                         if(c < sel->lockorder[k]) {
 748                                 sel->lockorder[j] = sel->lockorder[k];
 749                                 j = k;
 750                                 continue;
 751                         }
 752                         break;
 753                 }
 754                 sel->lockorder[j] = c;
 755         }
 756         /*
 757         for(i=0; i+1<sel->ncase; i++)
 758                 if(sel->lockorder[i] > sel->lockorder[i+1]) {
 759                         runtime_printf("i=%d %p %p\n", i, sel->lockorder[i], sel->lockorder[i+1]);
 760                         runtime_throw("select: broken sort");
 761                 }
 762         */
 763         sellock(sel);
 764
 765 loop:
 766         // pass 1 - look for something already waiting
 767         dfl = nil;
 768         for(i=0; i<sel->ncase; i++) {
 769                 o = sel->pollorder[i];
 770                 cas = &sel->scase[o];
 771                 c = cas->chan;
 772
 773                 switch(cas->kind) {
 774                 case CaseRecv:
 775                         if(c->dataqsiz > 0) {
 776                                 if(c->qcount > 0)
 777                                         goto asyncrecv;
 778                         } else {
 779                                 sg = dequeue(&c->sendq);
 780                                 if(sg != nil)
 781                                         goto syncrecv;
 782                         }
 783                         if(c->closed)
 784                                 goto rclose;
 785                         break;
 786
 787                 case CaseSend:
 788                         if(raceenabled)
 789                                 runtime_racereadpc(c, runtime_selectgo, chansend);
 790                         if(c->closed)
 791                                 goto sclose;
 792                         if(c->dataqsiz > 0) {
 793                                 if(c->qcount < c->dataqsiz)
 794                                         goto asyncsend;
 795                         } else {
 796                                 sg = dequeue(&c->recvq);
 797                                 if(sg != nil)
 798                                         goto syncsend;
 799                         }
 800                         break;
 801
 802                 case CaseDefault:
 803                         dfl = cas;
 804                         break;
 805                 }
 806         }
 807
 808         if(dfl != nil) {
 809                 selunlock(sel);
 810                 cas = dfl;
 811                 goto retc;
 812         }
 813
 814
 815         // pass 2 - enqueue on all chans
 816         done = 0;
 817         for(i=0; i<sel->ncase; i++) {
 818                 o = sel->pollorder[i];
 819                 cas = &sel->scase[o];
 820                 c = cas->chan;
 821                 sg = &cas->sg;
 822                 sg->g = g;
 823                 sg->selectdone = &done;
 824
 825                 switch(cas->kind) {
 826                 case CaseRecv:
 827                         enqueue(&c->recvq, sg);
 828                         break;
 829
 830                 case CaseSend:
 831                         enqueue(&c->sendq, sg);
 832                         break;
 833                 }
 834         }
 835
 836         g->param = nil;
 837         runtime_park(selparkcommit, sel, "select");
 838
 839         sellock(sel);
 840         sg = g->param;
 841
 842         // pass 3 - dequeue from unsuccessful chans
 843         // otherwise they stack up on quiet channels
 844         for(i=0; i<sel->ncase; i++) {
 845                 cas = &sel->scase[i];
 846                 if(cas != (Scase*)sg) {
 847                         c = cas->chan;
 848                         if(cas->kind == CaseSend)
 849                                 dequeueg(&c->sendq);
 850                         else
 851                                 dequeueg(&c->recvq);
 852                 }
 853         }
 854
 855         if(sg == nil)
 856                 goto loop;
 857
 858         cas = (Scase*)sg;
 859         c = cas->chan;
 860
 861         if(c->dataqsiz > 0)
 862                 runtime_throw("selectgo: shouldn't happen");
 863
 864         if(debug)
 865                 runtime_printf("wait-return: sel=%p c=%p cas=%p kind=%d\n",
 866                         sel, c, cas, cas->kind);
 867
 868         if(cas->kind == CaseRecv) {
 869                 if(cas->receivedp != nil)
 870                         *cas->receivedp = true;
 871         }
 872
 873         if(raceenabled) {
 874                 if(cas->kind == CaseRecv && cas->sg.elem != nil)
 875                         runtime_racewriteobjectpc(cas->sg.elem, c->elemtype, selectgo, chanrecv);
 876                 else if(cas->kind == CaseSend)
 877                         runtime_racereadobjectpc(cas->sg.elem, c->elemtype, selectgo, chansend);
 878         }
 879
 880         selunlock(sel);
 881         goto retc;
 882
 883 asyncrecv:
 884         // can receive from buffer
 885         if(raceenabled) {
 886                 if(cas->sg.elem != nil)
 887                         runtime_racewriteobjectpc(cas->sg.elem, c->elemtype, selectgo, chanrecv);
 888                 runtime_raceacquire(chanbuf(c, c->recvx));
 889         }
 890         if(cas->receivedp != nil)
 891                 *cas->receivedp = true;
 892         if(cas->sg.elem != nil)
 893                 runtime_memmove(cas->sg.elem, chanbuf(c, c->recvx), c->elemsize);
 894         runtime_memclr(chanbuf(c, c->recvx), c->elemsize);
 895         if(++c->recvx == c->dataqsiz)
 896                 c->recvx = 0;
 897         c->qcount--;
 898         sg = dequeue(&c->sendq);
 899         if(sg != nil) {
 900                 gp = sg->g;
 901                 selunlock(sel);
 902                 if(sg->releasetime)
 903                         sg->releasetime = runtime_cputicks();
 904                 runtime_ready(gp);
 905         } else {
 906                 selunlock(sel);
 907         }
 908         goto retc;
 909
 910 asyncsend:
 911         // can send to buffer
 912         if(raceenabled) {
 913                 runtime_racerelease(chanbuf(c, c->sendx));
 914                 runtime_racereadobjectpc(cas->sg.elem, c->elemtype, selectgo, chansend);
 915         }
 916         runtime_memmove(chanbuf(c, c->sendx), cas->sg.elem, c->elemsize);
 917         if(++c->sendx == c->dataqsiz)
 918                 c->sendx = 0;
 919         c->qcount++;
 920         sg = dequeue(&c->recvq);
 921         if(sg != nil) {
 922                 gp = sg->g;
 923                 selunlock(sel);
 924                 if(sg->releasetime)
 925                         sg->releasetime = runtime_cputicks();
 926                 runtime_ready(gp);
 927         } else {
 928                 selunlock(sel);
 929         }
 930         goto retc;
 931
 932 syncrecv:
 933         // can receive from sleeping sender (sg)
 934         if(raceenabled) {
 935                 if(cas->sg.elem != nil)
 936                         runtime_racewriteobjectpc(cas->sg.elem, c->elemtype, selectgo, chanrecv);
 937                 racesync(c, sg);
 938         }
 939         selunlock(sel);
 940         if(debug)
 941                 runtime_printf("syncrecv: sel=%p c=%p o=%d\n", sel, c, o);
 942         if(cas->receivedp != nil)
 943                 *cas->receivedp = true;
 944         if(cas->sg.elem != nil)
 945                 runtime_memmove(cas->sg.elem, sg->elem, c->elemsize);
 946         gp = sg->g;
 947         gp->param = sg;
 948         if(sg->releasetime)
 949                 sg->releasetime = runtime_cputicks();
 950         runtime_ready(gp);
 951         goto retc;
 952
 953 rclose:
 954         // read at end of closed channel
 955         selunlock(sel);
 956         if(cas->receivedp != nil)
 957                 *cas->receivedp = false;
 958         if(cas->sg.elem != nil)
 959                 runtime_memclr(cas->sg.elem, c->elemsize);
 960         if(raceenabled)
 961                 runtime_raceacquire(c);
 962         goto retc;
 963
 964 syncsend:
 965         // can send to sleeping receiver (sg)
 966         if(raceenabled) {
 967                 runtime_racereadobjectpc(cas->sg.elem, c->elemtype, selectgo, chansend);
 968                 racesync(c, sg);
 969         }
 970         selunlock(sel);
 971         if(debug)
 972                 runtime_printf("syncsend: sel=%p c=%p o=%d\n", sel, c, o);
 973         if(sg->elem != nil)
 974                 runtime_memmove(sg->elem, cas->sg.elem, c->elemsize);
 975         gp = sg->g;
 976         gp->param = sg;
 977         if(sg->releasetime)
 978                 sg->releasetime = runtime_cputicks();
 979         runtime_ready(gp);
 980
 981 retc:
 982         // return index corresponding to chosen case
 983         index = cas->index;
 984         if(cas->sg.releasetime > 0)
 985                 runtime_blockevent(cas->sg.releasetime - t0, 2);
 986         runtime_free(sel);
 987         return index;
 988
 989 sclose:
 990         // send on closed channel
 991         selunlock(sel);
 992         runtime_panicstring("send on closed channel");
 993         return 0;  // not reached
 994 }
 995
 996 // This struct must match ../reflect/value.go:/runtimeSelect.
 997 typedef struct runtimeSelect runtimeSelect;
 998 struct runtimeSelect
 999 {
1000         uintptr dir;
1001         ChanType *typ;
1002         Hchan *ch;
1003         byte *val;
1004 };
1005
1006 // This enum must match ../reflect/value.go:/SelectDir.
1007 enum SelectDir {
1008         SelectSend = 1,
1009         SelectRecv,
1010         SelectDefault,
1011 };
1012
1013 func reflect.rselect(cases Slice) (chosen int, recvOK bool) {
1014         int32 i;
1015         Select *sel;
1016         runtimeSelect* rcase, *rc;
1017
1018         chosen = -1;
1019         recvOK = false;
1020
1021         rcase = (runtimeSelect*)cases.__values;
1022
1023         sel = newselect(cases.__count);
1024         for(i=0; i<cases.__count; i++) {
1025                 rc = &rcase[i];
1026                 switch(rc->dir) {
1027                 case SelectDefault:
1028                         selectdefault(sel, i);
1029                         break;
1030                 case SelectSend:
1031                         if(rc->ch == nil)
1032                                 break;
1033                         selectsend(sel, rc->ch, i, rc->val);
1034                         break;
1035                 case SelectRecv:
1036                         if(rc->ch == nil)
1037                                 break;
1038                         selectrecv(sel, rc->ch, i, rc->val, &recvOK);
1039                         break;
1040                 }
1041         }
1042
1043         chosen = (intgo)(uintptr)selectgo(&sel);
1044 }
1045
1046 static void closechan(Hchan *c, void *pc);
1047
1048 func closechan(c *Hchan) {
1049         closechan(c, runtime_getcallerpc(&c));
1050 }
1051
1052 func reflect.chanclose(c *Hchan) {
1053         closechan(c, runtime_getcallerpc(&c));
1054 }
1055
1056 static void
1057 closechan(Hchan *c, void *pc)
1058 {
1059         SudoG *sg;
1060         G* gp;
1061
1062         if(c == nil)
1063                 runtime_panicstring("close of nil channel");
1064
1065         if(runtime_gcwaiting())
1066                 runtime_gosched();
1067
1068         runtime_lock(c);
1069         if(c->closed) {
1070                 runtime_unlock(c);
1071                 runtime_panicstring("close of closed channel");
1072         }
1073
1074         if(raceenabled) {
1075                 runtime_racewritepc(c, pc, runtime_closechan);
1076                 runtime_racerelease(c);
1077         }
1078
1079         c->closed = true;
1080
1081         // release all readers
1082         for(;;) {
1083                 sg = dequeue(&c->recvq);
1084                 if(sg == nil)
1085                         break;
1086                 gp = sg->g;
1087                 gp->param = nil;
1088                 if(sg->releasetime)
1089                         sg->releasetime = runtime_cputicks();
1090                 runtime_ready(gp);
1091         }
1092
1093         // release all writers
1094         for(;;) {
1095                 sg = dequeue(&c->sendq);
1096                 if(sg == nil)
1097                         break;
1098                 gp = sg->g;
1099                 gp->param = nil;
1100                 if(sg->releasetime)
1101                         sg->releasetime = runtime_cputicks();
1102                 runtime_ready(gp);
1103         }
1104
1105         runtime_unlock(c);
1106 }
1107
1108 void
1109 __go_builtin_close(Hchan *c)
1110 {
1111         runtime_closechan(c);
1112 }
1113
1114 func reflect.chanlen(c *Hchan) (len int) {
1115         if(c == nil)
1116                 len = 0;
1117         else
1118                 len = c->qcount;
1119 }
1120
1121 intgo
1122 __go_chan_len(Hchan *c)
1123 {
1124         return reflect_chanlen(c);
1125 }
1126
1127 func reflect.chancap(c *Hchan) (cap int) {
1128         if(c == nil)
1129                 cap = 0;
1130         else
1131                 cap = c->dataqsiz;
1132 }
1133
1134 intgo
1135 __go_chan_cap(Hchan *c)
1136 {
1137         return reflect_chancap(c);
1138 }
1139
1140 static SudoG*
1141 dequeue(WaitQ *q)
1142 {
1143         SudoG *sgp;
1144
1145 loop:
1146         sgp = q->first;
1147         if(sgp == nil)
1148                 return nil;
1149         q->first = sgp->link;
1150
1151         // if sgp participates in a select and is already signaled, ignore it
1152         if(sgp->selectdone != nil) {
1153                 // claim the right to signal
1154                 if(*sgp->selectdone != 0 || !runtime_cas(sgp->selectdone, 0, 1))
1155                         goto loop;
1156         }
1157
1158         return sgp;
1159 }
1160
1161 static void
1162 dequeueg(WaitQ *q)
1163 {
1164         SudoG **l, *sgp, *prevsgp;
1165         G *g;
1166
1167         g = runtime_g();
1168         prevsgp = nil;
1169         for(l=&q->first; (sgp=*l) != nil; l=&sgp->link, prevsgp=sgp) {
1170                 if(sgp->g == g) {
1171                         *l = sgp->link;
1172                         if(q->last == sgp)
1173                                 q->last = prevsgp;
1174                         break;
1175                 }
1176         }
1177 }
1178
1179 static void
1180 enqueue(WaitQ *q, SudoG *sgp)
1181 {
1182         sgp->link = nil;
1183         if(q->first == nil) {
1184                 q->first = sgp;
1185                 q->last = sgp;
1186                 return;
1187         }
1188         q->last->link = sgp;
1189         q->last = sgp;
1190 }
1191
1192 static void
1193 racesync(Hchan *c, SudoG *sg)
1194 {
1195         runtime_racerelease(chanbuf(c, 0));
1196         runtime_raceacquireg(sg->g, chanbuf(c, 0));
1197         runtime_racereleaseg(sg->g, chanbuf(c, 0));
1198         runtime_raceacquire(chanbuf(c, 0));
1199 }