queueset_poll_linux.go 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100
  1. //go:build linux && !android
  2. package tio
  3. import (
  4. "encoding/binary"
  5. "errors"
  6. "fmt"
  7. "sync/atomic"
  8. "golang.org/x/sys/unix"
  9. )
  10. type pollQueueSet struct {
  11. pq []*Poll
  12. // pqi is exactly the same as pq, but stored as the interface type
  13. pqi []Queue
  14. shutdownFd int
  15. shutdown atomic.Bool
  16. closed atomic.Bool
  17. }
  18. func NewPollQueueSet() (QueueSet, error) {
  19. shutdownFd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
  20. if err != nil {
  21. return nil, fmt.Errorf("failed to create eventfd: %w", err)
  22. }
  23. out := &pollQueueSet{
  24. pq: []*Poll{},
  25. pqi: []Queue{},
  26. shutdownFd: shutdownFd,
  27. }
  28. return out, nil
  29. }
  30. func (c *pollQueueSet) Queues() []Queue {
  31. return c.pqi
  32. }
  33. func (c *pollQueueSet) Add(fd int) error {
  34. if c.closed.Load() {
  35. return errors.New("queue set already closed")
  36. }
  37. x, err := newPoll(fd, c.shutdownFd)
  38. if err != nil {
  39. return err
  40. }
  41. c.pq = append(c.pq, x)
  42. c.pqi = append(c.pqi, x)
  43. return nil
  44. }
  45. func (c *pollQueueSet) wakeForShutdown() error {
  46. var buf [8]byte
  47. binary.NativeEndian.PutUint64(buf[:], 1)
  48. _, err := unix.Write(int(c.shutdownFd), buf[:])
  49. return err
  50. }
  51. // Shutdown wakes every reader without releasing any fds, they see POLLIN on the eventfd and return os.ErrClosed
  52. func (c *pollQueueSet) Shutdown() error {
  53. if c.shutdown.Swap(true) {
  54. return nil
  55. }
  56. for _, x := range c.pq {
  57. x.shutdown.Store(true)
  58. }
  59. return c.wakeForShutdown()
  60. }
  61. func (c *pollQueueSet) Close() error {
  62. if c.closed.Swap(true) {
  63. return nil
  64. }
  65. errs := []error{}
  66. if err := c.Shutdown(); err != nil {
  67. errs = append(errs, err)
  68. }
  69. // Close the per-queue tun fds; this also unblocks any in-flight reads.
  70. for _, x := range c.pq {
  71. if err := x.Close(); err != nil {
  72. errs = append(errs, err)
  73. }
  74. }
  75. // Close the shutdown eventfd last: every reader's pollfd set references it,
  76. // so it must outlive the wake + per-queue teardown above.
  77. if err := unix.Close(c.shutdownFd); err != nil {
  78. errs = append(errs, err)
  79. }
  80. c.shutdownFd = -1
  81. return errors.Join(errs...)
  82. }