container_poll_linux.go 1.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869
  1. package tio
  2. import (
  3. "encoding/binary"
  4. "errors"
  5. "fmt"
  6. "golang.org/x/sys/unix"
  7. )
  8. type pollContainer struct {
  9. pq []*Poll
  10. // pqi is exactly the same as pq, but stored as the interface type
  11. pqi []Queue
  12. shutdownFd int
  13. }
  14. func NewPollContainer() (Container, error) {
  15. shutdownFd, err := unix.Eventfd(0, unix.EFD_NONBLOCK|unix.EFD_CLOEXEC)
  16. if err != nil {
  17. return nil, fmt.Errorf("failed to create eventfd: %w", err)
  18. }
  19. out := &pollContainer{
  20. pq: []*Poll{},
  21. pqi: []Queue{},
  22. shutdownFd: shutdownFd,
  23. }
  24. return out, nil
  25. }
  26. func (c *pollContainer) Queues() []Queue {
  27. return c.pqi
  28. }
  29. func (c *pollContainer) Add(fd int) error {
  30. x, err := newPoll(fd, c.shutdownFd)
  31. if err != nil {
  32. return err
  33. }
  34. c.pq = append(c.pq, x)
  35. c.pqi = append(c.pqi, x)
  36. return nil
  37. }
  38. func (c *pollContainer) wakeForShutdown() error {
  39. var buf [8]byte
  40. binary.NativeEndian.PutUint64(buf[:], 1)
  41. _, err := unix.Write(int(c.shutdownFd), buf[:])
  42. return err
  43. }
  44. func (c *pollContainer) Close() error {
  45. errs := []error{}
  46. if err := c.wakeForShutdown(); err != nil {
  47. errs = append(errs, err)
  48. }
  49. for _, x := range c.pq {
  50. if err := x.Close(); err != nil {
  51. errs = append(errs, err)
  52. }
  53. }
  54. return errors.Join(errs...)
  55. }