Source file src/pkg/net/fd.go

     1	// Copyright 2009 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 freebsd linux netbsd openbsd
     6	
     7	package net
     8	
     9	import (
    10		"errors"
    11		"io"
    12		"os"
    13		"sync"
    14		"syscall"
    15		"time"
    16	)
    17	
    18	// Network file descriptor.
    19	type netFD struct {
    20		// locking/lifetime of sysfd
    21		sysmu  sync.Mutex
    22		sysref int
    23	
    24		// must lock both sysmu and pollserver to write
    25		// can lock either to read
    26		closing bool
    27	
    28		// immutable until Close
    29		sysfd       int
    30		family      int
    31		sotype      int
    32		isConnected bool
    33		sysfile     *os.File
    34		cr          chan error
    35		cw          chan error
    36		net         string
    37		laddr       Addr
    38		raddr       Addr
    39	
    40		// owned by client
    41		rdeadline int64
    42		rio       sync.Mutex
    43		wdeadline int64
    44		wio       sync.Mutex
    45	
    46		// owned by fd wait server
    47		ncr, ncw int
    48	}
    49	
    50	// A pollServer helps FDs determine when to retry a non-blocking
    51	// read or write after they get EAGAIN.  When an FD needs to wait,
    52	// send the fd on s.cr (for a read) or s.cw (for a write) to pass the
    53	// request to the poll server.  Then receive on fd.cr/fd.cw.
    54	// When the pollServer finds that i/o on FD should be possible
    55	// again, it will send fd on fd.cr/fd.cw to wake any waiting processes.
    56	// This protocol is implemented as s.WaitRead() and s.WaitWrite().
    57	//
    58	// There is one subtlety: when sending on s.cr/s.cw, the
    59	// poll server is probably in a system call, waiting for an fd
    60	// to become ready.  It's not looking at the request channels.
    61	// To resolve this, the poll server waits not just on the FDs it has
    62	// been given but also its own pipe.  After sending on the
    63	// buffered channel s.cr/s.cw, WaitRead/WaitWrite writes a
    64	// byte to the pipe, causing the pollServer's poll system call to
    65	// return.  In response to the pipe being readable, the pollServer
    66	// re-polls its request channels.
    67	//
    68	// Note that the ordering is "send request" and then "wake up server".
    69	// If the operations were reversed, there would be a race: the poll
    70	// server might wake up and look at the request channel, see that it
    71	// was empty, and go back to sleep, all before the requester managed
    72	// to send the request.  Because the send must complete before the wakeup,
    73	// the request channel must be buffered.  A buffer of size 1 is sufficient
    74	// for any request load.  If many processes are trying to submit requests,
    75	// one will succeed, the pollServer will read the request, and then the
    76	// channel will be empty for the next process's request.  A larger buffer
    77	// might help batch requests.
    78	//
    79	// To avoid races in closing, all fd operations are locked and
    80	// refcounted. when netFD.Close() is called, it calls syscall.Shutdown
    81	// and sets a closing flag. Only when the last reference is removed
    82	// will the fd be closed.
    83	
    84	type pollServer struct {
    85		cr, cw     chan *netFD // buffered >= 1
    86		pr, pw     *os.File
    87		poll       *pollster // low-level OS hooks
    88		sync.Mutex           // controls pending and deadline
    89		pending    map[int]*netFD
    90		deadline   int64 // next deadline (nsec since 1970)
    91	}
    92	
    93	func (s *pollServer) AddFD(fd *netFD, mode int) error {
    94		s.Lock()
    95		intfd := fd.sysfd
    96		if intfd < 0 || fd.closing {
    97			// fd closed underfoot
    98			s.Unlock()
    99			return errClosing
   100		}
   101	
   102		var t int64
   103		key := intfd << 1
   104		if mode == 'r' {
   105			fd.ncr++
   106			t = fd.rdeadline
   107		} else {
   108			fd.ncw++
   109			key++
   110			t = fd.wdeadline
   111		}
   112		s.pending[key] = fd
   113		doWakeup := false
   114		if t > 0 && (s.deadline == 0 || t < s.deadline) {
   115			s.deadline = t
   116			doWakeup = true
   117		}
   118	
   119		wake, err := s.poll.AddFD(intfd, mode, false)
   120		if err != nil {
   121			panic("pollServer AddFD " + err.Error())
   122		}
   123		if wake {
   124			doWakeup = true
   125		}
   126		s.Unlock()
   127	
   128		if doWakeup {
   129			s.Wakeup()
   130		}
   131		return nil
   132	}
   133	
   134	// Evict evicts fd from the pending list, unblocking
   135	// any I/O running on fd.  The caller must have locked
   136	// pollserver.
   137	func (s *pollServer) Evict(fd *netFD) {
   138		if s.pending[fd.sysfd<<1] == fd {
   139			s.WakeFD(fd, 'r', errClosing)
   140			s.poll.DelFD(fd.sysfd, 'r')
   141			delete(s.pending, fd.sysfd<<1)
   142		}
   143		if s.pending[fd.sysfd<<1|1] == fd {
   144			s.WakeFD(fd, 'w', errClosing)
   145			s.poll.DelFD(fd.sysfd, 'w')
   146			delete(s.pending, fd.sysfd<<1|1)
   147		}
   148	}
   149	
   150	var wakeupbuf [1]byte
   151	
   152	func (s *pollServer) Wakeup() { s.pw.Write(wakeupbuf[0:]) }
   153	
   154	func (s *pollServer) LookupFD(fd int, mode int) *netFD {
   155		key := fd << 1
   156		if mode == 'w' {
   157			key++
   158		}
   159		netfd, ok := s.pending[key]
   160		if !ok {
   161			return nil
   162		}
   163		delete(s.pending, key)
   164		return netfd
   165	}
   166	
   167	func (s *pollServer) WakeFD(fd *netFD, mode int, err error) {
   168		if mode == 'r' {
   169			for fd.ncr > 0 {
   170				fd.ncr--
   171				fd.cr <- err
   172			}
   173		} else {
   174			for fd.ncw > 0 {
   175				fd.ncw--
   176				fd.cw <- err
   177			}
   178		}
   179	}
   180	
   181	func (s *pollServer) Now() int64 {
   182		return time.Now().UnixNano()
   183	}
   184	
   185	func (s *pollServer) CheckDeadlines() {
   186		now := s.Now()
   187		// TODO(rsc): This will need to be handled more efficiently,
   188		// probably with a heap indexed by wakeup time.
   189	
   190		var next_deadline int64
   191		for key, fd := range s.pending {
   192			var t int64
   193			var mode int
   194			if key&1 == 0 {
   195				mode = 'r'
   196			} else {
   197				mode = 'w'
   198			}
   199			if mode == 'r' {
   200				t = fd.rdeadline
   201			} else {
   202				t = fd.wdeadline
   203			}
   204			if t > 0 {
   205				if t <= now {
   206					delete(s.pending, key)
   207					if mode == 'r' {
   208						s.poll.DelFD(fd.sysfd, mode)
   209						fd.rdeadline = -1
   210					} else {
   211						s.poll.DelFD(fd.sysfd, mode)
   212						fd.wdeadline = -1
   213					}
   214					s.WakeFD(fd, mode, nil)
   215				} else if next_deadline == 0 || t < next_deadline {
   216					next_deadline = t
   217				}
   218			}
   219		}
   220		s.deadline = next_deadline
   221	}
   222	
   223	func (s *pollServer) Run() {
   224		var scratch [100]byte
   225		s.Lock()
   226		defer s.Unlock()
   227		for {
   228			var t = s.deadline
   229			if t > 0 {
   230				t = t - s.Now()
   231				if t <= 0 {
   232					s.CheckDeadlines()
   233					continue
   234				}
   235			}
   236			fd, mode, err := s.poll.WaitFD(s, t)
   237			if err != nil {
   238				print("pollServer WaitFD: ", err.Error(), "\n")
   239				return
   240			}
   241			if fd < 0 {
   242				// Timeout happened.
   243				s.CheckDeadlines()
   244				continue
   245			}
   246			if fd == int(s.pr.Fd()) {
   247				// Drain our wakeup pipe (we could loop here,
   248				// but it's unlikely that there are more than
   249				// len(scratch) wakeup calls).
   250				s.pr.Read(scratch[0:])
   251				s.CheckDeadlines()
   252			} else {
   253				netfd := s.LookupFD(fd, mode)
   254				if netfd == nil {
   255					// This can happen because the WaitFD runs without
   256					// holding s's lock, so there might be a pending wakeup
   257					// for an fd that has been evicted.  No harm done.
   258					continue
   259				}
   260				s.WakeFD(netfd, mode, nil)
   261			}
   262		}
   263	}
   264	
   265	func (s *pollServer) WaitRead(fd *netFD) error {
   266		err := s.AddFD(fd, 'r')
   267		if err == nil {
   268			err = <-fd.cr
   269		}
   270		return err
   271	}
   272	
   273	func (s *pollServer) WaitWrite(fd *netFD) error {
   274		err := s.AddFD(fd, 'w')
   275		if err == nil {
   276			err = <-fd.cw
   277		}
   278		return err
   279	}
   280	
   281	// Network FD methods.
   282	// All the network FDs use a single pollServer.
   283	
   284	var pollserver *pollServer
   285	var onceStartServer sync.Once
   286	
   287	func startServer() {
   288		p, err := newPollServer()
   289		if err != nil {
   290			print("Start pollServer: ", err.Error(), "\n")
   291		}
   292		pollserver = p
   293	}
   294	
   295	func newFD(fd, family, sotype int, net string) (*netFD, error) {
   296		onceStartServer.Do(startServer)
   297		if err := syscall.SetNonblock(fd, true); err != nil {
   298			return nil, err
   299		}
   300		netfd := &netFD{
   301			sysfd:  fd,
   302			family: family,
   303			sotype: sotype,
   304			net:    net,
   305		}
   306		netfd.cr = make(chan error, 1)
   307		netfd.cw = make(chan error, 1)
   308		return netfd, nil
   309	}
   310	
   311	func (fd *netFD) setAddr(laddr, raddr Addr) {
   312		fd.laddr = laddr
   313		fd.raddr = raddr
   314		var ls, rs string
   315		if laddr != nil {
   316			ls = laddr.String()
   317		}
   318		if raddr != nil {
   319			rs = raddr.String()
   320		}
   321		fd.sysfile = os.NewFile(uintptr(fd.sysfd), fd.net+":"+ls+"->"+rs)
   322	}
   323	
   324	func (fd *netFD) connect(ra syscall.Sockaddr) error {
   325		err := syscall.Connect(fd.sysfd, ra)
   326		if err == syscall.EINPROGRESS {
   327			if err = pollserver.WaitWrite(fd); err != nil {
   328				return err
   329			}
   330			var e int
   331			e, err = syscall.GetsockoptInt(fd.sysfd, syscall.SOL_SOCKET, syscall.SO_ERROR)
   332			if err != nil {
   333				return os.NewSyscallError("getsockopt", err)
   334			}
   335			if e != 0 {
   336				err = syscall.Errno(e)
   337			}
   338		}
   339		return err
   340	}
   341	
   342	var errClosing = errors.New("use of closed network connection")
   343	
   344	// Add a reference to this fd.
   345	// If closing==true, pollserver must be locked; mark the fd as closing.
   346	// Returns an error if the fd cannot be used.
   347	func (fd *netFD) incref(closing bool) error {
   348		if fd == nil {
   349			return errClosing
   350		}
   351		fd.sysmu.Lock()
   352		if fd.closing {
   353			fd.sysmu.Unlock()
   354			return errClosing
   355		}
   356		fd.sysref++
   357		if closing {
   358			fd.closing = true
   359		}
   360		fd.sysmu.Unlock()
   361		return nil
   362	}
   363	
   364	// Remove a reference to this FD and close if we've been asked to do so (and
   365	// there are no references left.
   366	func (fd *netFD) decref() {
   367		if fd == nil {
   368			return
   369		}
   370		fd.sysmu.Lock()
   371		fd.sysref--
   372		if fd.closing && fd.sysref == 0 && fd.sysfile != nil {
   373			fd.sysfile.Close()
   374			fd.sysfile = nil
   375			fd.sysfd = -1
   376		}
   377		fd.sysmu.Unlock()
   378	}
   379	
   380	func (fd *netFD) Close() error {
   381		pollserver.Lock() // needed for both fd.incref(true) and pollserver.Evict
   382		defer pollserver.Unlock()
   383		if err := fd.incref(true); err != nil {
   384			return err
   385		}
   386		// Unblock any I/O.  Once it all unblocks and returns,
   387		// so that it cannot be referring to fd.sysfd anymore,
   388		// the final decref will close fd.sysfd.  This should happen
   389		// fairly quickly, since all the I/O is non-blocking, and any
   390		// attempts to block in the pollserver will return errClosing.
   391		pollserver.Evict(fd)
   392		fd.decref()
   393		return nil
   394	}
   395	
   396	func (fd *netFD) shutdown(how int) error {
   397		if err := fd.incref(false); err != nil {
   398			return err
   399		}
   400		defer fd.decref()
   401		err := syscall.Shutdown(fd.sysfd, how)
   402		if err != nil {
   403			return &OpError{"shutdown", fd.net, fd.laddr, err}
   404		}
   405		return nil
   406	}
   407	
   408	func (fd *netFD) CloseRead() error {
   409		return fd.shutdown(syscall.SHUT_RD)
   410	}
   411	
   412	func (fd *netFD) CloseWrite() error {
   413		return fd.shutdown(syscall.SHUT_WR)
   414	}
   415	
   416	func (fd *netFD) Read(p []byte) (n int, err error) {
   417		fd.rio.Lock()
   418		defer fd.rio.Unlock()
   419		if err := fd.incref(false); err != nil {
   420			return 0, err
   421		}
   422		defer fd.decref()
   423		for {
   424			n, err = syscall.Read(int(fd.sysfd), p)
   425			if err == syscall.EAGAIN {
   426				err = errTimeout
   427				if fd.rdeadline >= 0 {
   428					if err = pollserver.WaitRead(fd); err == nil {
   429						continue
   430					}
   431				}
   432			}
   433			if err != nil {
   434				n = 0
   435			} else if n == 0 && err == nil && fd.sotype != syscall.SOCK_DGRAM {
   436				err = io.EOF
   437			}
   438			break
   439		}
   440		if err != nil && err != io.EOF {
   441			err = &OpError{"read", fd.net, fd.raddr, err}
   442		}
   443		return
   444	}
   445	
   446	func (fd *netFD) ReadFrom(p []byte) (n int, sa syscall.Sockaddr, err error) {
   447		fd.rio.Lock()
   448		defer fd.rio.Unlock()
   449		if err := fd.incref(false); err != nil {
   450			return 0, nil, err
   451		}
   452		defer fd.decref()
   453		for {
   454			n, sa, err = syscall.Recvfrom(fd.sysfd, p, 0)
   455			if err == syscall.EAGAIN {
   456				err = errTimeout
   457				if fd.rdeadline >= 0 {
   458					if err = pollserver.WaitRead(fd); err == nil {
   459						continue
   460					}
   461				}
   462			}
   463			if err != nil {
   464				n = 0
   465			}
   466			break
   467		}
   468		if err != nil && err != io.EOF {
   469			err = &OpError{"read", fd.net, fd.laddr, err}
   470		}
   471		return
   472	}
   473	
   474	func (fd *netFD) ReadMsg(p []byte, oob []byte) (n, oobn, flags int, sa syscall.Sockaddr, err error) {
   475		fd.rio.Lock()
   476		defer fd.rio.Unlock()
   477		if err := fd.incref(false); err != nil {
   478			return 0, 0, 0, nil, err
   479		}
   480		defer fd.decref()
   481		for {
   482			n, oobn, flags, sa, err = syscall.Recvmsg(fd.sysfd, p, oob, 0)
   483			if err == syscall.EAGAIN {
   484				err = errTimeout
   485				if fd.rdeadline >= 0 {
   486					if err = pollserver.WaitRead(fd); err == nil {
   487						continue
   488					}
   489				}
   490			}
   491			if err == nil && n == 0 {
   492				err = io.EOF
   493			}
   494			break
   495		}
   496		if err != nil && err != io.EOF {
   497			err = &OpError{"read", fd.net, fd.laddr, err}
   498			return
   499		}
   500		return
   501	}
   502	
   503	func (fd *netFD) Write(p []byte) (int, error) {
   504		fd.wio.Lock()
   505		defer fd.wio.Unlock()
   506		if err := fd.incref(false); err != nil {
   507			return 0, err
   508		}
   509		defer fd.decref()
   510		if fd.sysfile == nil {
   511			return 0, syscall.EINVAL
   512		}
   513	
   514		var err error
   515		nn := 0
   516		for {
   517			var n int
   518			n, err = syscall.Write(int(fd.sysfd), p[nn:])
   519			if n > 0 {
   520				nn += n
   521			}
   522			if nn == len(p) {
   523				break
   524			}
   525			if err == syscall.EAGAIN {
   526				err = errTimeout
   527				if fd.wdeadline >= 0 {
   528					if err = pollserver.WaitWrite(fd); err == nil {
   529						continue
   530					}
   531				}
   532			}
   533			if err != nil {
   534				n = 0
   535				break
   536			}
   537			if n == 0 {
   538				err = io.ErrUnexpectedEOF
   539				break
   540			}
   541		}
   542		if err != nil {
   543			err = &OpError{"write", fd.net, fd.raddr, err}
   544		}
   545		return nn, err
   546	}
   547	
   548	func (fd *netFD) WriteTo(p []byte, sa syscall.Sockaddr) (n int, err error) {
   549		fd.wio.Lock()
   550		defer fd.wio.Unlock()
   551		if err := fd.incref(false); err != nil {
   552			return 0, err
   553		}
   554		defer fd.decref()
   555		for {
   556			err = syscall.Sendto(fd.sysfd, p, 0, sa)
   557			if err == syscall.EAGAIN {
   558				err = errTimeout
   559				if fd.wdeadline >= 0 {
   560					if err = pollserver.WaitWrite(fd); err == nil {
   561						continue
   562					}
   563				}
   564			}
   565			break
   566		}
   567		if err == nil {
   568			n = len(p)
   569		} else {
   570			err = &OpError{"write", fd.net, fd.raddr, err}
   571		}
   572		return
   573	}
   574	
   575	func (fd *netFD) WriteMsg(p []byte, oob []byte, sa syscall.Sockaddr) (n int, oobn int, err error) {
   576		fd.wio.Lock()
   577		defer fd.wio.Unlock()
   578		if err := fd.incref(false); err != nil {
   579			return 0, 0, err
   580		}
   581		defer fd.decref()
   582		for {
   583			err = syscall.Sendmsg(fd.sysfd, p, oob, sa, 0)
   584			if err == syscall.EAGAIN {
   585				err = errTimeout
   586				if fd.wdeadline >= 0 {
   587					if err = pollserver.WaitWrite(fd); err == nil {
   588						continue
   589					}
   590				}
   591			}
   592			break
   593		}
   594		if err == nil {
   595			n = len(p)
   596			oobn = len(oob)
   597		} else {
   598			err = &OpError{"write", fd.net, fd.raddr, err}
   599		}
   600		return
   601	}
   602	
   603	func (fd *netFD) accept(toAddr func(syscall.Sockaddr) Addr) (netfd *netFD, err error) {
   604		if err := fd.incref(false); err != nil {
   605			return nil, err
   606		}
   607		defer fd.decref()
   608	
   609		// See ../syscall/exec.go for description of ForkLock.
   610		// It is okay to hold the lock across syscall.Accept
   611		// because we have put fd.sysfd into non-blocking mode.
   612		var s int
   613		var rsa syscall.Sockaddr
   614		for {
   615			syscall.ForkLock.RLock()
   616			s, rsa, err = syscall.Accept(fd.sysfd)
   617			if err != nil {
   618				syscall.ForkLock.RUnlock()
   619				if err == syscall.EAGAIN {
   620					err = errTimeout
   621					if fd.rdeadline >= 0 {
   622						if err = pollserver.WaitRead(fd); err == nil {
   623							continue
   624						}
   625					}
   626				} else if err == syscall.ECONNABORTED {
   627					// This means that a socket on the listen queue was closed
   628					// before we Accept()ed it; it's a silly error, so try again.
   629					continue
   630				}
   631				return nil, &OpError{"accept", fd.net, fd.laddr, err}
   632			}
   633			break
   634		}
   635		syscall.CloseOnExec(s)
   636		syscall.ForkLock.RUnlock()
   637	
   638		if netfd, err = newFD(s, fd.family, fd.sotype, fd.net); err != nil {
   639			syscall.Close(s)
   640			return nil, err
   641		}
   642		lsa, _ := syscall.Getsockname(netfd.sysfd)
   643		netfd.setAddr(toAddr(lsa), toAddr(rsa))
   644		return netfd, nil
   645	}
   646	
   647	func (fd *netFD) dup() (f *os.File, err error) {
   648		ns, err := syscall.Dup(fd.sysfd)
   649		if err != nil {
   650			return nil, &OpError{"dup", fd.net, fd.laddr, err}
   651		}
   652	
   653		// We want blocking mode for the new fd, hence the double negative.
   654		if err = syscall.SetNonblock(ns, false); err != nil {
   655			return nil, &OpError{"setnonblock", fd.net, fd.laddr, err}
   656		}
   657	
   658		return os.NewFile(uintptr(ns), fd.sysfile.Name()), nil
   659	}
   660	
   661	func closesocket(s int) error {
   662		return syscall.Close(s)
   663	}