netpoll.go

Documentation: runtime

     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