12475: Return non-nil error at EOF even if len(p)==0.

Arvados-DCO-1.1-Signed-off-by: Tom Clegg <tclegg@veritasgenetics.com>

diff --git a/sdk/go/asyncbuf/buf.go b/sdk/go/asyncbuf/buf.go
index ebe0301c3452fb197aab02ad30dd716fc88a7d38..05af02f5f5dae3ec643fcd2a36216901b72defda 100644 (file)
--- a/sdk/go/asyncbuf/buf.go
+++ b/sdk/go/asyncbuf/buf.go
@@ -87,17 +87,22 @@ type reader struct {
 func (r *reader) Read(p []byte) (int, error) {
        r.b.cond.L.Lock()
        for {
-               if r.b.data.Len() > r.read || len(p) == 0 {
+               switch {
+               case r.read < r.b.data.Len():
                        buf := r.b.data.Bytes()
                        r.b.cond.L.Unlock()
                        n := copy(p, buf[r.read:])
                        r.read += n
                        return n, nil
-               }
-               if r.b.err != nil {
+               case r.b.err != nil || len(p) == 0:
+                       // r.b.err != nil means we reached EOF.  And
+                       // even if we're not at EOF, there's no need
+                       // to block if len(p)==0.
+                       err := r.b.err
                        r.b.cond.L.Unlock()
-                       return 0, r.b.err
+                       return 0, err
+               default:
+                       r.b.cond.Wait()
                }
-               r.b.cond.Wait()
        }
 }

diff --git a/sdk/go/asyncbuf/buf_test.go b/sdk/go/asyncbuf/buf_test.go
index 198ebaf512fc71a1e42f4b7e0e64a4ae0b4581d3..cc742a8cbe1fe028917addf245ce04198677e4e6 100644 (file)
--- a/sdk/go/asyncbuf/buf_test.go
+++ b/sdk/go/asyncbuf/buf_test.go
@@ -90,6 +90,53 @@ func (s *Suite) TestWriteReadCloseRead(c *check.C) {
        <-done
 }
 
+func (s *Suite) TestReadAtEOF(c *check.C) {
+       buf := make([]byte, 8)
+
+       b := NewBuffer([]byte{1, 2, 3})
+
+       r := b.NewReader()
+       n, err := r.Read(buf)
+       c.Check(n, check.Equals, 3)
+       c.Check(err, check.IsNil)
+
+       // Reading zero bytes at EOF, but before Close(), doesn't
+       // block or error
+       done := make(chan bool)
+       go func() {
+               defer close(done)
+               n, err = r.Read(buf[:0])
+               c.Check(n, check.Equals, 0)
+               c.Check(err, check.IsNil)
+       }()
+       select {
+       case <-done:
+       case <-time.After(time.Second):
+               c.Error("timeout")
+       }
+
+       b.Close()
+
+       // Reading zero bytes after Close() returns EOF
+       n, err = r.Read(buf[:0])
+       c.Check(n, check.Equals, 0)
+       c.Check(err, check.Equals, io.EOF)
+
+       // Reading from start after Close() returns 3 bytes, then EOF
+       r = b.NewReader()
+       n, err = r.Read(buf)
+       c.Check(n, check.Equals, 3)
+       if err != nil {
+               c.Check(err, check.Equals, io.EOF)
+       }
+       n, err = r.Read(buf[:0])
+       c.Check(n, check.Equals, 0)
+       c.Check(err, check.Equals, io.EOF)
+       n, err = r.Read(buf)
+       c.Check(n, check.Equals, 0)
+       c.Check(err, check.Equals, io.EOF)
+}
+
 func (s *Suite) TestCloseWithError(c *check.C) {
        errFake := errors.New("it's not even a real error")