src/runtime/sema.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 // Semaphore implementation exposed to Go.
   6 // Intended use is provide a sleep and wakeup
   7 // primitive that can be used in the contended case
   8 // of other synchronization primitives.
   9 // Thus it targets the same goal as Linux's futex,
  10 // but it has much simpler semantics.
  11 //
  12 // That is, don't think of these as semaphores.
  13 // Think of them as a way to implement sleep and wakeup
  14 // such that every sleep is paired with a single wakeup,
  15 // even if, due to races, the wakeup happens before the sleep.
  16 //
  17 // See Mullender and Cox, ``Semaphores in Plan 9,''
  18 // http://swtch.com/semaphore.pdf
  19
  20 package runtime
  21
  22 import (
  23         "runtime/internal/atomic"
  24         "unsafe"
  25 )
  26
  27 // Asynchronous semaphore for sync.Mutex.
  28
  29 type semaRoot struct {
  30         lock  mutex
  31         head  *sudog
  32         tail  *sudog
  33         nwait uint32 // Number of waiters. Read w/o the lock.
  34 }
  35
  36 // Prime to not correlate with any user patterns.
  37 const semTabSize = 251
  38
  39 var semtable [semTabSize]struct {
  40         root semaRoot
  41         pad  [_CacheLineSize - unsafe.Sizeof(semaRoot{})]byte
  42 }
  43
  44 //go:linkname sync_runtime_Semacquire sync.runtime_Semacquire
  45 func sync_runtime_Semacquire(addr *uint32) {
  46         semacquire(addr, true)
  47 }
  48
  49 //go:linkname net_runtime_Semacquire net.runtime_Semacquire
  50 func net_runtime_Semacquire(addr *uint32) {
  51         semacquire(addr, true)
  52 }
  53
  54 //go:linkname sync_runtime_Semrelease sync.runtime_Semrelease
  55 func sync_runtime_Semrelease(addr *uint32) {
  56         semrelease(addr)
  57 }
  58
  59 //go:linkname net_runtime_Semrelease net.runtime_Semrelease
  60 func net_runtime_Semrelease(addr *uint32) {
  61         semrelease(addr)
  62 }
  63
  64 // Called from runtime.
  65 func semacquire(addr *uint32, profile bool) {
  66         gp := getg()
  67         if gp != gp.m.curg {
  68                 throw("semacquire not on the G stack")
  69         }
  70
  71         // Easy case.
  72         if cansemacquire(addr) {
  73                 return
  74         }
  75
  76         // Harder case:
  77         //      increment waiter count
  78         //      try cansemacquire one more time, return if succeeded
  79         //      enqueue itself as a waiter
  80         //      sleep
  81         //      (waiter descriptor is dequeued by signaler)
  82         s := acquireSudog()
  83         root := semroot(addr)
  84         t0 := int64(0)
  85         s.releasetime = 0
  86         if profile && blockprofilerate > 0 {
  87                 t0 = cputicks()
  88                 s.releasetime = -1
  89         }
  90         for {
  91                 lock(&root.lock)
  92                 // Add ourselves to nwait to disable "easy case" in semrelease.
  93                 atomic.Xadd(&root.nwait, 1)
  94                 // Check cansemacquire to avoid missed wakeup.
  95                 if cansemacquire(addr) {
  96                         atomic.Xadd(&root.nwait, -1)
  97                         unlock(&root.lock)
  98                         break
  99                 }
 100                 // Any semrelease after the cansemacquire knows we're waiting
 101                 // (we set nwait above), so go to sleep.
 102                 root.queue(addr, s)
 103                 goparkunlock(&root.lock, "semacquire", traceEvGoBlockSync, 4)
 104                 if cansemacquire(addr) {
 105                         break
 106                 }
 107         }
 108         if s.releasetime > 0 {
 109                 blockevent(int64(s.releasetime)-t0, 3)
 110         }
 111         releaseSudog(s)
 112 }
 113
 114 func semrelease(addr *uint32) {
 115         root := semroot(addr)
 116         atomic.Xadd(addr, 1)
 117
 118         // Easy case: no waiters?
 119         // This check must happen after the xadd, to avoid a missed wakeup
 120         // (see loop in semacquire).
 121         if atomic.Load(&root.nwait) == 0 {
 122                 return
 123         }
 124
 125         // Harder case: search for a waiter and wake it.
 126         lock(&root.lock)
 127         if atomic.Load(&root.nwait) == 0 {
 128                 // The count is already consumed by another goroutine,
 129                 // so no need to wake up another goroutine.
 130                 unlock(&root.lock)
 131                 return
 132         }
 133         s := root.head
 134         for ; s != nil; s = s.next {
 135                 if s.elem == unsafe.Pointer(addr) {
 136                         atomic.Xadd(&root.nwait, -1)
 137                         root.dequeue(s)
 138                         break
 139                 }
 140         }
 141         unlock(&root.lock)
 142         if s != nil {
 143                 if s.releasetime != 0 {
 144                         s.releasetime = cputicks()
 145                 }
 146                 goready(s.g, 4)
 147         }
 148 }
 149
 150 func semroot(addr *uint32) *semaRoot {
 151         return &semtable[(uintptr(unsafe.Pointer(addr))>>3)%semTabSize].root
 152 }
 153
 154 func cansemacquire(addr *uint32) bool {
 155         for {
 156                 v := atomic.Load(addr)
 157                 if v == 0 {
 158                         return false
 159                 }
 160                 if atomic.Cas(addr, v, v-1) {
 161                         return true
 162                 }
 163         }
 164 }
 165
 166 func (root *semaRoot) queue(addr *uint32, s *sudog) {
 167         s.g = getg()
 168         s.elem = unsafe.Pointer(addr)
 169         s.next = nil
 170         s.prev = root.tail
 171         if root.tail != nil {
 172                 root.tail.next = s
 173         } else {
 174                 root.head = s
 175         }
 176         root.tail = s
 177 }
 178
 179 func (root *semaRoot) dequeue(s *sudog) {
 180         if s.next != nil {
 181                 s.next.prev = s.prev
 182         } else {
 183                 root.tail = s.prev
 184         }
 185         if s.prev != nil {
 186                 s.prev.next = s.next
 187         } else {
 188                 root.head = s.next
 189         }
 190         s.elem = nil
 191         s.next = nil
 192         s.prev = nil
 193 }
 194
 195 // Synchronous semaphore for sync.Cond.
 196 type syncSema struct {
 197         lock mutex
 198         head *sudog
 199         tail *sudog
 200 }
 201
 202 // syncsemacquire waits for a pairing syncsemrelease on the same semaphore s.
 203 //go:linkname syncsemacquire sync.runtime_Syncsemacquire
 204 func syncsemacquire(s *syncSema) {
 205         lock(&s.lock)
 206         if s.head != nil && s.head.nrelease > 0 {
 207                 // Have pending release, consume it.
 208                 var wake *sudog
 209                 s.head.nrelease--
 210                 if s.head.nrelease == 0 {
 211                         wake = s.head
 212                         s.head = wake.next
 213                         if s.head == nil {
 214                                 s.tail = nil
 215                         }
 216                 }
 217                 unlock(&s.lock)
 218                 if wake != nil {
 219                         wake.next = nil
 220                         goready(wake.g, 4)
 221                 }
 222         } else {
 223                 // Enqueue itself.
 224                 w := acquireSudog()
 225                 w.g = getg()
 226                 w.nrelease = -1
 227                 w.next = nil
 228                 w.releasetime = 0
 229                 t0 := int64(0)
 230                 if blockprofilerate > 0 {
 231                         t0 = cputicks()
 232                         w.releasetime = -1
 233                 }
 234                 if s.tail == nil {
 235                         s.head = w
 236                 } else {
 237                         s.tail.next = w
 238                 }
 239                 s.tail = w
 240                 goparkunlock(&s.lock, "semacquire", traceEvGoBlockCond, 3)
 241                 if t0 != 0 {
 242                         blockevent(int64(w.releasetime)-t0, 2)
 243                 }
 244                 releaseSudog(w)
 245         }
 246 }
 247
 248 // syncsemrelease waits for n pairing syncsemacquire on the same semaphore s.
 249 //go:linkname syncsemrelease sync.runtime_Syncsemrelease
 250 func syncsemrelease(s *syncSema, n uint32) {
 251         lock(&s.lock)
 252         for n > 0 && s.head != nil && s.head.nrelease < 0 {
 253                 // Have pending acquire, satisfy it.
 254                 wake := s.head
 255                 s.head = wake.next
 256                 if s.head == nil {
 257                         s.tail = nil
 258                 }
 259                 if wake.releasetime != 0 {
 260                         wake.releasetime = cputicks()
 261                 }
 262                 wake.next = nil
 263                 goready(wake.g, 4)
 264                 n--
 265         }
 266         if n > 0 {
 267                 // enqueue itself
 268                 w := acquireSudog()
 269                 w.g = getg()
 270                 w.nrelease = int32(n)
 271                 w.next = nil
 272                 w.releasetime = 0
 273                 if s.tail == nil {
 274                         s.head = w
 275                 } else {
 276                         s.tail.next = w
 277                 }
 278                 s.tail = w
 279                 goparkunlock(&s.lock, "semarelease", traceEvGoBlockCond, 3)
 280                 releaseSudog(w)
 281         } else {
 282                 unlock(&s.lock)
 283         }
 284 }
 285
 286 //go:linkname syncsemcheck sync.runtime_Syncsemcheck
 287 func syncsemcheck(sz uintptr) {
 288         if sz != unsafe.Sizeof(syncSema{}) {
 289                 print("runtime: bad syncSema size - sync=", sz, " runtime=", unsafe.Sizeof(syncSema{}), "\n")
 290                 throw("bad syncSema size")
 291         }
 292 }