Source file src/runtime/netpoll_kqueue.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 darwin dragonfly freebsd netbsd openbsd
     6  
     7  package runtime
     8  
     9  // Integrated network poller (kqueue-based implementation).
    10  
    11  import "unsafe"
    12  
    13  var (
    14  	kq int32 = -1
    15  
    16  	netpollBreakRd, netpollBreakWr uintptr // for netpollBreak
    17  )
    18  
    19  func netpollinit() {
    20  	kq = kqueue()
    21  	if kq < 0 {
    22  		println("runtime: kqueue failed with", -kq)
    23  		throw("runtime: netpollinit failed")
    24  	}
    25  	closeonexec(kq)
    26  	r, w, errno := nonblockingPipe()
    27  	if errno != 0 {
    28  		println("runtime: pipe failed with", -errno)
    29  		throw("runtime: pipe failed")
    30  	}
    31  	ev := keventt{
    32  		filter: _EVFILT_READ,
    33  		flags:  _EV_ADD,
    34  	}
    35  	*(*uintptr)(unsafe.Pointer(&ev.ident)) = uintptr(r)
    36  	n := kevent(kq, &ev, 1, nil, 0, nil)
    37  	if n < 0 {
    38  		println("runtime: kevent failed with", -n)
    39  		throw("runtime: kevent failed")
    40  	}
    41  	netpollBreakRd = uintptr(r)
    42  	netpollBreakWr = uintptr(w)
    43  }
    44  
    45  func netpollIsPollDescriptor(fd uintptr) bool {
    46  	return fd == uintptr(kq) || fd == netpollBreakRd || fd == netpollBreakWr
    47  }
    48  
    49  func netpollopen(fd uintptr, pd *pollDesc) int32 {
    50  	// Arm both EVFILT_READ and EVFILT_WRITE in edge-triggered mode (EV_CLEAR)
    51  	// for the whole fd lifetime. The notifications are automatically unregistered
    52  	// when fd is closed.
    53  	var ev [2]keventt
    54  	*(*uintptr)(unsafe.Pointer(&ev[0].ident)) = fd
    55  	ev[0].filter = _EVFILT_READ
    56  	ev[0].flags = _EV_ADD | _EV_CLEAR
    57  	ev[0].fflags = 0
    58  	ev[0].data = 0
    59  	ev[0].udata = (*byte)(unsafe.Pointer(pd))
    60  	ev[1] = ev[0]
    61  	ev[1].filter = _EVFILT_WRITE
    62  	n := kevent(kq, &ev[0], 2, nil, 0, nil)
    63  	if n < 0 {
    64  		return -n
    65  	}
    66  	return 0
    67  }
    68  
    69  func netpollclose(fd uintptr) int32 {
    70  	// Don't need to unregister because calling close()
    71  	// on fd will remove any kevents that reference the descriptor.
    72  	return 0
    73  }
    74  
    75  func netpollarm(pd *pollDesc, mode int) {
    76  	throw("runtime: unused")
    77  }
    78  
    79  // netpollBreak interrupts an epollwait.
    80  func netpollBreak() {
    81  	for {
    82  		var b byte
    83  		n := write(netpollBreakWr, unsafe.Pointer(&b), 1)
    84  		if n == 1 || n == -_EAGAIN {
    85  			break
    86  		}
    87  		if n == -_EINTR {
    88  			continue
    89  		}
    90  		println("runtime: netpollBreak write failed with", -n)
    91  		throw("runtime: netpollBreak write failed")
    92  	}
    93  }
    94  
    95  // netpoll checks for ready network connections.
    96  // Returns list of goroutines that become runnable.
    97  // delay < 0: blocks indefinitely
    98  // delay == 0: does not block, just polls
    99  // delay > 0: block for up to that many nanoseconds
   100  func netpoll(delay int64) gList {
   101  	if kq == -1 {
   102  		return gList{}
   103  	}
   104  	var tp *timespec
   105  	var ts timespec
   106  	if delay < 0 {
   107  		tp = nil
   108  	} else if delay == 0 {
   109  		tp = &ts
   110  	} else {
   111  		ts.setNsec(delay)
   112  		if ts.tv_sec > 1e6 {
   113  			// Darwin returns EINVAL if the sleep time is too long.
   114  			ts.tv_sec = 1e6
   115  		}
   116  		tp = &ts
   117  	}
   118  	var events [64]keventt
   119  retry:
   120  	n := kevent(kq, nil, 0, &events[0], int32(len(events)), tp)
   121  	if n < 0 {
   122  		if n != -_EINTR {
   123  			println("runtime: kevent on fd", kq, "failed with", -n)
   124  			throw("runtime: netpoll failed")
   125  		}
   126  		// If a timed sleep was interrupted, just return to
   127  		// recalculate how long we should sleep now.
   128  		if delay > 0 {
   129  			return gList{}
   130  		}
   131  		goto retry
   132  	}
   133  	var toRun gList
   134  	for i := 0; i < int(n); i++ {
   135  		ev := &events[i]
   136  
   137  		if uintptr(ev.ident) == netpollBreakRd {
   138  			if ev.filter != _EVFILT_READ {
   139  				println("runtime: netpoll: break fd ready for", ev.filter)
   140  				throw("runtime: netpoll: break fd ready for something unexpected")
   141  			}
   142  			if delay != 0 {
   143  				// netpollBreak could be picked up by a
   144  				// nonblocking poll. Only read the byte
   145  				// if blocking.
   146  				var tmp [16]byte
   147  				read(int32(netpollBreakRd), noescape(unsafe.Pointer(&tmp[0])), int32(len(tmp)))
   148  			}
   149  			continue
   150  		}
   151  
   152  		var mode int32
   153  		switch ev.filter {
   154  		case _EVFILT_READ:
   155  			mode += 'r'
   156  
   157  			// On some systems when the read end of a pipe
   158  			// is closed the write end will not get a
   159  			// _EVFILT_WRITE event, but will get a
   160  			// _EVFILT_READ event with EV_EOF set.
   161  			// Note that setting 'w' here just means that we
   162  			// will wake up a goroutine waiting to write;
   163  			// that goroutine will try the write again,
   164  			// and the appropriate thing will happen based
   165  			// on what that write returns (success, EPIPE, EAGAIN).
   166  			if ev.flags&_EV_EOF != 0 {
   167  				mode += 'w'
   168  			}
   169  		case _EVFILT_WRITE:
   170  			mode += 'w'
   171  		}
   172  		if mode != 0 {
   173  			pd := (*pollDesc)(unsafe.Pointer(ev.udata))
   174  			pd.everr = false
   175  			if ev.flags == _EV_ERROR {
   176  				pd.everr = true
   177  			}
   178  			netpollready(&toRun, pd, mode)
   179  		}
   180  	}
   181  	return toRun
   182  }
   183  

View as plain text