golang netpoll 代码
文件路径：/src/runtime/netpoll.go
/ Copyright 2013 The Go Authors. All rights reserved.
/ Use of this source code is governed by a BSD-style
/ license that can be found in the LICENSE file.
/go:build unix || (js && wasm) || windows
package runtime
import (
"runtime/internal/atomic"
"unsafe"
)
/ Integrated network poller (platform-independent part).
/ A particular implementation (epoll/kqueue/port/AIX/Windows)
/ must define the following functions:
/
/ func netpollinit()
/ Initialize the poller. Only called once.
/
/ func netpollopen(fd uintptr, pd *pollDesc) int32
/ Arm edge-triggered notifications for fd. The pd argument is to pass
/ back to netpollready when fd is ready. Return an errno value.
/
/ func netpollclose(fd uintptr) int32
/ Disable notifications for fd. Return an errno value.
/
/ func netpoll(delta int64) gList
/ Poll the network. If delta < 0, block indefinitely. If delta == 0,
/ poll without blocking. If delta > 0, block for up to delta nanoseconds.
/ Return a list of goroutines built by calling netpollready.
/
/ func netpollBreak()
/ Wake up the network poller, assumed to be blocked in netpoll.
/
/ func netpollIsPollDescriptor(fd uintptr) bool
/ Reports whether fd is a file descriptor used by the poller.
/ Error codes returned by runtime_pollReset and runtime_pollWait.
/ These must match the values in internal/poll/fd_poll_runtime.go.
const (
pollNoError = 0 / no error
pollErrClosing = 1 / descriptor is closed
pollErrTimeout = 2 / I/O timeout
pollErrNotPollable = 3 / general error polling descriptor
)
/ pollDesc contains 2 binary semaphores, rg and wg, to park reader and writer
/ goroutines respectively. The semaphore can be in the following states:
/
/ pdReady - io readiness notification is pending;
/ a goroutine consumes the notification by changing the state to nil.
/ pdWait - a goroutine prepares to park on the semaphore, but not yet parked;
/ the goroutine commits to park by changing the state to G pointer,
/ or, alternatively, concurrent io notification changes the state to pdReady,
/ or, alternatively, concurrent timeout/close changes the state to nil.
/ G pointer - the goroutine is blocked on the semaphore;
/ io notification or timeout/close changes the state to pdReady or nil respectively
/ and unparks the goroutine.
/ nil - none of the above.
const (
pdReady uintptr = 1
pdWait uintptr = 2
)
const pollBlockSize = 4 * 1024
/ Network poller descriptor.
/
/ No heap pointers.
/
/go:notinheap
type pollDesc struct {
link *pollDesc / in pollcache, protected by pollcache.lock
fd uintptr / constant for pollDesc usage lifetime
/ atomicInfo holds bits from closing, rd, and wd,
/ which are only ever written while holding the lock,
/ summarized for use by netpollcheckerr,
/ which cannot acquire the lock.
/ After writing these fields under lock in a way that
/ might change the summary, code must call publishInfo
/ before releasing the lock.
/ Code that changes fields and then calls netpollunblock
/ (while still holding the lock) must call publishInfo
/ before calling netpollunblock, because publishInfo is what
/ stops netpollblock from blocking anew
/ (by changing the result of netpollcheckerr).
/ atomicInfo also holds the eventErr bit,
/ recording whether a poll event on the fd got an error;
/ atomicInfo is the only source of truth for that bit.
atomicInfo atomic.Uint32 / atomic pollInfo
/ rg, wg are accessed atomically and hold g pointers.
/ (Using atomic.Uintptr here is similar to using guintptr elsewhere.)
rg atomic.Uintptr / pdReady, pdWait, G waiting for read or nil
wg atomic.Uintptr / pdReady, pdWait, G waiting for write or nil
lock mutex / protects the following fields
closing bool
user uint32 / user settable cookie
rseq uintptr / protects from stale read timers
rt timer / read deadline timer (set if rt.f != nil)
rd int64 / read deadline (a nanotime in the future, -1 when expired)
wseq uintptr / protects from stale write timers
wt timer / write deadline timer
wd int64 / write deadline (a nanotime in the future, -1 when expired)
self *pollDesc / storage for indirect interface. See (*pollDesc).makeArg.
}
/ pollInfo is the bits needed by netpollcheckerr, stored atomically,
/ mostly duplicating state that is manipulated under lock in pollDesc.
/ The one exception is the pollEventErr bit, which is maintained only
/ in the pollInfo.
type pollInfo uint32
const (
pollClosing = 1 << iota
pollEventErr
pollExpiredReadDeadline
pollExpiredWriteDeadline
)
func (i pollInfo) closing() bool { return i&pollClosing != 0 }
func (i pollInfo) eventErr() bool { return i&pollEventErr != 0 }
func (i pollInfo) expiredReadDeadline() bool { return i&pollExpiredReadDeadline != 0 }
func (i pollInfo) expiredWriteDeadline() bool { return i&pollExpiredWriteDeadline != 0 }
/ info returns the pollInfo corresponding to pd.
func (pd *pollDesc) info() pollInfo {
return pollInfo(pd.atomicInfo.Load())
}
/ publishInfo updates pd.atomicInfo (returned by pd.info)
/ using the other values in pd.
/ It must be called while holding pd.lock,
/ and it must be called after changing anything
/ that might affect the info bits.
/ In practice this means after changing closing
/ or changing rd or wd from < 0 to >= 0.
func (pd *pollDesc) publishInfo() {
var info uint32
if pd.closing {
info |= pollClosing
}
if pd.rd < 0 {
info |= pollExpiredReadDeadline
}
if pd.wd < 0 {
info |= pollExpiredWriteDeadline
}
/ Set all of x except the pollEventErr bit.
x := pd.atomicInfo.Load()
for !pd.atomicInfo.CompareAndSwap(x, (x&pollEventErr)|info) {
x = pd.atomicInfo.Load()
}
}
/ setEventErr sets the result of pd.info().eventErr() to b.
func (pd *pollDesc) setEventErr(b bool) {
x := pd.atomicInfo.Load()
for (x&pollEventErr != 0) != b && !pd.atomicInfo.CompareAndSwap(x, x^pollEventErr) {
x = pd.atomicInfo.Load()
}
}
type pollCache struct {
lock mutex
first *pollDesc
/ PollDesc objects must be type-stable,
/ because we can get ready notification from epoll/kqueue
/ after the descriptor is closed/reused.
/ Stale notifications are detected using seq variable,
/ seq is incremented when deadlines are changed or descriptor is reused.
}
var (
netpollInitLock mutex
netpollInited uint32
pollcache pollCache
netpollWaiters uint32
)
/go:linkname poll_runtime_pollServerInit internal/poll.runtime_pollServerInit
func poll_runtime_pollServerInit() {
netpollGenericInit()
}
func netpollGenericInit() {
if atomic.Load(&netpollInited) == 0 {
lockInit(&netpollInitLock, lockRankNetpollInit)
lock(&netpollInitLock)
if netpollInited == 0 {
netpollinit()
atomic.Store(&netpollInited, 1)
}
unlock(&netpollInitLock)
}
}
func netpollinited() bool {
return atomic.Load(&netpollInited) != 0
}
/go:linkname poll_runtime_isPollServerDescriptor internal/poll.runtime_isPollServerDescriptor
/ poll_runtime_isPollServerDescriptor reports whether fd is a
/ descriptor being used by netpoll.
func poll_runtime_isPollServerDescriptor(fd uintptr) bool {
return netpollIsPollDescriptor(fd)
}
/go:linkname poll_runtime_pollOpen internal/poll.runtime_pollOpen
func poll_runtime_pollOpen(fd uintptr) (*pollDesc, int) {
pd := pollcache.alloc()
lock(&pd.lock)
wg := pd.wg.Load()
if wg != 0 && wg != pdReady {
throw("runtime: blocked write on free polldesc")
}
rg := pd.rg.Load()
if rg != 0 && rg != pdReady {
throw("runtime: blocked read on free polldesc")
}
pd.fd = fd
pd.closing = false
pd.setEventErr(false)
pd.rseq++
pd.rg.Store(0)
pd.rd = 0
pd.wseq++
pd.wg.Store(0)
pd.wd = 0
pd.self = pd
pd.publishInfo()
unlock(&pd.lock)
errno := netpollopen(fd, pd)
if errno != 0 {
pollcache.free(pd)
return nil, int(errno)
}
return pd, 0
}
/go:linkname poll_runtime_pollClose internal/poll.runtime_pollClose
func poll_runtime_pollClose(pd *pollDesc) {
if !pd.closing {
throw("runtime: close polldesc w/o unblock")
}
wg := pd.wg.Load()
if wg != 0 && wg != pdReady {
throw("runtime: blocked write on closing polldesc")
}
rg := pd.rg.Load()
if rg != 0 && rg != pdReady {
throw("runtime: blocked read on closing polldesc")
}
netpollclose(pd.fd)
pollcache.free(pd)
}
func (c *pollCache) free(pd *pollDesc) {
lock(&c.lock)
pd.link = c.first
c.first = pd
unlock(&c.lock)
}
/ poll_runtime_pollReset, which is internal/poll.runtime_pollReset,
/ prepares a descriptor for polling in mode, which is 'r' or 'w'.
/ This returns an error code; the codes are defined above.
/
/go:linkname poll_runtime_pollReset internal/poll.runtime_pollReset
func poll_runtime_pollReset(pd *pollDesc, mode int) int {
errcode := netpollcheckerr(pd, int32(mode))
if errcode != pollNoError {
return errcode
}
if mode == 'r' {
pd.rg.Store(0)
} else if mode == 'w' {
pd.wg.Store(0)
}
return pollNoError
}
/ poll_runtime_pollWait, which is internal/poll.runtime_pollWait,
/ waits for a descriptor to be ready for reading or writing,
/ according to mode, which is 'r' or 'w'.
/ This returns an error code; the codes are defined above.
/
/go:linkname poll_runtime_pollWait internal/poll.runtime_pollWait
func poll_runtime_pollWait(pd *pollDesc, mode int) int {
errcode := netpollcheckerr(pd, int32(mode))
if errcode != pollNoError {
return errcode
}
/ As for now only Solaris, illumos, and AIX use level-triggered IO.
if GOOS == "solaris" || GOOS == "illumos" || GOOS == "aix" {
netpollarm(pd, mode)
}
for !netpollblock(pd, int32(mode), false) {
errcode = netpollcheckerr(pd, int32(mode))
if errcode != pollNoError {
return errcode
}
/ Can happen if timeout has fired and unblocked us,
/ but before we had a chance to run, timeout has been reset.
/ Pretend it has not happened and retry.
}
return pollNoError
}
/go:linkname poll_runtime_pollWaitCanceled internal/poll.runtime_pollWaitCanceled
func poll_runtime_pollWaitCanceled(pd *pollDesc, mode int) {
/ This function is used only on windows after a failed attempt to cancel
/ a pending async IO operation. Wait for ioready, ignore closing or timeouts.
for !netpollblock(pd, int32(mode), true) {
}
}
/go:linkname poll_runtime_pollSetDeadline internal/poll.runtime_pollSetDeadline
func poll_runtime_pollSetDeadline(pd *pollDesc, d int64, mode int) {
lock(&pd.lock)
if pd.closing {
unlock(&pd.lock)
return
}
rd0, wd0 := pd.rd, pd.wd
combo0 := rd0 > 0 && rd0 == wd0
if d > 0 {
d += nanotime()
if d <= 0 {
/ If the user has a deadline in the future, but the delay calculation
/ overflows, then set the deadline to the maximum possible value.
d = 1<<63 - 1
}
}
if mode == 'r' || mode == 'r'+'w' {
pd.rd = d
}
if mode == 'w' || mode == 'r'+'w' {
pd.wd = d
}
pd.publishInfo()
combo := pd.rd > 0 && pd.rd == pd.wd
rtf := netpollReadDeadline
if combo {
rtf = netpollDeadline
}
if pd.rt.f == nil {
if pd.rd > 0 {
pd.rt.f = rtf
/ Copy current seq into the timer arg.
/ Timer func will check the seq against current descriptor seq,
/ if they differ the descriptor was reused or timers were reset.
pd.rt.arg = pd.makeArg()
pd.rt.seq = pd.rseq
resettimer(&pd.rt, pd.rd)
}
} else if pd.rd != rd0 || combo != combo0 {
pd.rseq++ / invalidate current timers
if pd.rd > 0 {
modtimer(&pd.rt, pd.rd, 0, rtf, pd.makeArg(), pd.rseq)
} else {
deltimer(&pd.rt)
pd.rt.f = nil
}
}
if pd.wt.f == nil {
if pd.wd > 0 && !combo {
pd.wt.f = netpollWriteDeadline
pd.wt.arg = pd.makeArg()
pd.wt.seq = pd.wseq
resettimer(&pd.wt, pd.wd)
}
} else if pd.wd != wd0 || combo != combo0 {
pd.wseq++ / invalidate current timers
if pd.wd > 0 && !combo {
modtimer(&pd.wt, pd.wd, 0, netpollWriteDeadline, pd.makeArg(), pd.wseq)
} else {
deltimer(&pd.wt)
pd.wt.f = nil
}
}
/ If we set the new deadline in the past, unblock currently pending IO if any.
/ Note that pd.publishInfo has already been called, above, immediately after modifying rd and wd.
var rg, wg *g
if pd.rd < 0 {
rg = netpollunblock(pd, 'r', false)
}
if pd.wd < 0 {
wg = netpollunblock(pd, 'w', false)
}
unlock(&pd.lock)
if rg != nil {
netpollgoready(rg, 3)
}
if wg != nil {
netpollgoready(wg, 3)
}
}
/go:linkname poll_runtime_pollUnblock internal/poll.runtime_pollUnblock
func poll_runtime_pollUnblock(pd *pollDesc) {
lock(&pd.lock)
if pd.closing {
throw("runtime: unblock on closing polldesc")
}
pd.closing = true
pd.rseq++
pd.wseq++
var rg, wg *g
pd.publishInfo()
rg = netpollunblock(pd, 'r', false)
wg = netpollunblock(pd, 'w', false)
if pd.rt.f != nil {
deltimer(&pd.rt)
pd.rt.f = nil
}
if pd.wt.f != nil {
deltimer(&pd.wt)
pd.wt.f = nil
}
unlock(&pd.lock)
if rg != nil {
netpollgoready(rg, 3)
}
if wg != nil {
netpollgoready(wg, 3)
}
}
/ netpollready is called by the platform-specific netpoll function.
/ It declares that the fd associated with pd is ready for I/O.
/ The toRun argument is used to build a list of goroutines to return
/ from netpoll. The mode argument is 'r', 'w', or 'r'+'w' to indicate
/ whether the fd is ready for reading or writing or both.
/
/ This may run while the world is stopped, so write barriers are not allowed.
/
/go:nowritebarrier
func netpollready(toRun *gList, pd *pollDesc, mode int32) {
var rg, wg *g
if mode == 'r' || mode == 'r'+'w' {
rg = netpollunblock(pd, 'r', true)
}
if mode == 'w' || mode == 'r'+'w' {
wg = netpollunblock(pd, 'w', true)
}
if rg != nil {
toRun.push(rg)
}
if wg != nil {
toRun.push(wg)
}
}
func netpollcheckerr(pd *pollDesc, mode int32) int {
info := pd.info()
if info.closing() {
return pollErrClosing
}
if (mode == 'r' && info.expiredReadDeadline()) || (mode == 'w' && info.expiredWriteDeadline()) {
return pollErrTimeout
}
/ Report an event scanning error only on a read event.
/ An error on a write event will be captured in a subsequent
/ write call that is able to report a more specific error.
if mode == 'r' && info.eventErr() {
return pollErrNotPollable
}
return pollNoError
}
func netpollblockcommit(gp *g, gpp unsafe.Pointer) bool {
r := atomic.Casuintptr((*uintptr)(gpp), pdWait, uintptr(unsafe.Pointer(gp)))
if r {
/ Bump the count of goroutines waiting for the poller.
/ The scheduler uses this to decide whether to block
/ waiting for the poller if there is nothing else to do.
atomic.Xadd(&netpollWaiters, 1)
}
return r
}
func netpollgoready(gp *g, traceskip int) {
atomic.Xadd(&netpollWaiters, -1)
goready(gp, traceskip+1)
}
/ returns true if IO is ready, or false if timedout or closed
/ waitio - wait only for completed IO, ignore errors
/ Concurrent calls to netpollblock in the same mode are forbidden, as pollDesc
/ can hold only a single waiting goroutine for each mode.
func netpollblock(pd *pollDesc, mode int32, waitio bool) bool {
gpp := &pd.rg
if mode == 'w' {
gpp = &pd.wg
}
/ set the gpp semaphore to pdWait
for {
/ Consume notification if already ready.
if gpp.CompareAndSwap(pdReady, 0) {
return true
}
if gpp.CompareAndSwap(0, pdWait) {
break
}
/ Double check that this isn't corrupt; otherwise we'd loop
/ forever.
if v := gpp.Load(); v != pdReady && v != 0 {
throw("runtime: double wait")
}
}
/ need to recheck error states after setting gpp to pdWait
/ this is necessary because runtime_pollUnblock/runtime_pollSetDeadline/deadlineimpl
/ do the opposite: store to closing/rd/wd, publishInfo, load of rg/wg
if waitio || netpollcheckerr(pd, mode) == pollNoError {
gopark(netpollblockcommit, unsafe.Pointer(gpp), waitReasonIOWait, traceEvGoBlockNet, 5)
}
/ be careful to not lose concurrent pdReady notification
old := gpp.Swap(0)
if old > pdWait {
throw("runtime: corrupted polldesc")
}
return old == pdReady
}
func netpollunblock(pd *pollDesc, mode int32, ioready bool) *g {
gpp := &pd.rg
if mode == 'w' {
gpp = &pd.wg
}
for {
old := gpp.Load()
if old == pdReady {
return nil
}
if old == 0 && !ioready {
/ Only set pdReady for ioready. runtime_pollWait
/ will check for timeout/cancel before waiting.
return nil
}
var new uintptr
if ioready {
new = pdReady
}
if gpp.CompareAndSwap(old, new) {
if old == pdWait {
old = 0
}
return (*g)(unsafe.Pointer(old))
}
}
}
func netpolldeadlineimpl(pd *pollDesc, seq uintptr, read, write bool) {
lock(&pd.lock)
/ Seq arg is seq when the timer was set.
/ If it's stale, ignore the timer event.
currentSeq := pd.rseq
if !read {
currentSeq = pd.wseq
}
if seq != currentSeq {
/ The descriptor was reused or timers were reset.
unlock(&pd.lock)
return
}
var rg *g
if read {
if pd.rd <= 0 || pd.rt.f == nil {
throw("runtime: inconsistent read deadline")
}
pd.rd = -1
pd.publishInfo()
rg = netpollunblock(pd, 'r', false)
}
var wg *g
if write {
if pd.wd <= 0 || pd.wt.f == nil && !read {
throw("runtime: inconsistent write deadline")
}
pd.wd = -1
pd.publishInfo()
wg = netpollunblock(pd, 'w', false)
}
unlock(&pd.lock)
if rg != nil {
netpollgoready(rg, 0)
}
if wg != nil {
netpollgoready(wg, 0)
}
}
func netpollDeadline(arg any, seq uintptr) {
netpolldeadlineimpl(arg.(*pollDesc), seq, true, true)
}
func netpollReadDeadline(arg any, seq uintptr) {
netpolldeadlineimpl(arg.(*pollDesc), seq, true, false)
}
func netpollWriteDeadline(arg any, seq uintptr) {
netpolldeadlineimpl(arg.(*pollDesc), seq, false, true)
}
func (c *pollCache) alloc() *pollDesc {
lock(&c.lock)
if c.first == nil {
const pdSize = unsafe.Sizeof(pollDesc{})
n := pollBlockSize / pdSize
if n == 0 {
n = 1
}
/ Must be in non-GC memory because can be referenced
/ only from epoll/kqueue internals.
mem := persistentalloc(n*pdSize, 0, &memstats.other_sys)
for i := uintptr(0); i < n; i++ {
pd := (*pollDesc)(add(mem, i*pdSize))
pd.link = c.first
c.first = pd
}
}
pd := c.first
c.first = pd.link
lockInit(&pd.lock, lockRankPollDesc)
unlock(&c.lock)
return pd
}
/ makeArg converts pd to an interface{}.
/ makeArg does not do any allocation. Normally, such
/ a conversion requires an allocation because pointers to
/ go:notinheap types (which pollDesc is) must be stored
/ in interfaces indirectly. See issue 42076.
func (pd *pollDesc) makeArg() (i any) {
x := (*eface)(unsafe.Pointer(&i))
x._type = pdType
x.data = unsafe.Pointer(&pd.self)
return
}
var (
pdEface any = (*pollDesc)(nil)
pdType *_type = efaceOf(&pdEface)._type
)