1 // Copyright 2013 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 // +build aix darwin dragonfly freebsd js,wasm linux nacl netbsd openbsd solaris windows 6 7 package runtime 8 9 import ( 10 "runtime/internal/atomic" 11 "unsafe" 12 ) 13 14 // Integrated network poller (platform-independent part). 15 // A particular implementation (epoll/kqueue) must define the following functions: 16 // func netpollinit() // to initialize the poller 17 // func netpollopen(fd uintptr, pd *pollDesc) int32 // to arm edge-triggered notifications 18 // and associate fd with pd. 19 // An implementation must call the following function to denote that the pd is ready. 20 // func netpollready(gpp **g, pd *pollDesc, mode int32) 21 22 // pollDesc contains 2 binary semaphores, rg and wg, to park reader and writer 23 // goroutines respectively. The semaphore can be in the following states: 24 // pdReady - io readiness notification is pending; 25 // a goroutine consumes the notification by changing the state to nil. 26 // pdWait - a goroutine prepares to park on the semaphore, but not yet parked; 27 // the goroutine commits to park by changing the state to G pointer, 28 // or, alternatively, concurrent io notification changes the state to READY, 29 // or, alternatively, concurrent timeout/close changes the state to nil. 30 // G pointer - the goroutine is blocked on the semaphore; 31 // io notification or timeout/close changes the state to READY or nil respectively 32 // and unparks the goroutine. 33 // nil - nothing of the above. 34 const ( 35 pdReady uintptr = 1 36 pdWait uintptr = 2 37 ) 38 39 const pollBlockSize = 4 * 1024 40 41 // Network poller descriptor. 42 // 43 // No heap pointers. 44 // 45 //go:notinheap 46 type pollDesc struct { 47 link *pollDesc // in pollcache, protected by pollcache.lock 48 49 // The lock protects pollOpen, pollSetDeadline, pollUnblock and deadlineimpl operations. 50 // This fully covers seq, rt and wt variables. fd is constant throughout the PollDesc lifetime. 51 // pollReset, pollWait, pollWaitCanceled and runtime·netpollready (IO readiness notification) 52 // proceed w/o taking the lock. So closing, rg, rd, wg and wd are manipulated 53 // in a lock-free way by all operations. 54 // NOTE(dvyukov): the following code uses uintptr to store *g (rg/wg), 55 // that will blow up when GC starts moving objects. 56 lock mutex // protects the following fields 57 fd uintptr 58 closing bool 59 user uint32 // user settable cookie 60 rseq uintptr // protects from stale read timers 61 rg uintptr // pdReady, pdWait, G waiting for read or nil 62 rt timer // read deadline timer (set if rt.f != nil) 63 rd int64 // read deadline 64 wseq uintptr // protects from stale write timers 65 wg uintptr // pdReady, pdWait, G waiting for write or nil 66 wt timer // write deadline timer 67 wd int64 // write deadline 68 } 69 70 type pollCache struct { 71 lock mutex 72 first *pollDesc 73 // PollDesc objects must be type-stable, 74 // because we can get ready notification from epoll/kqueue 75 // after the descriptor is closed/reused. 76 // Stale notifications are detected using seq variable, 77 // seq is incremented when deadlines are changed or descriptor is reused. 78 } 79 80 var ( 81 netpollInited uint32 82 pollcache pollCache 83 netpollWaiters uint32 84 ) 85 86 //go:linkname poll_runtime_pollServerInit internal/poll.runtime_pollServerInit 87 func poll_runtime_pollServerInit() { 88 netpollinit() 89 atomic.Store(&netpollInited, 1) 90 } 91 92 func netpollinited() bool { 93 return atomic.Load(&netpollInited) != 0 94 } 95 96 //go:linkname poll_runtime_isPollServerDescriptor internal/poll.runtime_isPollServerDescriptor 97 98 // poll_runtime_isPollServerDescriptor reports whether fd is a 99 // descriptor being used by netpoll. 100 func poll_runtime_isPollServerDescriptor(fd uintptr) bool { 101 fds := netpolldescriptor() 102 if GOOS != "aix" { 103 return fd == fds 104 } else { 105 // AIX have a pipe in its netpoll implementation. 106 // Therefore, two fd are returned by netpolldescriptor using a mask. 107 return fd == fds&0xFFFF || fd == (fds>>16)&0xFFFF 108 } 109 } 110 111 //go:linkname poll_runtime_pollOpen internal/poll.runtime_pollOpen 112 func poll_runtime_pollOpen(fd uintptr) (*pollDesc, int) { 113 pd := pollcache.alloc() 114 lock(&pd.lock) 115 if pd.wg != 0 && pd.wg != pdReady { 116 throw("runtime: blocked write on free polldesc") 117 } 118 if pd.rg != 0 && pd.rg != pdReady { 119 throw("runtime: blocked read on free polldesc") 120 } 121 pd.fd = fd 122 pd.closing = false 123 pd.rseq++ 124 pd.rg = 0 125 pd.rd = 0 126 pd.wseq++ 127 pd.wg = 0 128 pd.wd = 0 129 unlock(&pd.lock) 130 131 var errno int32 132 errno = netpollopen(fd, pd) 133 return pd, int(errno) 134 } 135 136 //go:linkname poll_runtime_pollClose internal/poll.runtime_pollClose 137 func poll_runtime_pollClose(pd *pollDesc) { 138 if !pd.closing { 139 throw("runtime: close polldesc w/o unblock") 140 } 141 if pd.wg != 0 && pd.wg != pdReady { 142 throw("runtime: blocked write on closing polldesc") 143 } 144 if pd.rg != 0 && pd.rg != pdReady { 145 throw("runtime: blocked read on closing polldesc") 146 } 147 netpollclose(pd.fd) 148 pollcache.free(pd) 149 } 150 151 func (c *pollCache) free(pd *pollDesc) { 152 lock(&c.lock) 153 pd.link = c.first 154 c.first = pd 155 unlock(&c.lock) 156 } 157 158 //go:linkname poll_runtime_pollReset internal/poll.runtime_pollReset 159 func poll_runtime_pollReset(pd *pollDesc, mode int) int { 160 err := netpollcheckerr(pd, int32(mode)) 161 if err != 0 { 162 return err 163 } 164 if mode == 'r' { 165 pd.rg = 0 166 } else if mode == 'w' { 167 pd.wg = 0 168 } 169 return 0 170 } 171 172 //go:linkname poll_runtime_pollWait internal/poll.runtime_pollWait 173 func poll_runtime_pollWait(pd *pollDesc, mode int) int { 174 err := netpollcheckerr(pd, int32(mode)) 175 if err != 0 { 176 return err 177 } 178 // As for now only Solaris and AIX use level-triggered IO. 179 if GOOS == "solaris" || GOOS == "aix" { 180 netpollarm(pd, mode) 181 } 182 for !netpollblock(pd, int32(mode), false) { 183 err = netpollcheckerr(pd, int32(mode)) 184 if err != 0 { 185 return err 186 } 187 // Can happen if timeout has fired and unblocked us, 188 // but before we had a chance to run, timeout has been reset. 189 // Pretend it has not happened and retry. 190 } 191 return 0 192 } 193 194 //go:linkname poll_runtime_pollWaitCanceled internal/poll.runtime_pollWaitCanceled 195 func poll_runtime_pollWaitCanceled(pd *pollDesc, mode int) { 196 // This function is used only on windows after a failed attempt to cancel 197 // a pending async IO operation. Wait for ioready, ignore closing or timeouts. 198 for !netpollblock(pd, int32(mode), true) { 199 } 200 } 201 202 //go:linkname poll_runtime_pollSetDeadline internal/poll.runtime_pollSetDeadline 203 func poll_runtime_pollSetDeadline(pd *pollDesc, d int64, mode int) { 204 lock(&pd.lock) 205 if pd.closing { 206 unlock(&pd.lock) 207 return 208 } 209 rd0, wd0 := pd.rd, pd.wd 210 combo0 := rd0 > 0 && rd0 == wd0 211 if d > 0 { 212 d += nanotime() 213 if d <= 0 { 214 // If the user has a deadline in the future, but the delay calculation 215 // overflows, then set the deadline to the maximum possible value. 216 d = 1<<63 - 1 217 } 218 } 219 if mode == 'r' || mode == 'r'+'w' { 220 pd.rd = d 221 } 222 if mode == 'w' || mode == 'r'+'w' { 223 pd.wd = d 224 } 225 combo := pd.rd > 0 && pd.rd == pd.wd 226 rtf := netpollReadDeadline 227 if combo { 228 rtf = netpollDeadline 229 } 230 if pd.rt.f == nil { 231 if pd.rd > 0 { 232 pd.rt.f = rtf 233 pd.rt.when = pd.rd 234 // Copy current seq into the timer arg. 235 // Timer func will check the seq against current descriptor seq, 236 // if they differ the descriptor was reused or timers were reset. 237 pd.rt.arg = pd 238 pd.rt.seq = pd.rseq 239 addtimer(&pd.rt) 240 } 241 } else if pd.rd != rd0 || combo != combo0 { 242 pd.rseq++ // invalidate current timers 243 if pd.rd > 0 { 244 modtimer(&pd.rt, pd.rd, 0, rtf, pd, pd.rseq) 245 } else { 246 deltimer(&pd.rt) 247 pd.rt.f = nil 248 } 249 } 250 if pd.wt.f == nil { 251 if pd.wd > 0 && !combo { 252 pd.wt.f = netpollWriteDeadline 253 pd.wt.when = pd.wd 254 pd.wt.arg = pd 255 pd.wt.seq = pd.wseq 256 addtimer(&pd.wt) 257 } 258 } else if pd.wd != wd0 || combo != combo0 { 259 pd.wseq++ // invalidate current timers 260 if pd.wd > 0 && !combo { 261 modtimer(&pd.wt, pd.wd, 0, netpollWriteDeadline, pd, pd.wseq) 262 } else { 263 deltimer(&pd.wt) 264 pd.wt.f = nil 265 } 266 } 267 // If we set the new deadline in the past, unblock currently pending IO if any. 268 var rg, wg *g 269 if pd.rd < 0 || pd.wd < 0 { 270 atomic.StorepNoWB(noescape(unsafe.Pointer(&wg)), nil) // full memory barrier between stores to rd/wd and load of rg/wg in netpollunblock 271 if pd.rd < 0 { 272 rg = netpollunblock(pd, 'r', false) 273 } 274 if pd.wd < 0 { 275 wg = netpollunblock(pd, 'w', false) 276 } 277 } 278 unlock(&pd.lock) 279 if rg != nil { 280 netpollgoready(rg, 3) 281 } 282 if wg != nil { 283 netpollgoready(wg, 3) 284 } 285 } 286 287 //go:linkname poll_runtime_pollUnblock internal/poll.runtime_pollUnblock 288 func poll_runtime_pollUnblock(pd *pollDesc) { 289 lock(&pd.lock) 290 if pd.closing { 291 throw("runtime: unblock on closing polldesc") 292 } 293 pd.closing = true 294 pd.rseq++ 295 pd.wseq++ 296 var rg, wg *g 297 atomic.StorepNoWB(noescape(unsafe.Pointer(&rg)), nil) // full memory barrier between store to closing and read of rg/wg in netpollunblock 298 rg = netpollunblock(pd, 'r', false) 299 wg = netpollunblock(pd, 'w', false) 300 if pd.rt.f != nil { 301 deltimer(&pd.rt) 302 pd.rt.f = nil 303 } 304 if pd.wt.f != nil { 305 deltimer(&pd.wt) 306 pd.wt.f = nil 307 } 308 unlock(&pd.lock) 309 if rg != nil { 310 netpollgoready(rg, 3) 311 } 312 if wg != nil { 313 netpollgoready(wg, 3) 314 } 315 } 316 317 // make pd ready, newly runnable goroutines (if any) are added to toRun. 318 // May run during STW, so write barriers are not allowed. 319 //go:nowritebarrier 320 func netpollready(toRun *gList, pd *pollDesc, mode int32) { 321 var rg, wg *g 322 if mode == 'r' || mode == 'r'+'w' { 323 rg = netpollunblock(pd, 'r', true) 324 } 325 if mode == 'w' || mode == 'r'+'w' { 326 wg = netpollunblock(pd, 'w', true) 327 } 328 if rg != nil { 329 toRun.push(rg) 330 } 331 if wg != nil { 332 toRun.push(wg) 333 } 334 } 335 336 func netpollcheckerr(pd *pollDesc, mode int32) int { 337 if pd.closing { 338 return 1 // errClosing 339 } 340 if (mode == 'r' && pd.rd < 0) || (mode == 'w' && pd.wd < 0) { 341 return 2 // errTimeout 342 } 343 return 0 344 } 345 346 func netpollblockcommit(gp *g, gpp unsafe.Pointer) bool { 347 r := atomic.Casuintptr((*uintptr)(gpp), pdWait, uintptr(unsafe.Pointer(gp))) 348 if r { 349 // Bump the count of goroutines waiting for the poller. 350 // The scheduler uses this to decide whether to block 351 // waiting for the poller if there is nothing else to do. 352 atomic.Xadd(&netpollWaiters, 1) 353 } 354 return r 355 } 356 357 func netpollgoready(gp *g, traceskip int) { 358 atomic.Xadd(&netpollWaiters, -1) 359 goready(gp, traceskip+1) 360 } 361 362 // returns true if IO is ready, or false if timedout or closed 363 // waitio - wait only for completed IO, ignore errors 364 func netpollblock(pd *pollDesc, mode int32, waitio bool) bool { 365 gpp := &pd.rg 366 if mode == 'w' { 367 gpp = &pd.wg 368 } 369 370 // set the gpp semaphore to WAIT 371 for { 372 old := *gpp 373 if old == pdReady { 374 *gpp = 0 375 return true 376 } 377 if old != 0 { 378 throw("runtime: double wait") 379 } 380 if atomic.Casuintptr(gpp, 0, pdWait) { 381 break 382 } 383 } 384 385 // need to recheck error states after setting gpp to WAIT 386 // this is necessary because runtime_pollUnblock/runtime_pollSetDeadline/deadlineimpl 387 // do the opposite: store to closing/rd/wd, membarrier, load of rg/wg 388 if waitio || netpollcheckerr(pd, mode) == 0 { 389 gopark(netpollblockcommit, unsafe.Pointer(gpp), waitReasonIOWait, traceEvGoBlockNet, 5) 390 } 391 // be careful to not lose concurrent READY notification 392 old := atomic.Xchguintptr(gpp, 0) 393 if old > pdWait { 394 throw("runtime: corrupted polldesc") 395 } 396 return old == pdReady 397 } 398 399 func netpollunblock(pd *pollDesc, mode int32, ioready bool) *g { 400 gpp := &pd.rg 401 if mode == 'w' { 402 gpp = &pd.wg 403 } 404 405 for { 406 old := *gpp 407 if old == pdReady { 408 return nil 409 } 410 if old == 0 && !ioready { 411 // Only set READY for ioready. runtime_pollWait 412 // will check for timeout/cancel before waiting. 413 return nil 414 } 415 var new uintptr 416 if ioready { 417 new = pdReady 418 } 419 if atomic.Casuintptr(gpp, old, new) { 420 if old == pdReady || old == pdWait { 421 old = 0 422 } 423 return (*g)(unsafe.Pointer(old)) 424 } 425 } 426 } 427 428 func netpolldeadlineimpl(pd *pollDesc, seq uintptr, read, write bool) { 429 lock(&pd.lock) 430 // Seq arg is seq when the timer was set. 431 // If it's stale, ignore the timer event. 432 currentSeq := pd.rseq 433 if !read { 434 currentSeq = pd.wseq 435 } 436 if seq != currentSeq { 437 // The descriptor was reused or timers were reset. 438 unlock(&pd.lock) 439 return 440 } 441 var rg *g 442 if read { 443 if pd.rd <= 0 || pd.rt.f == nil { 444 throw("runtime: inconsistent read deadline") 445 } 446 pd.rd = -1 447 atomic.StorepNoWB(unsafe.Pointer(&pd.rt.f), nil) // full memory barrier between store to rd and load of rg in netpollunblock 448 rg = netpollunblock(pd, 'r', false) 449 } 450 var wg *g 451 if write { 452 if pd.wd <= 0 || pd.wt.f == nil && !read { 453 throw("runtime: inconsistent write deadline") 454 } 455 pd.wd = -1 456 atomic.StorepNoWB(unsafe.Pointer(&pd.wt.f), nil) // full memory barrier between store to wd and load of wg in netpollunblock 457 wg = netpollunblock(pd, 'w', false) 458 } 459 unlock(&pd.lock) 460 if rg != nil { 461 netpollgoready(rg, 0) 462 } 463 if wg != nil { 464 netpollgoready(wg, 0) 465 } 466 } 467 468 func netpollDeadline(arg interface{}, seq uintptr) { 469 netpolldeadlineimpl(arg.(*pollDesc), seq, true, true) 470 } 471 472 func netpollReadDeadline(arg interface{}, seq uintptr) { 473 netpolldeadlineimpl(arg.(*pollDesc), seq, true, false) 474 } 475 476 func netpollWriteDeadline(arg interface{}, seq uintptr) { 477 netpolldeadlineimpl(arg.(*pollDesc), seq, false, true) 478 } 479 480 func (c *pollCache) alloc() *pollDesc { 481 lock(&c.lock) 482 if c.first == nil { 483 const pdSize = unsafe.Sizeof(pollDesc{}) 484 n := pollBlockSize / pdSize 485 if n == 0 { 486 n = 1 487 } 488 // Must be in non-GC memory because can be referenced 489 // only from epoll/kqueue internals. 490 mem := persistentalloc(n*pdSize, 0, &memstats.other_sys) 491 for i := uintptr(0); i < n; i++ { 492 pd := (*pollDesc)(add(mem, i*pdSize)) 493 pd.link = c.first 494 c.first = pd 495 } 496 } 497 pd := c.first 498 c.first = pd.link 499 unlock(&c.lock) 500 return pd 501 } 502
View as plain text