Go-深入理解IO

type Reader interface {
	Read(p []byte) (n int, err error)
}

type Writer interface {
	Write(p []byte) (n int, err error)
}

type ReaderAt interface {
	ReadAt(p []byte, off int64) (n int, err error)
}

func Copy(dst Writer, src Reader) (written int64, err error) {
	return copyBuffer(dst, src, nil)
}

func copyBuffer(dst Writer, src Reader, buf []byte) (written int64, err error) {
	// If the reader has a WriteTo method, use it to do the copy.
	// Avoids an allocation and a copy.
	if wt, ok := src.(WriterTo); ok {
		return wt.WriteTo(dst)
	}
	// Similarly, if the writer has a ReadFrom method, use it to do the copy.
	if rt, ok := dst.(ReaderFrom); ok {
		return rt.ReadFrom(src)
	}
	if buf == nil {
		size := 32 * 1024
		// 如果src是LimitedReader，则将buf的大小设置为LimitedReader.N
		if l, ok := src.(*LimitedReader); ok && int64(size) > l.N {
			if l.N < 1 {
				size = 1
			} else {
				size = int(l.N)
			}
		}
		buf = make([]byte, size)
	}
	for {
		nr, er := src.Read(buf)
		if nr > 0 {
			nw, ew := dst.Write(buf[0:nr])
			if nw > 0 {
				written += int64(nw)
			}
			if ew != nil {
				err = ew
				break
			}
			// 写入和读取的字节不一致
			if nr != nw {
				err = ErrShortWrite
				break
			}
		}
		if er != nil {
			if er != EOF {
				err = er
			}
			break
		}
	}
	return written, err
}

// LimitReader returns a Reader that reads from r
// but stops with EOF after n bytes.
// The underlying implementation is a *LimitedReader.
func LimitReader(r Reader, n int64) Reader { return &LimitedReader{r, n} }

// A LimitedReader reads from R but limits the amount of
// data returned to just N bytes. Each call to Read
// updates N to reflect the new amount remaining.
// Read returns EOF when N <= 0 or when the underlying R returns EOF.
type LimitedReader struct {
	R Reader // underlying reader
	N int64  // max bytes remaining
}

func (l *LimitedReader) Read(p []byte) (n int, err error) {
	if l.N <= 0 {
		return 0, EOF
	}
	if int64(len(p)) > l.N {
		p = p[0:l.N]
	}
	n, err = l.R.Read(p)
	l.N -= int64(n)
	return
}

func Pipe() (*PipeReader, *PipeWriter) {
	// 都是无缓存的chan
	p := &pipe{
		wrCh: make(chan []byte),
		rdCh: make(chan int),
		done: make(chan struct{}),
	}
	return &PipeReader{p}, &PipeWriter{p}
}

type PipeReader struct {
	p *pipe
}
type PipeWriter struct {
	p *pipe
}

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
}

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

	select {
	case bw := <-p.wrCh:     // 从wrCh中读取数据
		nr := copy(b, bw)    // 将读取的数据copy到b中
		p.rdCh <- nr         // 将读取的字节数量写入到rdCh，这个在write里说下
		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()
	}

	// once主要的作用是防止当len(b) == 0 ，不会进行此判断，这样就无法和Read沟通
	for once := true; once || len(b) > 0; once = false {
		select {
		case p.wrCh <- b:    // 将数据写入到wrCh
			nw := <-p.rdCh   // 从rdCh同步上一次读取的数量（当Read一次没读取完，下次Read需要继续读取）
			b = b[nw:]       // 修改b
			n += nw          // 累加写入的字节数量
		case <-p.done:
			return n, p.writeCloseError()
		}
	}
	return n, nil
}