使用O_RDONLY调用os.OpenFile时，无写入端致命名管道挂起

问题背景 #

我正在编写一个守护进程，通过Unix命名管道接收临时CLI命令的通知，为此实现了一个简易Go包：启动独立goroutine从管道读取数据，将收到的通知发送至channel。

包的核心代码如下：

import (
    "bufio"
    "fmt"
    "os"
    "strings"
    "sync"
    "syscall"
)

type Writer struct {
    f *os.File
}

func NewWriter(ipc string) (*Writer, error) {
    f, err := os.OpenFile(ipc, os.O_WRONLY, 0600)
    if err != nil {
        return nil, fmt.Errorf("writer: open file: %w", err)
    }
    return &Writer{f: f}, nil
}

func (w *Writer) WriteString(str string) (int, error) {
    return w.f.WriteString(fmt.Sprint(str, "\n"))
}

func (w *Writer) Close() error {
    return w.f.Close()
}

type Reader struct {
    f    *os.File
    rmFn func() error
    quit chan struct{}
    done *sync.WaitGroup
}

func NewReader(ipc string) (*Reader, error) {
    err := syscall.Mkfifo(ipc, 0640)
    if err != nil {
        return nil, fmt.Errorf("reader: create fifo: %w", err)
    }

    f, err := os.OpenFile(ipc, os.O_RDONLY, 0640)
    if err != nil {
        return nil, fmt.Errorf("reader: open fifo: %w", err)
    }
    return &Reader{
        f:    f,
        quit: make(chan struct{}),
        done: &sync.WaitGroup{},
        rmFn: func() error {
            return os.Remove(ipc)
        },
    }, nil
}

func (r *Reader) PollRead() <-chan string {
    reader := bufio.NewReader(r.f)
    out := make(chan string)
    r.done.Add(1)
    go func() {
        defer r.done.Done()
        for {
            line, err := reader.ReadBytes('\n')
            if err != nil {
                fmt.Printf("error reading from named pipe: %v\n", err)
                return
            }

            nline := string(line)
            nline = strings.TrimRight(nline, "\n")
            select {
            case out <- nline:
            case <-r.quit:
                close(out)
                return
            }
        }
    }()

    return out
}

func (r *Reader) Close() error {
    close(r.quit)
    r.done.Wait()
    err := r.f.Close()
    if err != nil {
        return fmt.Errorf("error closing named pipe: %v", err)
    }

    err = r.rmFn()
    if err != nil {
        return fmt.Errorf("error removing named pipe: %v", err)
    }
    return nil
}

异常现象 #

该包可正常处理读写，但存在不符合预期的行为：没有写入端时，读取端无法完成初始化——这与常规认知相反，通常是写入端因无读取端挂起，而此处NewReader会卡在打开管道的步骤，无法返回Reader实例。

问题原因 #

这是Unix命名管道的原生行为导致的：当以O_RDONLY模式打开管道时，open()系统调用会阻塞，直到有进程以写入模式打开该管道。你的NewReader函数在创建管道后立即执行打开操作，未设置非阻塞标志，因此没有写入端时，该调用会一直挂起，导致读取端无法完成初始化。

解决方案 #

方案一：将管道打开逻辑移至读取goroutine（推荐） #

把打开管道的操作放到PollRead的goroutine中，这样NewReader可以立即返回，读取逻辑在后台阻塞等待写入端连接，完全符合“读取端先启动、等待写入”的预期行为。

修改后的代码如下：
首先更新Reader结构体，添加管道路径字段：

type Reader struct {
    ipc  string // 新增：存储管道路径
    f    *os.File
    rmFn func() error
    quit chan struct{}
    done *sync.WaitGroup
}

修改NewReader函数，不再提前打开管道：

func NewReader(ipc string) (*Reader, error) {
    err := syscall.Mkfifo(ipc, 0640)
    if err != nil {
        return nil, fmt.Errorf("reader: create fifo: %w", err)
    }

    return &Reader{
        ipc:  ipc,
        quit: make(chan struct{}),
        done: &sync.WaitGroup{},
        rmFn: func() error {
            return os.Remove(ipc)
        },
    }, nil
}

更新PollRead函数，在goroutine中打开并读取管道：

func (r *Reader) PollRead() <-chan string {
    out := make(chan string)
    r.done.Add(1)
    go func() {
        defer r.done.Done()
        defer close(out)

        // 在goroutine中打开管道，阻塞等待写入端连接
        f, err := os.OpenFile(r.ipc, os.O_RDONLY, 0640)
        if err != nil {
            fmt.Printf("error opening named pipe: %v\n", err)
            return
        }
        r.f = f
        defer func() {
            _ = r.f.Close()
        }()

        reader := bufio.NewReader(r.f)
        for {
            line, err := reader.ReadBytes('\n')
            if err != nil {
                fmt.Printf("error reading from named pipe: %v\n", err)
                return
            }

            nline := strings.TrimRight(string(line), "\n")
            select {
            case out <- nline:
            case <-r.quit:
                return
            }
        }
    }()

    return out
}

最后更新Close函数，处理管道未打开的情况：

func (r *Reader) Close() error {
    close(r.quit)
    r.done.Wait()

    var closeErr error
    if r.f != nil {
        closeErr = r.f.Close()
        if closeErr != nil {
            closeErr = fmt.Errorf("error closing named pipe: %w", closeErr)
        }
    }

    rmErr := r.rmFn()
    if rmErr != nil {
        if closeErr != nil {
            return fmt.Errorf("%w; error removing named pipe: %w", closeErr, rmErr)
        }
        return fmt.Errorf("error removing named pipe: %w", rmErr)
    }
    return closeErr
}

方案二：非阻塞模式打开管道 #

通过添加syscall.O_NONBLOCK标志让管道立即打开，然后在读取逻辑中处理无写入端的情况（轮询重试）。这种方式适合需要NewReader立即返回的场景，但会引入轮询开销。

修改NewReader中的打开逻辑：

f, err := os.OpenFile(ipc, os.O_RDONLY|syscall.O_NONBLOCK, 0640)

更新PollRead的读取逻辑，处理非阻塞错误：

import "time"
import "errors"
import "io"

func (r *Reader) PollRead() <-chan string {
    reader := bufio.NewReader(r.f)
    out := make(chan string)
    r.done.Add(1)
    go func() {
        defer r.done.Done()
        defer close(out)
        for {
            select {
            case <-r.quit:
                return
            default:
                line, err := reader.ReadBytes('\n')
                if err != nil {
                    if errors.Is(err, syscall.EAGAIN) || errors.Is(err, syscall.EWOULDBLOCK) {
                        // 无写入端时短暂休眠后重试
                        time.Sleep(100 * time.Millisecond)
                        continue
                    }
                    if err != io.EOF {
                        fmt.Printf("error reading from named pipe: %v\n", err)
                    }
                    return
                }

                nline := strings.TrimRight(string(line), "\n")
                select {
                case out <- nline:
                case <-r.quit:
                    return
                }
            }
        }
    }()

    return out
}

总结 #

方案一更符合Unix命名管道的原生设计，逻辑简洁且无额外开销，是首选方案；方案二则适合对初始化速度有要求的场景。

内容的提问来源于stack exchange，提问作者acud