Go語言io pipe原始碼分析詳情

2022-02-09 13:01:23

1.結構分析

對外暴露的是一個建構函式和構造的兩個物件. 兩個物件分別暴露了方法,同時這兩個物件還有一個共同的底層物件. 實際上,這兩個物件暴露的方法是直接呼叫底層物件的, 那麼核心還是在底層物件上,只是通過兩個物件和一個構造方法將底層物件的細節隱藏了.

2.pipe sruct分析

pipe的方法不多,新的寫法卻不少.

 type atomicError struct{ v atomic.Value }

    func (a *atomicError) Store(err error) {
      a.v.Store(struct{ error }{err})
    }
    func (a *atomicError) Load() error {
      err, _ := a.v.Load().(struct{ error })
      return err.error
    }

atomicError提供了error的原子讀寫.

  type pipe struct {
      wrMu sync.Mutex // Serializes Write operations
      wrCh chan []byte
      rdCh chan int

      once sync.Once // Protects closing done
      done chan struct{}
      rerr atomicError
      werr atomicError
    }

可以看到pipe結構體中主要分兩塊:

讀寫通道
- 兩個無緩衝通道
- 一個互斥量(保護暴露的寫函數)
結束標識
- once保證done的關閉只執行一次
- done標誌整個讀寫的結束
- 剩下兩個用於儲存讀寫錯誤
- PipeReader/PipeWriter的分析

3.PipeReader對外暴露的是讀/關閉

    type PipeReader struct {
      p *pipe
    }

    func (r *PipeReader) Read(data []byte) (n int, err error) {
      return r.p.Read(data)
    }

    func (r *PipeReader) Close() error {
      return r.CloseWithError(nil)
    }

    func (r *PipeReader) CloseWithError(err error) error {
      return r.p.CloseRead(err)
    }

PipeWriter對外暴露的是寫/關閉

    type PipeWriter struct {
       p *pipe
     }

    func (w *PipeWriter) Write(data []byte) (n int, err error) {
      return w.p.Write(data)
    }

    func (w *PipeWriter) Close() error {
      return w.CloseWithError(nil)
    }

    func (w *PipeWriter) CloseWithError(err error) error {
      return w.p.CloseWrite(err)
    }

他們的方法集都是指標接收者.具體方法的實現是通過pipe的方法完成的. pipe的方法更加明確:讀/獲取讀錯誤/結束讀寫並設定讀錯誤; 寫/獲取寫錯誤/結束讀寫並設定寫錯誤.思路相當明確.

下面主要分析pipe的讀寫

   func (p *pipe) Read(b []byte) (n int, err error) {
      select {
      case <-p.done:
        return 0, p.readCloseError()
      default:
      }

      select {
      case bw := <-p.wrCh:
        nr := copy(b, bw)
        p.rdCh <- nr
        return nr, nil
      case <-p.done:
        return 0, p.readCloseError()
      }
    }

    func (p *pipe) Write(b []byte) (n int, err error) {
      select {
      case <-p.done:
        return 0, p.writeCloseError()
      default:
        p.wrMu.Lock()
        defer p.wrMu.Unlock()
      }

      for once := true; once || len(b) > 0; once = false {
        select {
        case p.wrCh <- b:
          nw := <-p.rdCh
          b = b[nw:]
          n += nw
        case <-p.done:
          return n, p.writeCloseError()
        }
      }
      return n, nil
    }

讀寫都是利用兩個階段的select來完成,第一個階段的select是判斷讀寫有沒有結束, 第二階段處理實際的讀寫.