diff --git a/transport/ray/direct.go b/transport/ray/direct.go index 18ba161d..30f5b70c 100644 --- a/transport/ray/direct.go +++ b/transport/ray/direct.go @@ -57,11 +57,13 @@ func (this *Stream) Read() (*alloc.Buffer, error) { return nil, io.EOF } this.access.RLock() - defer this.access.RUnlock() if this.buffer == nil { + this.access.RUnlock() return nil, io.EOF } - result, open := <-this.buffer + channel := this.buffer + this.access.RUnlock() + result, open := <-channel if !open { return nil, io.EOF } @@ -72,17 +74,11 @@ func (this *Stream) Write(data *alloc.Buffer) error { if this.closed { return io.EOF } - if this.buffer == nil { - return io.EOF - } this.access.RLock() defer this.access.RUnlock() if this.closed { return io.EOF } - if this.buffer == nil { - return io.EOF - } this.buffer <- data return nil } @@ -91,8 +87,8 @@ func (this *Stream) Close() { if this.closed { return } - this.access.RLock() - defer this.access.RUnlock() + this.access.Lock() + defer this.access.Unlock() if this.closed { return }