poll_linux.go

//go:build linux
// +build linux

package uv

import (
	"errors"
	"fmt"
	"io"
	"os"
	"sync"
	"time"

	"golang.org/x/sys/unix"
)

// newPollReader creates a new pollReader for the given io.Reader.
func newPollReader(reader io.Reader) (pollReader, error) {
	file, ok := reader.(File)
	if !ok {
		return newFallbackReader(reader)
	}

	epoll, err := unix.EpollCreate1(0)
	if err != nil {
		return nil, fmt.Errorf("create epoll: %w", err)
	}

	r := &epollReader{
		reader: reader,
		file:   file,
		epoll:  epoll,
	}

	r.cancelSignalReader, r.cancelSignalWriter, err = os.Pipe()
	if err != nil {
		_ = unix.Close(epoll)
		return nil, err
	}

	err = unix.EpollCtl(epoll, unix.EPOLL_CTL_ADD, int(file.Fd()), &unix.EpollEvent{
		Events: unix.EPOLLIN,
		Fd:     int32(file.Fd()),
	})
	if err != nil {
		_ = unix.Close(epoll)
		_ = r.cancelSignalReader.Close()
		_ = r.cancelSignalWriter.Close()
		return nil, fmt.Errorf("add reader to epoll interest list: %w", err)
	}

	err = unix.EpollCtl(epoll, unix.EPOLL_CTL_ADD, int(r.cancelSignalReader.Fd()), &unix.EpollEvent{
		Events: unix.EPOLLIN,
		Fd:     int32(r.cancelSignalReader.Fd()),
	})
	if err != nil {
		_ = unix.Close(epoll)
		_ = r.cancelSignalReader.Close()
		_ = r.cancelSignalWriter.Close()
		return nil, fmt.Errorf("add cancel signal to epoll interest list: %w", err)
	}

	return r, nil
}

// epollReader implements pollReader using the Linux epoll API.
type epollReader struct {
	reader             io.Reader
	file               File
	cancelSignalReader *os.File
	cancelSignalWriter *os.File
	epoll              int
	mu                 sync.Mutex
	canceled           bool
}

// Read reads data from the underlying reader.
func (r *epollReader) Read(p []byte) (int, error) {
	r.mu.Lock()
	if r.canceled {
		r.mu.Unlock()
		return 0, ErrCanceled
	}
	r.mu.Unlock()

	return r.reader.Read(p)
}

// Poll waits for data to be available to read with the given timeout.
func (r *epollReader) Poll(timeout time.Duration) (bool, error) {
	r.mu.Lock()
	if r.canceled {
		r.mu.Unlock()
		return false, ErrCanceled
	}
	r.mu.Unlock()

	events := make([]unix.EpollEvent, 1)

	timeoutMs := -1
	if timeout >= 0 {
		timeoutMs = int(timeout.Milliseconds())
	}

	for {
		n, err := unix.EpollWait(r.epoll, events, timeoutMs)
		if errors.Is(err, unix.EINTR) {
			continue // try again if the syscall was interrupted
		}

		if err != nil {
			return false, fmt.Errorf("epoll wait: %w", err)
		}

		if n == 0 {
			return false, nil // timeout
		}

		break
	}

	switch events[0].Fd {
	case int32(r.file.Fd()):
		return true, nil
	case int32(r.cancelSignalReader.Fd()):
		// remove signal from pipe
		var b [1]byte
		_, readErr := r.cancelSignalReader.Read(b[:])
		if readErr != nil {
			return false, fmt.Errorf("reading cancel signal: %w", readErr)
		}
		return false, ErrCanceled
	}

	return false, fmt.Errorf("unknown error")
}

// Cancel cancels any ongoing poll or read operations.
func (r *epollReader) Cancel() bool {
	r.mu.Lock()
	r.canceled = true
	r.mu.Unlock()

	// send cancel signal
	_, err := r.cancelSignalWriter.Write([]byte{'c'})
	return err == nil
}

// Close closes the reader and releases any resources.
func (r *epollReader) Close() error {
	var errMsgs []error

	// close epoll
	err := unix.Close(r.epoll)
	if err != nil {
		errMsgs = append(errMsgs, fmt.Errorf("closing epoll: %w", err))
	}

	// close pipe
	err = r.cancelSignalWriter.Close()
	if err != nil {
		errMsgs = append(errMsgs, fmt.Errorf("closing cancel signal writer: %w", err))
	}

	err = r.cancelSignalReader.Close()
	if err != nil {
		errMsgs = append(errMsgs, fmt.Errorf("closing cancel signal reader: %w", err))
	}

	if len(errMsgs) > 0 {
		return errors.Join(errMsgs...)
	}

	return nil
}