io包中通過pipe實現(xiàn)了管道

源碼

// onceError is an object that will only store an error once.
// 加鎖的錯誤信息
// 保證并發(fā)讀寫的安全
type onceError struct {
    sync.Mutex // guards following
    err        error
}

// 加鎖防止并發(fā)讀寫
func (a *onceError) Store(err error) {
    a.Lock()
    defer a.Unlock()
    if a.err != nil {
        return
    }
    a.err = err
}

func (a *onceError) Load() error {
    a.Lock()
    defer a.Unlock()
    return a.err
}

// ErrClosedPipe is the error used for read or write operations on a closed pipe.
var ErrClosedPipe = errors.New("io: read/write on closed pipe")

// A pipe is the shared pipe structure underlying PipeReader and PipeWriter.
// pipe管道結(jié)構，注意是小寫，外部不可見
type pipe struct {
    // 鎖
    wrMu sync.Mutex // Serializes Write operations
    // 通道channel，無緩存
    wrCh chan []byte
    // 記錄最近一次讀取出的字節(jié)數(shù)，也是無緩存channel
    rdCh chan int

    // 保證close(done)只會執(zhí)行一次，多次會panic
    // sync.Once之前源碼解析過
    once sync.Once // Protects closing done
    // 用來標記pipe是否關閉
    done chan struct{}
    // 記錄讀寫錯誤信息
    rerr onceError
    werr onceError
}

// 從pipe讀取數(shù)據(jù)到b
func (p *pipe) Read(b []byte) (n int, err error) {
    // 檢查一次pipe是否關閉
    select {
    case <-p.done:
        return 0, p.readCloseError()
    default:
    }

    // 要么從wrCh讀取出數(shù)據(jù)
    // 要么done被close，否則會阻塞等待
    select {
    // 從channel中讀取數(shù)據(jù)
    case bw := <-p.wrCh:
        // rdCh記錄實際讀取的字節(jié)數(shù)
        // 因為b可能比bw小,所以read并不一定會把bw全部讀出
        nr := copy(b, bw)
        p.rdCh <- nr
        return nr, nil
    // 判斷pipe是否關閉
    case <-p.done:
        return 0, p.readCloseError()
    }
}

// 返回一個讀取已關閉pipe的錯誤
func (p *pipe) readCloseError() error {
    rerr := p.rerr.Load()
    if werr := p.werr.Load(); rerr == nil && werr != nil {
        return werr
    }
    return ErrClosedPipe
}

// 讀端主動關閉pipe
func (p *pipe) CloseRead(err error) error {
    if err == nil {
        err = ErrClosedPipe
    }
    p.rerr.Store(err)
    p.once.Do(func() { close(p.done) })
    return nil
}

// 將b的數(shù)據(jù)寫入pipe
func (p *pipe) Write(b []byte) (n int, err error) {
    // 同樣先檢查pipe是否關閉
    select {
    case <-p.done:
        return 0, p.writeCloseError()
    default:
        // 注意如果pipe未關閉，繼續(xù)執(zhí)行后面之前需要加鎖，至于為什么，往下看
        p.wrMu.Lock()
        defer p.wrMu.Unlock()
    }

    // 不管b是不是空的，至少保證執(zhí)行一次，原因就是解除正在等待的reader的阻塞狀態(tài)
    // 第一次運行之后，后面就判斷b是否已經(jīng)全部通過pipe寫入
    for once := true; once || len(b) > 0; once = false {
        select {
        // 將b寫入到wrCh中
        // 因為wrCh沒有緩存
        // 如果沒有reader在等待讀，就跳過這個case
        // 如果有reader在等待讀，就將p直接傳遞給reader(具體實現(xiàn)可以看之前的channel源碼解析)
        case p.wrCh <- b:
            // 這里rdCh發(fā)揮作用了
            // 到這一步，reader已經(jīng)讀取完了
            // 通過獲取reader實際讀取到的字來判斷p是否被讀取完了
            // 如果沒有讀取完，還會繼續(xù)往pipe中寫，直到下次reader繼續(xù)讀取
            // 這里也能解答為何上面會上鎖，因為p可能分兩次寫pipe，但是對于寫端是黑盒的，寫端認為是一次原子寫入
            nw := <-p.rdCh
            // b有可能沒有讀完
            b = b[nw:]
            n += nw
        // 如果pipe關閉了，就返回實際寫入到字節(jié)數(shù)和錯誤信息
        case <-p.done:
            return n, p.writeCloseError()
        }
    }
    return n, nil
}

func (p *pipe) writeCloseError() error {
    werr := p.werr.Load()
    if rerr := p.rerr.Load(); werr == nil && rerr != nil {
        return rerr
    }
    return ErrClosedPipe
}

// 寫端主動關閉pipe
// 除了錯誤信息不一樣，其他動作跟讀端主動關閉pipe是一致的
func (p *pipe) CloseWrite(err error) error {
    if err == nil {
        err = EOF
    }
    p.werr.Store(err)
    p.once.Do(func() { close(p.done) })
    return nil
}

// A PipeReader is the read half of a pipe.
// 后面分別使用PipeReader和PipeWriter來包裝pipe的讀寫能力
// 即讀端和寫端，讀端只提供讀的能力，寫端只提供寫的能力
type PipeReader struct {
    p *pipe
}

// Read implements the standard Read interface:
// it reads data from the pipe, blocking until a writer
// arrives or the write end is closed.
// If the write end is closed with an error, that error is
// returned as err; otherwise err is EOF.
func (r *PipeReader) Read(data []byte) (n int, err error) {
    return r.p.Read(data)
}

// Close closes the reader; subsequent writes to the
// write half of the pipe will return the error ErrClosedPipe.
func (r *PipeReader) Close() error {
    return r.CloseWithError(nil)
}

// CloseWithError closes the reader; subsequent writes
// to the write half of the pipe will return the error err.
//
// CloseWithError never overwrites the previous error if it exists
// and always returns nil.
func (r *PipeReader) CloseWithError(err error) error {
    return r.p.CloseRead(err)
}

// A PipeWriter is the write half of a pipe.
type PipeWriter struct {
    p *pipe
}

// Write implements the standard Write interface:
// it writes data to the pipe, blocking until one or more readers
// have consumed all the data or the read end is closed.
// If the read end is closed with an error, that err is
// returned as err; otherwise err is ErrClosedPipe.
func (w *PipeWriter) Write(data []byte) (n int, err error) {
    return w.p.Write(data)
}

// Close closes the writer; subsequent reads from the
// read half of the pipe will return no bytes and EOF.
func (w *PipeWriter) Close() error {
    return w.CloseWithError(nil)
}

// CloseWithError closes the writer; subsequent reads from the
// read half of the pipe will return no bytes and the error err,
// or EOF if err is nil.
//
// CloseWithError never overwrites the previous error if it exists
// and always returns nil.
func (w *PipeWriter) CloseWithError(err error) error {
    return w.p.CloseWrite(err)
}

// Pipe creates a synchronous in-memory pipe.
// It can be used to connect code expecting an io.Reader
// with code expecting an io.Writer.
//
// Reads and Writes on the pipe are matched one to one
// except when multiple Reads are needed to consume a single Write.
// That is, each Write to the PipeWriter blocks until it has satisfied
// one or more Reads from the PipeReader that fully consume
// the written data.
// The data is copied directly from the Write to the corresponding
// Read (or Reads); there is no internal buffering.
//
// It is safe to call Read and Write in parallel with each other or with Close.
// Parallel calls to Read and parallel calls to Write are also safe:
// the individual calls will be gated sequentially.
// 構造一個包含讀端和寫端的pipe
// 通過Pipe獲得一個可以立即使用的pipe
func Pipe() (*PipeReader, *PipeWriter) {
    p := &pipe{
        wrCh: make(chan []byte),
        rdCh: make(chan int),
        done: make(chan struct{}),
    }
    return &PipeReader{p}, &PipeWriter{p}
}

總結(jié)

pipe最核心還是通過channel來進行通信，利用無緩沖channel實現(xiàn)了讀端和寫端的阻塞等待和喚醒，同時通過記錄讀取字節(jié)數(shù)和鎖實現(xiàn)了順序流式數(shù)據(jù)傳遞的管道，并對外提供了構建pipe的能力，支持開箱即用