fdconn.go raw

   1  package io
   2  
   3  import (
   4  	"errors"
   5  	"io"
   6  	"net"
   7  	"sync"
   8  	"sync/atomic"
   9  	"time"
  10  
  11  	"git.smesh.lol/torly/pkg/unix"
  12  )
  13  
  14  // fdConn is a non-blocking socket connection owned by one reader and one
  15  // writer domain. The reader and writer only share the fd capability; each uses
  16  // its own direction and its own deadline field.
  17  type fdConn struct {
  18  	fd     int
  19  	remote fdAddr
  20  
  21  	readDeadline  time.Time
  22  	writeDeadline time.Time
  23  
  24  	closeOnce sync.Once
  25  	closed    atomic.Bool
  26  }
  27  
  28  type fdAddr string
  29  
  30  func (a fdAddr) Network() string { return "tcp" }
  31  func (a fdAddr) String() string  { return string(a) }
  32  
  33  type timeoutError struct{}
  34  
  35  func (timeoutError) Error() string   { return "i/o timeout" }
  36  func (timeoutError) Timeout() bool   { return true }
  37  func (timeoutError) Temporary() bool { return true }
  38  
  39  var errFDClosed = errors.New("io: file already closed")
  40  
  41  func newFDConn(fd int, remote string) *fdConn {
  42  	return &fdConn{fd: fd, remote: fdAddr(remote)}
  43  }
  44  
  45  func (c *fdConn) Read(p []byte) (int, error) {
  46  	for {
  47  		if c.isClosed() {
  48  			return 0, errFDClosed
  49  		}
  50  		n, err := unix.Read(c.fd, p)
  51  		if err == nil {
  52  			if n == 0 {
  53  				return 0, io.EOF
  54  			}
  55  			return n, nil
  56  		}
  57  		if err == unix.EINTR {
  58  			continue
  59  		}
  60  		if err == unix.EAGAIN {
  61  			if werr := waitFD(c.fd, unix.POLLIN, c.readDeadline); werr != nil {
  62  				return 0, werr
  63  			}
  64  			continue
  65  		}
  66  		return 0, err
  67  	}
  68  }
  69  
  70  func (c *fdConn) Write(p []byte) (int, error) {
  71  	written := 0
  72  	for written < len(p) {
  73  		if c.isClosed() {
  74  			return written, errFDClosed
  75  		}
  76  		n, err := unix.Write(c.fd, p[written:])
  77  		if err == nil {
  78  			written += n
  79  			continue
  80  		}
  81  		if err == unix.EINTR {
  82  			continue
  83  		}
  84  		if err == unix.EAGAIN {
  85  			if werr := waitFD(c.fd, unix.POLLOUT, c.writeDeadline); werr != nil {
  86  				return written, werr
  87  			}
  88  			continue
  89  		}
  90  		return written, err
  91  	}
  92  	return written, nil
  93  }
  94  
  95  func (c *fdConn) Close() error {
  96  	var err error
  97  	c.closeOnce.Do(func() {
  98  		c.closed.Store(true)
  99  		unix.Shutdown(c.fd, unix.SHUT_RDWR)
 100  		err = unix.Close(c.fd)
 101  	})
 102  	return err
 103  }
 104  
 105  func (c *fdConn) LocalAddr() net.Addr  { return fdAddr("") }
 106  func (c *fdConn) RemoteAddr() net.Addr { return c.remote }
 107  
 108  func (c *fdConn) SetDeadline(t time.Time) error {
 109  	c.readDeadline = t
 110  	c.writeDeadline = t
 111  	return nil
 112  }
 113  func (c *fdConn) SetReadDeadline(t time.Time) error  { c.readDeadline = t; return nil }
 114  func (c *fdConn) SetWriteDeadline(t time.Time) error { c.writeDeadline = t; return nil }
 115  
 116  func (c *fdConn) isClosed() bool { return c.closed.Load() }
 117  
 118  // waitFD waits for an event on fd, waking at least every 200 ms so a shut-down
 119  // breaker in another domain is noticed even without a socket event.
 120  func waitFD(fd int, events int16, deadline time.Time) error {
 121  	for {
 122  		timeout := 200
 123  		if !deadline.IsZero() {
 124  			left := time.Until(deadline)
 125  			if left <= 0 {
 126  				return timeoutError{}
 127  			}
 128  			ms := int(left / time.Millisecond)
 129  			if ms < timeout {
 130  				timeout = ms
 131  			}
 132  		}
 133  		if timeout < 0 {
 134  			timeout = 0
 135  		}
 136  		pfd := []unix.PollFd{{Fd: int32(fd), Events: events}}
 137  		n, err := unix.Poll(pfd, timeout)
 138  		if err != nil {
 139  			if err == unix.EINTR {
 140  				continue
 141  			}
 142  			return err
 143  		}
 144  		if n > 0 {
 145  			return nil
 146  		}
 147  		if !deadline.IsZero() && time.Now().After(deadline) {
 148  			return timeoutError{}
 149  		}
 150  	}
 151  }
 152