Imported Upstream version 1.1~hg20130323upstream/1.1_hg20130323

1 files changed, 59 insertions, 304 deletions

diff --git a/src/pkg/net/fd_unix.go b/src/pkg/net/fd_unix.go
index 0540df825..2b418a868 100644
--- a/src/pkg/net/fd_unix.go
+++ b/src/pkg/net/fd_unix.go
@@ -9,7 +9,6 @@ package net
 import (
 	"io"
 	"os"
-	"runtime"
 	"sync"
 	"syscall"
 	"time"
@@ -21,7 +20,7 @@ type netFD struct {
 	sysmu  sync.Mutex
 	sysref int
 
-	// must lock both sysmu and pollserver to write
+	// must lock both sysmu and pollDesc to write
 	// can lock either to read
 	closing bool
 
@@ -31,8 +30,6 @@ type netFD struct {
 	sotype      int
 	isConnected bool
 	sysfile     *os.File
-	cr          chan error
-	cw          chan error
 	net         string
 	laddr       Addr
 	raddr       Addr
@@ -40,255 +37,8 @@ type netFD struct {
 	// serialize access to Read and Write methods
 	rio, wio sync.Mutex
 
-	// read and write deadlines
-	rdeadline, wdeadline deadline
-
-	// owned by fd wait server
-	ncr, ncw int
-
 	// wait server
-	pollServer *pollServer
-}
-
-// A pollServer helps FDs determine when to retry a non-blocking
-// read or write after they get EAGAIN.  When an FD needs to wait,
-// call s.WaitRead() or s.WaitWrite() to pass the request to the poll server.
-// When the pollServer finds that i/o on FD should be possible
-// again, it will send on fd.cr/fd.cw to wake any waiting goroutines.
-//
-// To avoid races in closing, all fd operations are locked and
-// refcounted. when netFD.Close() is called, it calls syscall.Shutdown
-// and sets a closing flag. Only when the last reference is removed
-// will the fd be closed.
-
-type pollServer struct {
-	pr, pw     *os.File
-	poll       *pollster // low-level OS hooks
-	sync.Mutex           // controls pending and deadline
-	pending    map[int]*netFD
-	deadline   int64 // next deadline (nsec since 1970)
-}
-
-func (s *pollServer) AddFD(fd *netFD, mode int) error {
-	s.Lock()
-	intfd := fd.sysfd
-	if intfd < 0 || fd.closing {
-		// fd closed underfoot
-		s.Unlock()
-		return errClosing
-	}
-
-	var t int64
-	key := intfd << 1
-	if mode == 'r' {
-		fd.ncr++
-		t = fd.rdeadline.value()
-	} else {
-		fd.ncw++
-		key++
-		t = fd.wdeadline.value()
-	}
-	s.pending[key] = fd
-	doWakeup := false
-	if t > 0 && (s.deadline == 0 || t < s.deadline) {
-		s.deadline = t
-		doWakeup = true
-	}
-
-	wake, err := s.poll.AddFD(intfd, mode, false)
-	s.Unlock()
-	if err != nil {
-		return &OpError{"addfd", fd.net, fd.laddr, err}
-	}
-	if wake || doWakeup {
-		s.Wakeup()
-	}
-	return nil
-}
-
-// Evict evicts fd from the pending list, unblocking
-// any I/O running on fd.  The caller must have locked
-// pollserver.
-// Return value is whether the pollServer should be woken up.
-func (s *pollServer) Evict(fd *netFD) bool {
-	doWakeup := false
-	if s.pending[fd.sysfd<<1] == fd {
-		s.WakeFD(fd, 'r', errClosing)
-		if s.poll.DelFD(fd.sysfd, 'r') {
-			doWakeup = true
-		}
-		delete(s.pending, fd.sysfd<<1)
-	}
-	if s.pending[fd.sysfd<<1|1] == fd {
-		s.WakeFD(fd, 'w', errClosing)
-		if s.poll.DelFD(fd.sysfd, 'w') {
-			doWakeup = true
-		}
-		delete(s.pending, fd.sysfd<<1|1)
-	}
-	return doWakeup
-}
-
-var wakeupbuf [1]byte
-
-func (s *pollServer) Wakeup() { s.pw.Write(wakeupbuf[0:]) }
-
-func (s *pollServer) LookupFD(fd int, mode int) *netFD {
-	key := fd << 1
-	if mode == 'w' {
-		key++
-	}
-	netfd, ok := s.pending[key]
-	if !ok {
-		return nil
-	}
-	delete(s.pending, key)
-	return netfd
-}
-
-func (s *pollServer) WakeFD(fd *netFD, mode int, err error) {
-	if mode == 'r' {
-		for fd.ncr > 0 {
-			fd.ncr--
-			fd.cr <- err
-		}
-	} else {
-		for fd.ncw > 0 {
-			fd.ncw--
-			fd.cw <- err
-		}
-	}
-}
-
-func (s *pollServer) CheckDeadlines() {
-	now := time.Now().UnixNano()
-	// TODO(rsc): This will need to be handled more efficiently,
-	// probably with a heap indexed by wakeup time.
-
-	var nextDeadline int64
-	for key, fd := range s.pending {
-		var t int64
-		var mode int
-		if key&1 == 0 {
-			mode = 'r'
-		} else {
-			mode = 'w'
-		}
-		if mode == 'r' {
-			t = fd.rdeadline.value()
-		} else {
-			t = fd.wdeadline.value()
-		}
-		if t > 0 {
-			if t <= now {
-				delete(s.pending, key)
-				s.poll.DelFD(fd.sysfd, mode)
-				s.WakeFD(fd, mode, errTimeout)
-			} else if nextDeadline == 0 || t < nextDeadline {
-				nextDeadline = t
-			}
-		}
-	}
-	s.deadline = nextDeadline
-}
-
-func (s *pollServer) Run() {
-	var scratch [100]byte
-	s.Lock()
-	defer s.Unlock()
-	for {
-		var timeout int64 // nsec to wait for or 0 for none
-		if s.deadline > 0 {
-			timeout = s.deadline - time.Now().UnixNano()
-			if timeout <= 0 {
-				s.CheckDeadlines()
-				continue
-			}
-		}
-		fd, mode, err := s.poll.WaitFD(s, timeout)
-		if err != nil {
-			print("pollServer WaitFD: ", err.Error(), "\n")
-			return
-		}
-		if fd < 0 {
-			// Timeout happened.
-			s.CheckDeadlines()
-			continue
-		}
-		if fd == int(s.pr.Fd()) {
-			// Drain our wakeup pipe (we could loop here,
-			// but it's unlikely that there are more than
-			// len(scratch) wakeup calls).
-			s.pr.Read(scratch[0:])
-			s.CheckDeadlines()
-		} else {
-			netfd := s.LookupFD(fd, mode)
-			if netfd == nil {
-				// This can happen because the WaitFD runs without
-				// holding s's lock, so there might be a pending wakeup
-				// for an fd that has been evicted.  No harm done.
-				continue
-			}
-			s.WakeFD(netfd, mode, nil)
-		}
-	}
-}
-
-func (s *pollServer) WaitRead(fd *netFD) error {
-	err := s.AddFD(fd, 'r')
-	if err == nil {
-		err = <-fd.cr
-	}
-	return err
-}
-
-func (s *pollServer) WaitWrite(fd *netFD) error {
-	err := s.AddFD(fd, 'w')
-	if err == nil {
-		err = <-fd.cw
-	}
-	return err
-}
-
-// Network FD methods.
-// Spread network FDs over several pollServers.
-
-var pollMaxN int
-var pollservers []*pollServer
-var startServersOnce []func()
-
-var canCancelIO = true // used for testing current package
-
-func sysInit() {
-	pollMaxN = runtime.NumCPU()
-	if pollMaxN > 8 {
-		pollMaxN = 8 // No improvement then.
-	}
-	pollservers = make([]*pollServer, pollMaxN)
-	startServersOnce = make([]func(), pollMaxN)
-	for i := 0; i < pollMaxN; i++ {
-		k := i
-		once := new(sync.Once)
-		startServersOnce[i] = func() { once.Do(func() { startServer(k) }) }
-	}
-}
-
-func startServer(k int) {
-	p, err := newPollServer()
-	if err != nil {
-		panic(err)
-	}
-	pollservers[k] = p
-}
-
-func server(fd int) *pollServer {
-	pollN := runtime.GOMAXPROCS(0)
-	if pollN > pollMaxN {
-		pollN = pollMaxN
-	}
-	k := fd % pollN
-	startServersOnce[k]()
-	return pollservers[k]
+	pd pollDesc
 }
 
 func dialTimeout(net, addr string, timeout time.Duration) (Conn, error) {
@@ -307,9 +57,9 @@ func newFD(fd, family, sotype int, net string) (*netFD, error) {
 		sotype: sotype,
 		net:    net,
 	}
-	netfd.cr = make(chan error, 1)
-	netfd.cw = make(chan error, 1)
-	netfd.pollServer = server(fd)
+	if err := netfd.pd.Init(netfd); err != nil {
+		return nil, err
+	}
 	return netfd, nil
 }
 
@@ -331,25 +81,28 @@ func (fd *netFD) name() string {
 }
 
 func (fd *netFD) connect(ra syscall.Sockaddr) error {
-	err := syscall.Connect(fd.sysfd, ra)
-	if err == syscall.EINPROGRESS {
-		if err = fd.pollServer.WaitWrite(fd); err != nil {
-			return err
+	fd.wio.Lock()
+	defer fd.wio.Unlock()
+	if err := fd.pd.PrepareWrite(); err != nil {
+		return err
+	}
+	for {
+		err := syscall.Connect(fd.sysfd, ra)
+		if err == nil || err == syscall.EISCONN {
+			break
 		}
-		var e int
-		e, err = syscall.GetsockoptInt(fd.sysfd, syscall.SOL_SOCKET, syscall.SO_ERROR)
-		if err != nil {
-			return os.NewSyscallError("getsockopt", err)
+		if err != syscall.EINPROGRESS && err != syscall.EALREADY && err != syscall.EINTR {
+			return err
 		}
-		if e != 0 {
-			err = syscall.Errno(e)
+		if err = fd.pd.WaitWrite(); err != nil {
+			return err
 		}
 	}
-	return err
+	return nil
 }
 
 // Add a reference to this fd.
-// If closing==true, pollserver must be locked; mark the fd as closing.
+// If closing==true, pollDesc must be locked; mark the fd as closing.
 // Returns an error if the fd cannot be used.
 func (fd *netFD) incref(closing bool) error {
 	fd.sysmu.Lock()
@@ -371,6 +124,9 @@ func (fd *netFD) decref() {
 	fd.sysmu.Lock()
 	fd.sysref--
 	if fd.closing && fd.sysref == 0 && fd.sysfile != nil {
+		// Poller may want to unregister fd in readiness notification mechanism,
+		// so this must be executed before sysfile.Close().
+		fd.pd.Close()
 		fd.sysfile.Close()
 		fd.sysfile = nil
 		fd.sysfd = -1
@@ -379,21 +135,21 @@ func (fd *netFD) decref() {
 }
 
 func (fd *netFD) Close() error {
-	fd.pollServer.Lock() // needed for both fd.incref(true) and pollserver.Evict
+	fd.pd.Lock() // needed for both fd.incref(true) and pollDesc.Evict
 	if err := fd.incref(true); err != nil {
-		fd.pollServer.Unlock()
+		fd.pd.Unlock()
 		return err
 	}
 	// Unblock any I/O.  Once it all unblocks and returns,
 	// so that it cannot be referring to fd.sysfd anymore,
 	// the final decref will close fd.sysfd.  This should happen
 	// fairly quickly, since all the I/O is non-blocking, and any
-	// attempts to block in the pollserver will return errClosing.
-	doWakeup := fd.pollServer.Evict(fd)
-	fd.pollServer.Unlock()
+	// attempts to block in the pollDesc will return errClosing.
+	doWakeup := fd.pd.Evict()
+	fd.pd.Unlock()
 	fd.decref()
 	if doWakeup {
-		fd.pollServer.Wakeup()
+		fd.pd.Wakeup()
 	}
 	return nil
 }
@@ -425,16 +181,15 @@ func (fd *netFD) Read(p []byte) (n int, err error) {
 		return 0, err
 	}
 	defer fd.decref()
+	if err := fd.pd.PrepareRead(); err != nil {
+		return 0, &OpError{"read", fd.net, fd.raddr, err}
+	}
 	for {
-		if fd.rdeadline.expired() {
-			err = errTimeout
-			break
-		}
 		n, err = syscall.Read(int(fd.sysfd), p)
 		if err != nil {
 			n = 0
 			if err == syscall.EAGAIN {
-				if err = fd.pollServer.WaitRead(fd); err == nil {
+				if err = fd.pd.WaitRead(); err == nil {
 					continue
 				}
 			}
@@ -455,16 +210,15 @@ func (fd *netFD) ReadFrom(p []byte) (n int, sa syscall.Sockaddr, err error) {
 		return 0, nil, err
 	}
 	defer fd.decref()
+	if err := fd.pd.PrepareRead(); err != nil {
+		return 0, nil, &OpError{"read", fd.net, fd.laddr, err}
+	}
 	for {
-		if fd.rdeadline.expired() {
-			err = errTimeout
-			break
-		}
 		n, sa, err = syscall.Recvfrom(fd.sysfd, p, 0)
 		if err != nil {
 			n = 0
 			if err == syscall.EAGAIN {
-				if err = fd.pollServer.WaitRead(fd); err == nil {
+				if err = fd.pd.WaitRead(); err == nil {
 					continue
 				}
 			}
@@ -485,16 +239,15 @@ func (fd *netFD) ReadMsg(p []byte, oob []byte) (n, oobn, flags int, sa syscall.S
 		return 0, 0, 0, nil, err
 	}
 	defer fd.decref()
+	if err := fd.pd.PrepareRead(); err != nil {
+		return 0, 0, 0, nil, &OpError{"read", fd.net, fd.laddr, err}
+	}
 	for {
-		if fd.rdeadline.expired() {
-			err = errTimeout
-			break
-		}
 		n, oobn, flags, sa, err = syscall.Recvmsg(fd.sysfd, p, oob, 0)
 		if err != nil {
 			// TODO(dfc) should n and oobn be set to 0
 			if err == syscall.EAGAIN {
-				if err = fd.pollServer.WaitRead(fd); err == nil {
+				if err = fd.pd.WaitRead(); err == nil {
 					continue
 				}
 			}
@@ -522,11 +275,10 @@ func (fd *netFD) Write(p []byte) (nn int, err error) {
 		return 0, err
 	}
 	defer fd.decref()
+	if err := fd.pd.PrepareWrite(); err != nil {
+		return 0, &OpError{"write", fd.net, fd.raddr, err}
+	}
 	for {
-		if fd.wdeadline.expired() {
-			err = errTimeout
-			break
-		}
 		var n int
 		n, err = syscall.Write(int(fd.sysfd), p[nn:])
 		if n > 0 {
@@ -536,7 +288,7 @@ func (fd *netFD) Write(p []byte) (nn int, err error) {
 			break
 		}
 		if err == syscall.EAGAIN {
-			if err = fd.pollServer.WaitWrite(fd); err == nil {
+			if err = fd.pd.WaitWrite(); err == nil {
 				continue
 			}
 		}
@@ -562,14 +314,13 @@ func (fd *netFD) WriteTo(p []byte, sa syscall.Sockaddr) (n int, err error) {
 		return 0, err
 	}
 	defer fd.decref()
+	if err := fd.pd.PrepareWrite(); err != nil {
+		return 0, &OpError{"write", fd.net, fd.raddr, err}
+	}
 	for {
-		if fd.wdeadline.expired() {
-			err = errTimeout
-			break
-		}
 		err = syscall.Sendto(fd.sysfd, p, 0, sa)
 		if err == syscall.EAGAIN {
-			if err = fd.pollServer.WaitWrite(fd); err == nil {
+			if err = fd.pd.WaitWrite(); err == nil {
 				continue
 			}
 		}
@@ -590,14 +341,13 @@ func (fd *netFD) WriteMsg(p []byte, oob []byte, sa syscall.Sockaddr) (n int, oob
 		return 0, 0, err
 	}
 	defer fd.decref()
+	if err := fd.pd.PrepareWrite(); err != nil {
+		return 0, 0, &OpError{"write", fd.net, fd.raddr, err}
+	}
 	for {
-		if fd.wdeadline.expired() {
-			err = errTimeout
-			break
-		}
 		err = syscall.Sendmsg(fd.sysfd, p, oob, sa, 0)
 		if err == syscall.EAGAIN {
-			if err = fd.pollServer.WaitWrite(fd); err == nil {
+			if err = fd.pd.WaitWrite(); err == nil {
 				continue
 			}
 		}
@@ -613,6 +363,8 @@ func (fd *netFD) WriteMsg(p []byte, oob []byte, sa syscall.Sockaddr) (n int, oob
 }
 
 func (fd *netFD) accept(toAddr func(syscall.Sockaddr) Addr) (netfd *netFD, err error) {
+	fd.rio.Lock()
+	defer fd.rio.Unlock()
 	if err := fd.incref(false); err != nil {
 		return nil, err
 	}
@@ -620,11 +372,14 @@ func (fd *netFD) accept(toAddr func(syscall.Sockaddr) Addr) (netfd *netFD, err e
 
 	var s int
 	var rsa syscall.Sockaddr
+	if err = fd.pd.PrepareRead(); err != nil {
+		return nil, &OpError{"accept", fd.net, fd.laddr, err}
+	}
 	for {
 		s, rsa, err = accept(fd.sysfd)
 		if err != nil {
 			if err == syscall.EAGAIN {
-				if err = fd.pollServer.WaitRead(fd); err == nil {
+				if err = fd.pd.WaitRead(); err == nil {
 					continue
 				}
 			} else if err == syscall.ECONNABORTED {