poll_select.go

//go:build solaris || darwin || freebsd || netbsd || openbsd || dragonfly
// +build solaris darwin freebsd netbsd openbsd dragonfly

package uv

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

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

// newSelectPollReader creates a new SelectReader for the given io.Reader.
func newSelectPollReader(reader io.Reader) (pollReader, error) {
	file, ok := reader.(File)
	if !ok || file.Fd() >= unix.FD_SETSIZE {
		return newFallbackReader(reader)
	}

	r := &selectReader{
		reader: reader,
		file:   file,
	}

	var err error
	r.cancelSignalReader, r.cancelSignalWriter, err = os.Pipe()
	if err != nil {
		return nil, err
	}

	return r, nil
}

// selectReader implements pollReader using the POSIX select API.
type selectReader struct {
	reader             io.Reader
	file               File
	cancelSignalReader *os.File
	cancelSignalWriter *os.File
	mu                 sync.Mutex
	canceled           bool
}

// Read reads data from the underlying reader.
func (r *selectReader) 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 *selectReader) Poll(timeout time.Duration) (bool, error) {
	r.mu.Lock()
	if r.canceled {
		r.mu.Unlock()
		return false, ErrCanceled
	}
	r.mu.Unlock()

	for {
		readerFd := int(r.file.Fd())
		abortFd := int(r.cancelSignalReader.Fd())

		maxFd := readerFd
		if abortFd > maxFd {
			maxFd = abortFd
		}

		// this is a limitation of the select syscall
		if maxFd >= unix.FD_SETSIZE {
			return false, fmt.Errorf("cannot select on file descriptor %d which is larger than %d", maxFd, unix.FD_SETSIZE)
		}

		fdSet := &unix.FdSet{}
		fdSet.Set(readerFd)
		fdSet.Set(abortFd)

		var tv *unix.Timeval
		if timeout >= 0 {
			t := unix.NsecToTimeval(timeout.Nanoseconds())
			tv = &t
		}

		n, err := unix.Select(maxFd+1, fdSet, nil, nil, tv)
		if errors.Is(err, unix.EINTR) {
			continue // try again if the syscall was interrupted
		}

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

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

		if fdSet.IsSet(abortFd) {
			// 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
		}

		if fdSet.IsSet(readerFd) {
			return true, nil
		}

		return false, fmt.Errorf("select returned without setting a file descriptor")
	}
}

// Cancel cancels any ongoing poll or read operations.
func (r *selectReader) 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 *selectReader) Close() error {
	var errMsgs []string

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

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

	if len(errMsgs) > 0 {
		return fmt.Errorf("%s", strings.Join(errMsgs, ", "))
	}

	return nil
}