|
|
@ -7,7 +7,6 @@ import (
|
|
|
|
"errors"
|
|
|
|
"errors"
|
|
|
|
"io"
|
|
|
|
"io"
|
|
|
|
"sync"
|
|
|
|
"sync"
|
|
|
|
"sync/atomic"
|
|
|
|
|
|
|
|
"time"
|
|
|
|
"time"
|
|
|
|
)
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
@ -20,7 +19,7 @@ type bufferedPipe struct {
|
|
|
|
// only alloc when on first Read or Write
|
|
|
|
// only alloc when on first Read or Write
|
|
|
|
buf *bytes.Buffer
|
|
|
|
buf *bytes.Buffer
|
|
|
|
|
|
|
|
|
|
|
|
closed uint32
|
|
|
|
closed bool
|
|
|
|
rwCond *sync.Cond
|
|
|
|
rwCond *sync.Cond
|
|
|
|
rDeadline time.Time
|
|
|
|
rDeadline time.Time
|
|
|
|
}
|
|
|
|
}
|
|
|
@ -39,7 +38,7 @@ func (p *bufferedPipe) Read(target []byte) (int, error) {
|
|
|
|
p.buf = new(bytes.Buffer)
|
|
|
|
p.buf = new(bytes.Buffer)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
for {
|
|
|
|
for {
|
|
|
|
if atomic.LoadUint32(&p.closed) == 1 && p.buf.Len() == 0 {
|
|
|
|
if p.closed && p.buf.Len() == 0 {
|
|
|
|
return 0, io.EOF
|
|
|
|
return 0, io.EOF
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if !p.rDeadline.IsZero() {
|
|
|
|
if !p.rDeadline.IsZero() {
|
|
|
@ -60,6 +59,36 @@ func (p *bufferedPipe) Read(target []byte) (int, error) {
|
|
|
|
return n, err
|
|
|
|
return n, err
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
func (p *bufferedPipe) WriteTo(w io.Writer) (n int64, err error) {
|
|
|
|
|
|
|
|
p.rwCond.L.Lock()
|
|
|
|
|
|
|
|
defer p.rwCond.L.Unlock()
|
|
|
|
|
|
|
|
if p.buf == nil {
|
|
|
|
|
|
|
|
p.buf = new(bytes.Buffer)
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
for {
|
|
|
|
|
|
|
|
if p.closed && p.buf.Len() == 0 {
|
|
|
|
|
|
|
|
return 0, io.EOF
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
if !p.rDeadline.IsZero() {
|
|
|
|
|
|
|
|
d := time.Until(p.rDeadline)
|
|
|
|
|
|
|
|
if d <= 0 {
|
|
|
|
|
|
|
|
return 0, ErrTimeout
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
time.AfterFunc(d, p.rwCond.Broadcast)
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
if p.buf.Len() > 0 {
|
|
|
|
|
|
|
|
written, er := p.buf.WriteTo(w)
|
|
|
|
|
|
|
|
n += written
|
|
|
|
|
|
|
|
if er != nil {
|
|
|
|
|
|
|
|
p.rwCond.Broadcast()
|
|
|
|
|
|
|
|
return n, er
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
p.rwCond.Broadcast()
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
p.rwCond.Wait()
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
func (p *bufferedPipe) Write(input []byte) (int, error) {
|
|
|
|
func (p *bufferedPipe) Write(input []byte) (int, error) {
|
|
|
|
p.rwCond.L.Lock()
|
|
|
|
p.rwCond.L.Lock()
|
|
|
|
defer p.rwCond.L.Unlock()
|
|
|
|
defer p.rwCond.L.Unlock()
|
|
|
@ -67,7 +96,7 @@ func (p *bufferedPipe) Write(input []byte) (int, error) {
|
|
|
|
p.buf = new(bytes.Buffer)
|
|
|
|
p.buf = new(bytes.Buffer)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
for {
|
|
|
|
for {
|
|
|
|
if atomic.LoadUint32(&p.closed) == 1 {
|
|
|
|
if p.closed {
|
|
|
|
return 0, io.ErrClosedPipe
|
|
|
|
return 0, io.ErrClosedPipe
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if p.buf.Len() <= BUF_SIZE_LIMIT {
|
|
|
|
if p.buf.Len() <= BUF_SIZE_LIMIT {
|
|
|
@ -83,7 +112,10 @@ func (p *bufferedPipe) Write(input []byte) (int, error) {
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
func (p *bufferedPipe) Close() error {
|
|
|
|
func (p *bufferedPipe) Close() error {
|
|
|
|
atomic.StoreUint32(&p.closed, 1)
|
|
|
|
p.rwCond.L.Lock()
|
|
|
|
|
|
|
|
defer p.rwCond.L.Unlock()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
p.closed = true
|
|
|
|
p.rwCond.Broadcast()
|
|
|
|
p.rwCond.Broadcast()
|
|
|
|
return nil
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|