diff --git a/utils/io2/ring.go b/utils/io2/ring.go index cb0c1c8..525418b 100644 --- a/utils/io2/ring.go +++ b/utils/io2/ring.go @@ -1,8 +1,10 @@ package io2 import ( + "fmt" "io" "sync" + "time" ) type RingBuffer2 struct { @@ -18,7 +20,7 @@ type RingBuffer2 struct { DownstreamName string } -func RingBuffer(src io.ReadCloser, size int) io.ReadCloser { +func RingBuffer(src io.ReadCloser, size int) *RingBuffer2 { lk := &sync.Mutex{} return &RingBuffer2{ buf: make([]byte, size), @@ -41,9 +43,9 @@ func (r *RingBuffer2) Read(p []byte) (n int, err error) { return 0, r.err } - // startTime := time.Now() + startTime := time.Now() r.waitReading.Wait() - // fmt.Printf("%s wait data for %v\n", r.DownstreamName, time.Since(startTime)) + fmt.Printf("%s wait data for %v\n", r.DownstreamName, time.Since(startTime)) } writePos := r.writePos readPos := r.readPos @@ -79,7 +81,9 @@ func (r *RingBuffer2) reading() { // writePos不能和readPos重合,因为无法区分缓冲区是已经满了,还是完全是空的 // 所以writePos最多能到readPos的前一格 for r.writePos+1 == r.readPos { + startTime := time.Now() r.waitComsuming.Wait() + fmt.Printf("%s wait free space for %v\n", r.UpstreamName, time.Since(startTime)) if r.err != nil { return @@ -103,6 +107,9 @@ func (r *RingBuffer2) reading() { n, err = r.src.Read(r.buf[writePos:readPos]) } + // <-time.After(time.Second) + // fmt.Printf("read %d bytes from %s\n", n, r.UpstreamName) + // 无论成功还是失败,都发送一下信号通知读取端 r.waitReading.L.Lock() r.err = err