src/runtime/select.go

   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
   7 // This file contains the implementation of Go select statements.
   8
   9 import (
  10         "internal/abi"
  11         "unsafe"
  12 )
  13
  14 const debugSelect = false
  15
  16 // Select case descriptor.
  17 // Known to compiler.
  18 // Changes here must also be made in src/cmd/compile/internal/walk/select.go's scasetype.
  19 type scase struct {
  20         c    *hchan         // chan
  21         elem unsafe.Pointer // data element
  22 }
  23
  24 var (
  25         chansendpc = abi.FuncPCABIInternal(chansend)
  26         chanrecvpc = abi.FuncPCABIInternal(chanrecv)
  27 )
  28
  29 func selectsetpc(pc *uintptr) {
  30         *pc = getcallerpc()
  31 }
  32
  33 func sellock(scases []scase, lockorder []uint16) {
  34         var c *hchan
  35         for _, o := range lockorder {
  36                 c0 := scases[o].c
  37                 if c0 != c {
  38                         c = c0
  39                         lock(&c.lock)
  40                 }
  41         }
  42 }
  43
  44 func selunlock(scases []scase, lockorder []uint16) {
  45         // We must be very careful here to not touch sel after we have unlocked
  46         // the last lock, because sel can be freed right after the last unlock.
  47         // Consider the following situation.
  48         // First M calls runtime·park() in runtime·selectgo() passing the sel.
  49         // Once runtime·park() has unlocked the last lock, another M makes
  50         // the G that calls select runnable again and schedules it for execution.
  51         // When the G runs on another M, it locks all the locks and frees sel.
  52         // Now if the first M touches sel, it will access freed memory.
  53         for i := len(lockorder) - 1; i >= 0; i-- {
  54                 c := scases[lockorder[i]].c
  55                 if i > 0 && c == scases[lockorder[i-1]].c {
  56                         continue // will unlock it on the next iteration
  57                 }
  58                 unlock(&c.lock)
  59         }
  60 }
  61
  62 func selparkcommit(gp *g, _ unsafe.Pointer) bool {
  63         // There are unlocked sudogs that point into gp's stack. Stack
  64         // copying must lock the channels of those sudogs.
  65         // Set activeStackChans here instead of before we try parking
  66         // because we could self-deadlock in stack growth on a
  67         // channel lock.
  68         gp.activeStackChans = true
  69         // Mark that it's safe for stack shrinking to occur now,
  70         // because any thread acquiring this G's stack for shrinking
  71         // is guaranteed to observe activeStackChans after this store.
  72         gp.parkingOnChan.Store(false)
  73         // Make sure we unlock after setting activeStackChans and
  74         // unsetting parkingOnChan. The moment we unlock any of the
  75         // channel locks we risk gp getting readied by a channel operation
  76         // and so gp could continue running before everything before the
  77         // unlock is visible (even to gp itself).
  78
  79         // This must not access gp's stack (see gopark). In
  80         // particular, it must not access the *hselect. That's okay,
  81         // because by the time this is called, gp.waiting has all
  82         // channels in lock order.
  83         var lastc *hchan
  84         for sg := gp.waiting; sg != nil; sg = sg.waitlink {
  85                 if sg.c != lastc && lastc != nil {
  86                         // As soon as we unlock the channel, fields in
  87                         // any sudog with that channel may change,
  88                         // including c and waitlink. Since multiple
  89                         // sudogs may have the same channel, we unlock
  90                         // only after we've passed the last instance
  91                         // of a channel.
  92                         unlock(&lastc.lock)
  93                 }
  94                 lastc = sg.c
  95         }
  96         if lastc != nil {
  97                 unlock(&lastc.lock)
  98         }
  99         return true
 100 }
 101
 102 func block() {
 103         gopark(nil, nil, waitReasonSelectNoCases, traceBlockForever, 1) // forever
 104 }
 105
 106 // selectgo implements the select statement.
 107 //
 108 // cas0 points to an array of type [ncases]scase, and order0 points to
 109 // an array of type [2*ncases]uint16 where ncases must be <= 65536.
 110 // Both reside on the goroutine's stack (regardless of any escaping in
 111 // selectgo).
 112 //
 113 // For race detector builds, pc0 points to an array of type
 114 // [ncases]uintptr (also on the stack); for other builds, it's set to
 115 // nil.
 116 //
 117 // selectgo returns the index of the chosen scase, which matches the
 118 // ordinal position of its respective select{recv,send,default} call.
 119 // Also, if the chosen scase was a receive operation, it reports whether
 120 // a value was received.
 121 func selectgo(cas0 *scase, order0 *uint16, pc0 *uintptr, nsends, nrecvs int, block bool) (int, bool) {
 122         if debugSelect {
 123                 print("select: cas0=", cas0, "\n")
 124         }
 125
 126         // NOTE: In order to maintain a lean stack size, the number of scases
 127         // is capped at 65536.
 128         cas1 := (*[1 << 16]scase)(unsafe.Pointer(cas0))
 129         order1 := (*[1 << 17]uint16)(unsafe.Pointer(order0))
 130
 131         ncases := nsends + nrecvs
 132         scases := cas1[:ncases:ncases]
 133         pollorder := order1[:ncases:ncases]
 134         lockorder := order1[ncases:][:ncases:ncases]
 135         // NOTE: pollorder/lockorder's underlying array was not zero-initialized by compiler.
 136
 137         // Even when raceenabled is true, there might be select
 138         // statements in packages compiled without -race (e.g.,
 139         // ensureSigM in runtime/signal_unix.go).
 140         var pcs []uintptr
 141         if raceenabled && pc0 != nil {
 142                 pc1 := (*[1 << 16]uintptr)(unsafe.Pointer(pc0))
 143                 pcs = pc1[:ncases:ncases]
 144         }
 145         casePC := func(casi int) uintptr {
 146                 if pcs == nil {
 147                         return 0
 148                 }
 149                 return pcs[casi]
 150         }
 151
 152         var t0 int64
 153         if blockprofilerate > 0 {
 154                 t0 = cputicks()
 155         }
 156
 157         // The compiler rewrites selects that statically have
 158         // only 0 or 1 cases plus default into simpler constructs.
 159         // The only way we can end up with such small sel.ncase
 160         // values here is for a larger select in which most channels
 161         // have been nilled out. The general code handles those
 162         // cases correctly, and they are rare enough not to bother
 163         // optimizing (and needing to test).
 164
 165         // generate permuted order
 166         norder := 0
 167         for i := range scases {
 168                 cas := &scases[i]
 169
 170                 // Omit cases without channels from the poll and lock orders.
 171                 if cas.c == nil {
 172                         cas.elem = nil // allow GC
 173                         continue
 174                 }
 175
 176                 j := fastrandn(uint32(norder + 1))
 177                 pollorder[norder] = pollorder[j]
 178                 pollorder[j] = uint16(i)
 179                 norder++
 180         }
 181         pollorder = pollorder[:norder]
 182         lockorder = lockorder[:norder]
 183
 184         // sort the cases by Hchan address to get the locking order.
 185         // simple heap sort, to guarantee n log n time and constant stack footprint.
 186         for i := range lockorder {
 187                 j := i
 188                 // Start with the pollorder to permute cases on the same channel.
 189                 c := scases[pollorder[i]].c
 190                 for j > 0 && scases[lockorder[(j-1)/2]].c.sortkey() < c.sortkey() {
 191                         k := (j - 1) / 2
 192                         lockorder[j] = lockorder[k]
 193                         j = k
 194                 }
 195                 lockorder[j] = pollorder[i]
 196         }
 197         for i := len(lockorder) - 1; i >= 0; i-- {
 198                 o := lockorder[i]
 199                 c := scases[o].c
 200                 lockorder[i] = lockorder[0]
 201                 j := 0
 202                 for {
 203                         k := j*2 + 1
 204                         if k >= i {
 205                                 break
 206                         }
 207                         if k+1 < i && scases[lockorder[k]].c.sortkey() < scases[lockorder[k+1]].c.sortkey() {
 208                                 k++
 209                         }
 210                         if c.sortkey() < scases[lockorder[k]].c.sortkey() {
 211                                 lockorder[j] = lockorder[k]
 212                                 j = k
 213                                 continue
 214                         }
 215                         break
 216                 }
 217                 lockorder[j] = o
 218         }
 219
 220         if debugSelect {
 221                 for i := 0; i+1 < len(lockorder); i++ {
 222                         if scases[lockorder[i]].c.sortkey() > scases[lockorder[i+1]].c.sortkey() {
 223                                 print("i=", i, " x=", lockorder[i], " y=", lockorder[i+1], "\n")
 224                                 throw("select: broken sort")
 225                         }
 226                 }
 227         }
 228
 229         // lock all the channels involved in the select
 230         sellock(scases, lockorder)
 231
 232         var (
 233                 gp     *g
 234                 sg     *sudog
 235                 c      *hchan
 236                 k      *scase
 237                 sglist *sudog
 238                 sgnext *sudog
 239                 qp     unsafe.Pointer
 240                 nextp  **sudog
 241         )
 242
 243         // pass 1 - look for something already waiting
 244         var casi int
 245         var cas *scase
 246         var caseSuccess bool
 247         var caseReleaseTime int64 = -1
 248         var recvOK bool
 249         for _, casei := range pollorder {
 250                 casi = int(casei)
 251                 cas = &scases[casi]
 252                 c = cas.c
 253
 254                 if casi >= nsends {
 255                         sg = c.sendq.dequeue()
 256                         if sg != nil {
 257                                 goto recv
 258                         }
 259                         if c.qcount > 0 {
 260                                 goto bufrecv
 261                         }
 262                         if c.closed != 0 {
 263                                 goto rclose
 264                         }
 265                 } else {
 266                         if raceenabled {
 267                                 racereadpc(c.raceaddr(), casePC(casi), chansendpc)
 268                         }
 269                         if c.closed != 0 {
 270                                 goto sclose
 271                         }
 272                         sg = c.recvq.dequeue()
 273                         if sg != nil {
 274                                 goto send
 275                         }
 276                         if c.qcount < c.dataqsiz {
 277                                 goto bufsend
 278                         }
 279                 }
 280         }
 281
 282         if !block {
 283                 selunlock(scases, lockorder)
 284                 casi = -1
 285                 goto retc
 286         }
 287
 288         // pass 2 - enqueue on all chans
 289         gp = getg()
 290         if gp.waiting != nil {
 291                 throw("gp.waiting != nil")
 292         }
 293         nextp = &gp.waiting
 294         for _, casei := range lockorder {
 295                 casi = int(casei)
 296                 cas = &scases[casi]
 297                 c = cas.c
 298                 sg := acquireSudog()
 299                 sg.g = gp
 300                 sg.isSelect = true
 301                 // No stack splits between assigning elem and enqueuing
 302                 // sg on gp.waiting where copystack can find it.
 303                 sg.elem = cas.elem
 304                 sg.releasetime = 0
 305                 if t0 != 0 {
 306                         sg.releasetime = -1
 307                 }
 308                 sg.c = c
 309                 // Construct waiting list in lock order.
 310                 *nextp = sg
 311                 nextp = &sg.waitlink
 312
 313                 if casi < nsends {
 314                         c.sendq.enqueue(sg)
 315                 } else {
 316                         c.recvq.enqueue(sg)
 317                 }
 318         }
 319
 320         // wait for someone to wake us up
 321         gp.param = nil
 322         // Signal to anyone trying to shrink our stack that we're about
 323         // to park on a channel. The window between when this G's status
 324         // changes and when we set gp.activeStackChans is not safe for
 325         // stack shrinking.
 326         gp.parkingOnChan.Store(true)
 327         gopark(selparkcommit, nil, waitReasonSelect, traceBlockSelect, 1)
 328         gp.activeStackChans = false
 329
 330         sellock(scases, lockorder)
 331
 332         gp.selectDone.Store(0)
 333         sg = (*sudog)(gp.param)
 334         gp.param = nil
 335
 336         // pass 3 - dequeue from unsuccessful chans
 337         // otherwise they stack up on quiet channels
 338         // record the successful case, if any.
 339         // We singly-linked up the SudoGs in lock order.
 340         casi = -1
 341         cas = nil
 342         caseSuccess = false
 343         sglist = gp.waiting
 344         // Clear all elem before unlinking from gp.waiting.
 345         for sg1 := gp.waiting; sg1 != nil; sg1 = sg1.waitlink {
 346                 sg1.isSelect = false
 347                 sg1.elem = nil
 348                 sg1.c = nil
 349         }
 350         gp.waiting = nil
 351
 352         for _, casei := range lockorder {
 353                 k = &scases[casei]
 354                 if sg == sglist {
 355                         // sg has already been dequeued by the G that woke us up.
 356                         casi = int(casei)
 357                         cas = k
 358                         caseSuccess = sglist.success
 359                         if sglist.releasetime > 0 {
 360                                 caseReleaseTime = sglist.releasetime
 361                         }
 362                 } else {
 363                         c = k.c
 364                         if int(casei) < nsends {
 365                                 c.sendq.dequeueSudoG(sglist)
 366                         } else {
 367                                 c.recvq.dequeueSudoG(sglist)
 368                         }
 369                 }
 370                 sgnext = sglist.waitlink
 371                 sglist.waitlink = nil
 372                 releaseSudog(sglist)
 373                 sglist = sgnext
 374         }
 375
 376         if cas == nil {
 377                 throw("selectgo: bad wakeup")
 378         }
 379
 380         c = cas.c
 381
 382         if debugSelect {
 383                 print("wait-return: cas0=", cas0, " c=", c, " cas=", cas, " send=", casi < nsends, "\n")
 384         }
 385
 386         if casi < nsends {
 387                 if !caseSuccess {
 388                         goto sclose
 389                 }
 390         } else {
 391                 recvOK = caseSuccess
 392         }
 393
 394         if raceenabled {
 395                 if casi < nsends {
 396                         raceReadObjectPC(c.elemtype, cas.elem, casePC(casi), chansendpc)
 397                 } else if cas.elem != nil {
 398                         raceWriteObjectPC(c.elemtype, cas.elem, casePC(casi), chanrecvpc)
 399                 }
 400         }
 401         if msanenabled {
 402                 if casi < nsends {
 403                         msanread(cas.elem, c.elemtype.Size_)
 404                 } else if cas.elem != nil {
 405                         msanwrite(cas.elem, c.elemtype.Size_)
 406                 }
 407         }
 408         if asanenabled {
 409                 if casi < nsends {
 410                         asanread(cas.elem, c.elemtype.Size_)
 411                 } else if cas.elem != nil {
 412                         asanwrite(cas.elem, c.elemtype.Size_)
 413                 }
 414         }
 415
 416         selunlock(scases, lockorder)
 417         goto retc
 418
 419 bufrecv:
 420         // can receive from buffer
 421         if raceenabled {
 422                 if cas.elem != nil {
 423                         raceWriteObjectPC(c.elemtype, cas.elem, casePC(casi), chanrecvpc)
 424                 }
 425                 racenotify(c, c.recvx, nil)
 426         }
 427         if msanenabled && cas.elem != nil {
 428                 msanwrite(cas.elem, c.elemtype.Size_)
 429         }
 430         if asanenabled && cas.elem != nil {
 431                 asanwrite(cas.elem, c.elemtype.Size_)
 432         }
 433         recvOK = true
 434         qp = chanbuf(c, c.recvx)
 435         if cas.elem != nil {
 436                 typedmemmove(c.elemtype, cas.elem, qp)
 437         }
 438         typedmemclr(c.elemtype, qp)
 439         c.recvx++
 440         if c.recvx == c.dataqsiz {
 441                 c.recvx = 0
 442         }
 443         c.qcount--
 444         selunlock(scases, lockorder)
 445         goto retc
 446
 447 bufsend:
 448         // can send to buffer
 449         if raceenabled {
 450                 racenotify(c, c.sendx, nil)
 451                 raceReadObjectPC(c.elemtype, cas.elem, casePC(casi), chansendpc)
 452         }
 453         if msanenabled {
 454                 msanread(cas.elem, c.elemtype.Size_)
 455         }
 456         if asanenabled {
 457                 asanread(cas.elem, c.elemtype.Size_)
 458         }
 459         typedmemmove(c.elemtype, chanbuf(c, c.sendx), cas.elem)
 460         c.sendx++
 461         if c.sendx == c.dataqsiz {
 462                 c.sendx = 0
 463         }
 464         c.qcount++
 465         selunlock(scases, lockorder)
 466         goto retc
 467
 468 recv:
 469         // can receive from sleeping sender (sg)
 470         recv(c, sg, cas.elem, func() { selunlock(scases, lockorder) }, 2)
 471         if debugSelect {
 472                 print("syncrecv: cas0=", cas0, " c=", c, "\n")
 473         }
 474         recvOK = true
 475         goto retc
 476
 477 rclose:
 478         // read at end of closed channel
 479         selunlock(scases, lockorder)
 480         recvOK = false
 481         if cas.elem != nil {
 482                 typedmemclr(c.elemtype, cas.elem)
 483         }
 484         if raceenabled {
 485                 raceacquire(c.raceaddr())
 486         }
 487         goto retc
 488
 489 send:
 490         // can send to a sleeping receiver (sg)
 491         if raceenabled {
 492                 raceReadObjectPC(c.elemtype, cas.elem, casePC(casi), chansendpc)
 493         }
 494         if msanenabled {
 495                 msanread(cas.elem, c.elemtype.Size_)
 496         }
 497         if asanenabled {
 498                 asanread(cas.elem, c.elemtype.Size_)
 499         }
 500         send(c, sg, cas.elem, func() { selunlock(scases, lockorder) }, 2)
 501         if debugSelect {
 502                 print("syncsend: cas0=", cas0, " c=", c, "\n")
 503         }
 504         goto retc
 505
 506 retc:
 507         if caseReleaseTime > 0 {
 508                 blockevent(caseReleaseTime-t0, 1)
 509         }
 510         return casi, recvOK
 511
 512 sclose:
 513         // send on closed channel
 514         selunlock(scases, lockorder)
 515         panic(plainError("send on closed channel"))
 516 }
 517
 518 func (c *hchan) sortkey() uintptr {
 519         return uintptr(unsafe.Pointer(c))
 520 }
 521
 522 // A runtimeSelect is a single case passed to rselect.
 523 // This must match ../reflect/value.go:/runtimeSelect
 524 type runtimeSelect struct {
 525         dir selectDir
 526         typ unsafe.Pointer // channel type (not used here)
 527         ch  *hchan         // channel
 528         val unsafe.Pointer // ptr to data (SendDir) or ptr to receive buffer (RecvDir)
 529 }
 530
 531 // These values must match ../reflect/value.go:/SelectDir.
 532 type selectDir int
 533
 534 const (
 535         _             selectDir = iota
 536         selectSend              // case Chan <- Send
 537         selectRecv              // case <-Chan:
 538         selectDefault           // default
 539 )
 540
 541 //go:linkname reflect_rselect reflect.rselect
 542 func reflect_rselect(cases []runtimeSelect) (int, bool) {
 543         if len(cases) == 0 {
 544                 block()
 545         }
 546         sel := make([]scase, len(cases))
 547         orig := make([]int, len(cases))
 548         nsends, nrecvs := 0, 0
 549         dflt := -1
 550         for i, rc := range cases {
 551                 var j int
 552                 switch rc.dir {
 553                 case selectDefault:
 554                         dflt = i
 555                         continue
 556                 case selectSend:
 557                         j = nsends
 558                         nsends++
 559                 case selectRecv:
 560                         nrecvs++
 561                         j = len(cases) - nrecvs
 562                 }
 563
 564                 sel[j] = scase{c: rc.ch, elem: rc.val}
 565                 orig[j] = i
 566         }
 567
 568         // Only a default case.
 569         if nsends+nrecvs == 0 {
 570                 return dflt, false
 571         }
 572
 573         // Compact sel and orig if necessary.
 574         if nsends+nrecvs < len(cases) {
 575                 copy(sel[nsends:], sel[len(cases)-nrecvs:])
 576                 copy(orig[nsends:], orig[len(cases)-nrecvs:])
 577         }
 578
 579         order := make([]uint16, 2*(nsends+nrecvs))
 580         var pc0 *uintptr
 581         if raceenabled {
 582                 pcs := make([]uintptr, nsends+nrecvs)
 583                 for i := range pcs {
 584                         selectsetpc(&pcs[i])
 585                 }
 586                 pc0 = &pcs[0]
 587         }
 588
 589         chosen, recvOK := selectgo(&sel[0], &order[0], pc0, nsends, nrecvs, dflt == -1)
 590
 591         // Translate chosen back to caller's ordering.
 592         if chosen < 0 {
 593                 chosen = dflt
 594         } else {
 595                 chosen = orig[chosen]
 596         }
 597         return chosen, recvOK
 598 }
 599
 600 func (q *waitq) dequeueSudoG(sgp *sudog) {
 601         x := sgp.prev
 602         y := sgp.next
 603         if x != nil {
 604                 if y != nil {
 605                         // middle of queue
 606                         x.next = y
 607                         y.prev = x
 608                         sgp.next = nil
 609                         sgp.prev = nil
 610                         return
 611                 }
 612                 // end of queue
 613                 x.next = nil
 614                 q.last = x
 615                 sgp.prev = nil
 616                 return
 617         }
 618         if y != nil {
 619                 // start of queue
 620                 y.prev = nil
 621                 q.first = y
 622                 sgp.next = nil
 623                 return
 624         }
 625
 626         // x==y==nil. Either sgp is the only element in the queue,
 627         // or it has already been removed. Use q.first to disambiguate.
 628         if q.first == sgp {
 629                 q.first = nil
 630                 q.last = nil
 631         }
 632 }