code review 164057: 'close' race fixed/suggestions from rcs

src/pkg/net/fd.go

 // Copyright 2009 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.
 
 // TODO(rsc): All the prints in this file should go to standard error.
 
 package net
 
 import (
         "once";
         "os";
         "sync";
         "syscall";
 )
-
-import "fmt"    // XXX
 
 // Network file descriptor.
 type netFD struct {
         // locking/lifetime of sysfd
         sysmu   sync.Mutex;
         sysref  int;
         closing bool;
 
         // immutable until Close
         sysfd   int;
@@ skipping 40 matching lines @@
 // If the operations were reversed, there would be a race: the poll
 // server might wake up and look at the request channel, see that it
 // was empty, and go back to sleep, all before the requester managed
 // to send the request.  Because the send must complete before the wakeup,
 // the request channel must be buffered.  A buffer of size 1 is sufficient
 // for any request load.  If many processes are trying to submit requests,
 // one will succeed, the pollServer will read the request, and then the
 // channel will be empty for the next process's request.  A larger buffer
 // might help batch requests.
 //
-// In order to prevent race conditions, pollServer has an additional cc channel
-// that receives fds to be closed. pollServer doesn't make the close system
-// call, it just sets fd.sysfile = nil and fd.fd = -1. Because of this,
-// pollServer is always in sync with the kernel's view of a given descriptor.
+// 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 {
         cr, cw          chan *netFD;    // buffered >= 1
         pr, pw          *os.File;
         pending         map[int]*netFD;
         poll            *pollster;      // low-level OS hooks
         deadline        int64;          // next deadline (nsec since 1970)
 }
 
 func newPollServer() (s *pollServer, err os.Error) {
@@ skipping 21 matching lines @@
         if err = s.poll.AddFD(s.pr.Fd(), 'r', true); err != nil {
                 s.poll.Close();
                 goto Error;
         }
         s.pending = make(map[int]*netFD);
         go s.Run();
         return s, nil;
 }
 
 func (s *pollServer) AddFD(fd *netFD, mode int) {
-        // This check verifies that the underlying file descriptor hasn't been
-        // closed in the mean time. Any time a netFD is closed, the closing
-        // goroutine makes a round trip to the pollServer which sets file = nil
-        // and fd = -1. The goroutine then closes the actual file descriptor.
-        // Thus fd.fd mirrors the kernel's view of the file descriptor.
-
-        // TODO(rsc,agl): There is still a race in Read and Write,
-        // because they optimistically try to use the fd and don't
-        // call into the PollServer unless they get EAGAIN.
         intfd := fd.sysfd;
         if intfd < 0 {
                 // fd closed underfoot
                 if mode == 'r' {
                         fd.cr <- fd
                 } else {
                         fd.cw <- fd
                 }
                 return;
         }
@@ skipping 54 matching lines @@
         return nsec;
 }
 
 func (s *pollServer) CheckDeadlines() {
         now := s.Now();
         // TODO(rsc): This will need to be handled more efficiently,
         // probably with a heap indexed by wakeup time.
 
         var next_deadline int64;
         for key, fd := range s.pending {
-                fmt.Printf("XXX in poll server\n");
                 var t int64;
                 var mode int;
                 if key&1 == 0 {
                         mode = 'r'
                 } else {
                         mode = 'w'
                 }
                 if mode == 'r' {
                         t = fd.rdeadline
                 } else {
                         t = fd.wdeadline
                 }
                 if t > 0 {
                         if t <= now {
+                                fd.incref();

[rsc, 2009/12/01 19:58:56]

I don't think this needs incref/decref.
If the fd has been passed to the poll server,
the goroutine who gave it to the poll server is
in charge of keeping the ref count up.

(If it did need an incref, that would be a bug,
because the kernel has the sysfd right now too.)

                                 s.pending[key] = nil, false;
                                 if mode == 'r' {
                                         s.poll.DelFD(fd.sysfd, mode);
                                         fd.rdeadline = -1;
                                 } else {
                                         s.poll.DelFD(fd.sysfd, mode);
                                         fd.wdeadline = -1;
                                 }
                                 s.WakeFD(fd, mode);
+                                fd.decref();
                         } else if next_deadline == 0 || t < next_deadline {
                                 next_deadline = t
                         }
                 }
         }
         s.deadline = next_deadline;
 }
 
 func (s *pollServer) Run() {
         var scratch [100]byte;
@@ skipping 14 matching lines @@
                 if fd < 0 {
                         // Timeout happened.
                         s.CheckDeadlines();
                         continue;
                 }
                 if fd == s.pr.Fd() {
                         // Drain our wakeup pipe.
                         for nn, _ := s.pr.Read(&scratch); nn > 0; {
                                 nn, _ = s.pr.Read(&scratch)
                         }
-
                         // Read from channels
                         for fd, ok := <-s.cr; ok; fd, ok = <-s.cr {
                                 s.AddFD(fd, 'r')
                         }
                         for fd, ok := <-s.cw; ok; fd, ok = <-s.cw {
                                 s.AddFD(fd, 'w')
                         }
                 } else {
                         netfd := s.LookupFD(fd, mode);
+
                         if netfd == nil {
                                 print("pollServer: unexpected wakeup for fd=", netfd, " mode=", string(mode), "\n");
                                 continue;
                         }
                         s.WakeFD(netfd, mode);
                 }
         }
 }
 
 var wakeupbuf [1]byte
@@ skipping 44 matching lines @@
         }
         if raddr != nil {
                 rs = raddr.String()
         }
         f.sysfile = os.NewFile(fd, net+":"+ls+"->"+rs);
         f.cr = make(chan *netFD, 1);
         f.cw = make(chan *netFD, 1);
         return f, nil;
 }
 
-// locksysfd allows the caller to use sysfd.  When finished, the
-// caller must call unlocksysfd.
-func (fd *netFD) locksysfd() {
+// Add a reference to this fd.
+func (fd *netFD) incref() {
         fd.sysmu.Lock();
         fd.sysref++;
         fd.sysmu.Unlock();
 }
 
-// unlocksysfd releases sysfd.  After calling this, the caller should
-// not refer to fd.sysfd again without calling locksysfd.
-func (fd *netFD) unlocksysfd() {
+// Remove a reference to this FD and close if we've been asked to do so (and
+// there are no references left.
+func (fd *netFD) decref() {
         fd.sysmu.Lock();
-        if fd.sysref--; fd.sysref == 0 && fd.closing && fd.sysfd >= 0 {
-
+        fd.sysref--;
+        if fd.closing && fd.sysref == 0 && fd.sysfd >= 0 {
                 // In case the user has set linger, switch to blocking mode so
                 // the close blocks.  As long as this doesn't happen often, we
                 // can handle the extra OS processes.  Otherwise we'll need to
                 // use the pollserver for Close too.  Sigh.
                 syscall.SetNonblock(fd.sysfd, false);
                 fd.sysfile.Close();
-                fd.sysfile = nil;       // XXX needed?
+                fd.sysfile = nil;
                 fd.sysfd = -1;
         }
         fd.sysmu.Unlock();
 }
 
 func isEAGAIN(e os.Error) bool {
         if e1, ok := e.(*os.PathError); ok {
                 return e1.Error == os.EAGAIN
         }
         return e == os.EAGAIN;
 }
 
 func (fd *netFD) Close() os.Error {
-        fd.locksysfd();
+        if fd == nil || fd.sysfile == nil {
+                return os.EINVAL
+        }
+
+        fd.incref();
+        syscall.Shutdown(fd.sysfd, syscall.SHUT_RDWR);
         fd.closing = true;
-        fd.unlocksysfd();
+        fd.decref();
         return nil;
 }
 
 func (fd *netFD) Read(p []byte) (n int, err os.Error) {
         if fd == nil || fd.sysfile == nil {
                 return 0, os.EINVAL
         }
         fd.rio.Lock();
         defer fd.rio.Unlock();
-        fd.locksysfd();
-        defer fd.unlocksysfd();
+        fd.incref();
+        defer fd.decref();
         if fd.rdeadline_delta > 0 {
                 fd.rdeadline = pollserver.Now() + fd.rdeadline_delta
         } else {
                 fd.rdeadline = 0
         }
         for {
-                n, err = fd.sysfile.Read(p);    // fd.sysfile
+                n, err = fd.sysfile.Read(p);
                 if isEAGAIN(err) && fd.rdeadline >= 0 {
                         pollserver.WaitRead(fd);
                         continue;
                 }
                 break;
         }
         return;
 }
 
 func (fd *netFD) ReadFrom(p []byte) (n int, sa syscall.Sockaddr, err os.Error) {
         if fd == nil || fd.sysfile == nil {
                 return 0, nil, os.EINVAL
         }
         fd.rio.Lock();
         defer fd.rio.Unlock();
-        fd.locksysfd();
-        defer fd.unlocksysfd();
+        fd.incref();
+        defer fd.decref();
         if fd.rdeadline_delta > 0 {
                 fd.rdeadline = pollserver.Now() + fd.rdeadline_delta
         } else {
                 fd.rdeadline = 0
         }
         for {
                 var errno int;
                 n, sa, errno = syscall.Recvfrom(fd.sysfd, p, 0);
                 if errno == syscall.EAGAIN && fd.rdeadline >= 0 {
                         pollserver.WaitRead(fd);
                         continue;
                 }
                 if errno != 0 {
                         n = 0;
                         err = &os.PathError{"recvfrom", fd.sysfile.Name(), os.Errno(errno)};
                 }
                 break;
         }
         return;
 }
 
 func (fd *netFD) Write(p []byte) (n int, err os.Error) {
         if fd == nil || fd.sysfile == nil {
                 return 0, os.EINVAL
         }
         fd.wio.Lock();
         defer fd.wio.Unlock();
-        fd.locksysfd();
-        defer fd.unlocksysfd();
+        fd.incref();
+        defer fd.decref();
         if fd.wdeadline_delta > 0 {
                 fd.wdeadline = pollserver.Now() + fd.wdeadline_delta
         } else {
                 fd.wdeadline = 0
         }
         err = nil;
         nn := 0;
         for nn < len(p) {
                 n, err = fd.sysfile.Write(p[nn:]);
                 if n > 0 {
@@ skipping 12 matching lines @@
         }
         return nn, err;
 }
 
 func (fd *netFD) WriteTo(p []byte, sa syscall.Sockaddr) (n int, err os.Error) {
         if fd == nil || fd.sysfile == nil {
                 return 0, os.EINVAL
         }
         fd.wio.Lock();
         defer fd.wio.Unlock();
-        fd.locksysfd();
-        defer fd.unlocksysfd();
+        fd.incref();
+        defer fd.decref();
         if fd.wdeadline_delta > 0 {
                 fd.wdeadline = pollserver.Now() + fd.wdeadline_delta
         } else {
                 fd.wdeadline = 0
         }
         err = nil;
         for {
                 errno := syscall.Sendto(fd.sysfd, p, 0, sa);
                 if errno == syscall.EAGAIN && fd.wdeadline >= 0 {
                         pollserver.WaitWrite(fd);
@@ skipping 14 matching lines @@
         if fd == nil || fd.sysfile == nil {
                 return nil, os.EINVAL
         }
 
         // See ../syscall/exec.go for description of ForkLock.
         // It is okay to hold the lock across syscall.Accept
         // because we have put fd.sysfd into non-blocking mode.
         syscall.ForkLock.RLock();
         var s, e int;
         var sa syscall.Sockaddr;
+        fd.incref();

[rsc, 2009/12/01 19:58:56]

Please move these two lines up above the comment 
about syscall/exec.go.  Once syscall.ForkLock.RLock()
has been called, no other locks should be acquired,
and there's one in incref.

+        defer fd.decref();
         for {
-                fd.locksysfd();
                 s, sa, e = syscall.Accept(fd.sysfd);
-                fd.unlocksysfd();
                 if e != syscall.EAGAIN {
                         break
                 }
                 syscall.ForkLock.RUnlock();
                 pollserver.WaitRead(fd);
                 syscall.ForkLock.RLock();
         }
         if e != 0 {
                 syscall.ForkLock.RUnlock();
                 return nil, &OpError{"accept", fd.net, fd.laddr, os.Errno(e)};
         }
         syscall.CloseOnExec(s);
         syscall.ForkLock.RUnlock();
 
         if nfd, err = newFD(s, fd.family, fd.proto, fd.net, fd.laddr, toAddr(sa)); err != nil {
                 syscall.Close(s);
                 return nil, err;
         }
         return nfd, nil;
 }