summaryrefslogtreecommitdiff
path: root/src/pkg/io
diff options
context:
space:
mode:
Diffstat (limited to 'src/pkg/io')
-rw-r--r--src/pkg/io/io.go24
-rw-r--r--src/pkg/io/io_test.go33
-rw-r--r--src/pkg/io/ioutil/ioutil.go44
-rw-r--r--src/pkg/io/ioutil/tempfile.go2
-rw-r--r--src/pkg/io/multi.go6
-rw-r--r--src/pkg/io/pipe.go307
6 files changed, 166 insertions, 250 deletions
diff --git a/src/pkg/io/io.go b/src/pkg/io/io.go
index 3b8791897..d3707eb1d 100644
--- a/src/pkg/io/io.go
+++ b/src/pkg/io/io.go
@@ -136,6 +136,10 @@ type WriterTo interface {
// At the end of the input stream, ReadAt returns 0, os.EOF.
// ReadAt may return a non-zero number of bytes with a non-nil err.
// In particular, a ReadAt that exhausts the input may return n > 0, os.EOF.
+//
+// If ReadAt is reading from an data stream with a seek offset,
+// ReadAt should not affect nor be affected by the underlying
+// seek offset.
type ReaderAt interface {
ReadAt(p []byte, off int64) (n int, err os.Error)
}
@@ -182,16 +186,16 @@ func ReadAtLeast(r Reader, buf []byte, min int) (n int, err os.Error) {
if len(buf) < min {
return 0, ErrShortBuffer
}
- for n < min {
- nn, e := r.Read(buf[n:])
- if nn > 0 {
- n += nn
- }
- if e != nil {
- if e == os.EOF && n > 0 {
- e = ErrUnexpectedEOF
- }
- return n, e
+ for n < min && err == nil {
+ var nn int
+ nn, err = r.Read(buf[n:])
+ n += nn
+ }
+ if err == os.EOF {
+ if n >= min {
+ err = nil
+ } else if n > 0 {
+ err = ErrUnexpectedEOF
}
}
return
diff --git a/src/pkg/io/io_test.go b/src/pkg/io/io_test.go
index 4fcd85e69..bc4f354af 100644
--- a/src/pkg/io/io_test.go
+++ b/src/pkg/io/io_test.go
@@ -118,27 +118,50 @@ func TestCopynEOF(t *testing.T) {
func TestReadAtLeast(t *testing.T) {
var rb bytes.Buffer
+ testReadAtLeast(t, &rb)
+}
+
+// A version of bytes.Buffer that returns n > 0, os.EOF on Read
+// when the input is exhausted.
+type dataAndEOFBuffer struct {
+ bytes.Buffer
+}
+
+func (r *dataAndEOFBuffer) Read(p []byte) (n int, err os.Error) {
+ n, err = r.Buffer.Read(p)
+ if n > 0 && r.Buffer.Len() == 0 && err == nil {
+ err = os.EOF
+ }
+ return
+}
+
+func TestReadAtLeastWithDataAndEOF(t *testing.T) {
+ var rb dataAndEOFBuffer
+ testReadAtLeast(t, &rb)
+}
+
+func testReadAtLeast(t *testing.T, rb ReadWriter) {
rb.Write([]byte("0123"))
buf := make([]byte, 2)
- n, err := ReadAtLeast(&rb, buf, 2)
+ n, err := ReadAtLeast(rb, buf, 2)
if err != nil {
t.Error(err)
}
- n, err = ReadAtLeast(&rb, buf, 4)
+ n, err = ReadAtLeast(rb, buf, 4)
if err != ErrShortBuffer {
t.Errorf("expected ErrShortBuffer got %v", err)
}
if n != 0 {
t.Errorf("expected to have read 0 bytes, got %v", n)
}
- n, err = ReadAtLeast(&rb, buf, 1)
+ n, err = ReadAtLeast(rb, buf, 1)
if err != nil {
t.Error(err)
}
if n != 2 {
t.Errorf("expected to have read 2 bytes, got %v", n)
}
- n, err = ReadAtLeast(&rb, buf, 2)
+ n, err = ReadAtLeast(rb, buf, 2)
if err != os.EOF {
t.Errorf("expected EOF, got %v", err)
}
@@ -146,7 +169,7 @@ func TestReadAtLeast(t *testing.T) {
t.Errorf("expected to have read 0 bytes, got %v", n)
}
rb.Write([]byte("4"))
- n, err = ReadAtLeast(&rb, buf, 2)
+ n, err = ReadAtLeast(rb, buf, 2)
if err != ErrUnexpectedEOF {
t.Errorf("expected ErrUnexpectedEOF, got %v", err)
}
diff --git a/src/pkg/io/ioutil/ioutil.go b/src/pkg/io/ioutil/ioutil.go
index fb3fdcda1..57d797e85 100644
--- a/src/pkg/io/ioutil/ioutil.go
+++ b/src/pkg/io/ioutil/ioutil.go
@@ -13,16 +13,22 @@ import (
"sort"
)
+// readAll reads from r until an error or EOF and returns the data it read
+// from the internal buffer allocated with a specified capacity.
+func readAll(r io.Reader, capacity int64) ([]byte, os.Error) {
+ buf := bytes.NewBuffer(make([]byte, 0, capacity))
+ _, err := buf.ReadFrom(r)
+ return buf.Bytes(), err
+}
+
// ReadAll reads from r until an error or EOF and returns the data it read.
func ReadAll(r io.Reader) ([]byte, os.Error) {
- var buf bytes.Buffer
- _, err := io.Copy(&buf, r)
- return buf.Bytes(), err
+ return readAll(r, bytes.MinRead)
}
// ReadFile reads the file named by filename and returns the contents.
func ReadFile(filename string) ([]byte, os.Error) {
- f, err := os.Open(filename, os.O_RDONLY, 0)
+ f, err := os.Open(filename)
if err != nil {
return nil, err
}
@@ -34,23 +40,19 @@ func ReadFile(filename string) ([]byte, os.Error) {
if err == nil && fi.Size < 2e9 { // Don't preallocate a huge buffer, just in case.
n = fi.Size
}
- // Add a little extra in case Size is zero, and to avoid another allocation after
- // Read has filled the buffer.
- n += bytes.MinRead
- // Pre-allocate the correct size of buffer, then set its size to zero. The
- // Buffer will read into the allocated space cheaply. If the size was wrong,
- // we'll either waste some space off the end or reallocate as needed, but
+ // As initial capacity for readAll, use n + a little extra in case Size is zero,
+ // and to avoid another allocation after Read has filled the buffer. The readAll
+ // call will read into its allocated internal buffer cheaply. If the size was
+ // wrong, we'll either waste some space off the end or reallocate as needed, but
// in the overwhelmingly common case we'll get it just right.
- buf := bytes.NewBuffer(make([]byte, 0, n))
- _, err = buf.ReadFrom(f)
- return buf.Bytes(), err
+ return readAll(f, n+bytes.MinRead)
}
// WriteFile writes data to a file named by filename.
// If the file does not exist, WriteFile creates it with permissions perm;
// otherwise WriteFile truncates it before writing.
func WriteFile(filename string, data []byte, perm uint32) os.Error {
- f, err := os.Open(filename, os.O_WRONLY|os.O_CREAT|os.O_TRUNC, perm)
+ f, err := os.OpenFile(filename, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, perm)
if err != nil {
return err
}
@@ -72,7 +74,7 @@ func (f fileInfoList) Swap(i, j int) { f[i], f[j] = f[j], f[i] }
// ReadDir reads the directory named by dirname and returns
// a list of sorted directory entries.
func ReadDir(dirname string) ([]*os.FileInfo, os.Error) {
- f, err := os.Open(dirname, os.O_RDONLY, 0)
+ f, err := os.Open(dirname)
if err != nil {
return nil, err
}
@@ -88,3 +90,15 @@ func ReadDir(dirname string) ([]*os.FileInfo, os.Error) {
sort.Sort(fi)
return fi, nil
}
+
+type nopCloser struct {
+ io.Reader
+}
+
+func (nopCloser) Close() os.Error { return nil }
+
+// NopCloser returns a ReadCloser with a no-op Close method wrapping
+// the provided Reader r.
+func NopCloser(r io.Reader) io.ReadCloser {
+ return nopCloser{r}
+}
diff --git a/src/pkg/io/ioutil/tempfile.go b/src/pkg/io/ioutil/tempfile.go
index 62f8849c0..8e681bdc3 100644
--- a/src/pkg/io/ioutil/tempfile.go
+++ b/src/pkg/io/ioutil/tempfile.go
@@ -48,7 +48,7 @@ func TempFile(dir, prefix string) (f *os.File, err os.Error) {
nconflict := 0
for i := 0; i < 10000; i++ {
name := filepath.Join(dir, prefix+nextSuffix())
- f, err = os.Open(name, os.O_RDWR|os.O_CREATE|os.O_EXCL, 0600)
+ f, err = os.OpenFile(name, os.O_RDWR|os.O_CREATE|os.O_EXCL, 0600)
if pe, ok := err.(*os.PathError); ok && pe.Error == os.EEXIST {
if nconflict++; nconflict > 10 {
rand = reseed()
diff --git a/src/pkg/io/multi.go b/src/pkg/io/multi.go
index 88e4f1b76..d702d46c7 100644
--- a/src/pkg/io/multi.go
+++ b/src/pkg/io/multi.go
@@ -15,10 +15,8 @@ func (mr *multiReader) Read(p []byte) (n int, err os.Error) {
n, err = mr.readers[0].Read(p)
if n > 0 || err != os.EOF {
if err == os.EOF {
- // This shouldn't happen.
- // Well-behaved Readers should never
- // return non-zero bytes read with an
- // EOF. But if so, we clean it.
+ // Don't return EOF yet. There may be more bytes
+ // in the remaining readers.
err = nil
}
return
diff --git a/src/pkg/io/pipe.go b/src/pkg/io/pipe.go
index df76418b9..00be8efa2 100644
--- a/src/pkg/io/pipe.go
+++ b/src/pkg/io/pipe.go
@@ -9,7 +9,6 @@ package io
import (
"os"
- "runtime"
"sync"
)
@@ -18,208 +17,114 @@ type pipeResult struct {
err os.Error
}
-// Shared pipe structure.
+// A pipe is the shared pipe structure underlying PipeReader and PipeWriter.
type pipe struct {
- // Reader sends on cr1, receives on cr2.
- // Writer does the same on cw1, cw2.
- r1, w1 chan []byte
- r2, w2 chan pipeResult
-
- rclose chan os.Error // read close; error to return to writers
- wclose chan os.Error // write close; error to return to readers
-
- done chan int // read or write half is done
-}
-
-func (p *pipe) run() {
- var (
- rb []byte // pending Read
- wb []byte // pending Write
- wn int // amount written so far from wb
- rerr os.Error // if read end is closed, error to send to writers
- werr os.Error // if write end is closed, error to send to readers
- r1 chan []byte // p.cr1 or nil depending on whether Read is ok
- w1 chan []byte // p.cw1 or nil depending on whether Write is ok
- ndone int
- )
-
- // Read and Write are enabled at the start.
- r1 = p.r1
- w1 = p.w1
-
+ rl sync.Mutex // gates readers one at a time
+ wl sync.Mutex // gates writers one at a time
+ l sync.Mutex // protects remaining fields
+ data []byte // data remaining in pending write
+ rwait sync.Cond // waiting reader
+ wwait sync.Cond // waiting writer
+ rerr os.Error // if reader closed, error to give writes
+ werr os.Error // if writer closed, error to give reads
+}
+
+func (p *pipe) read(b []byte) (n int, err os.Error) {
+ // One reader at a time.
+ p.rl.Lock()
+ defer p.rl.Unlock()
+
+ p.l.Lock()
+ defer p.l.Unlock()
for {
- select {
- case <-p.done:
- if ndone++; ndone == 2 {
- // both reader and writer are gone
- // close out any existing i/o
- if r1 == nil {
- p.r2 <- pipeResult{0, os.EINVAL}
- }
- if w1 == nil {
- p.w2 <- pipeResult{0, os.EINVAL}
- }
- return
- }
- continue
- case rerr = <-p.rclose:
- if w1 == nil {
- // finish pending Write
- p.w2 <- pipeResult{wn, rerr}
- wn = 0
- w1 = p.w1 // allow another Write
- }
- if r1 == nil {
- // Close of read side during Read.
- // finish pending Read with os.EINVAL.
- p.r2 <- pipeResult{0, os.EINVAL}
- r1 = p.r1 // allow another Read
- }
- continue
- case werr = <-p.wclose:
- if r1 == nil {
- // finish pending Read
- p.r2 <- pipeResult{0, werr}
- r1 = p.r1 // allow another Read
- }
- if w1 == nil {
- // Close of write side during Write.
- // finish pending Write with os.EINVAL.
- p.w2 <- pipeResult{wn, os.EINVAL}
- wn = 0
- w1 = p.w1 // allow another Write
- }
- continue
- case rb = <-r1:
- if werr != nil {
- // write end is closed
- p.r2 <- pipeResult{0, werr}
- continue
- }
- if rerr != nil {
- // read end is closed
- p.r2 <- pipeResult{0, os.EINVAL}
- continue
- }
- r1 = nil // disable Read until this one is done
- case wb = <-w1:
- if rerr != nil {
- // read end is closed
- p.w2 <- pipeResult{0, rerr}
- continue
- }
- if werr != nil {
- // write end is closed
- p.w2 <- pipeResult{0, os.EINVAL}
- continue
- }
- w1 = nil // disable Write until this one is done
+ if p.rerr != nil {
+ return 0, os.EINVAL
}
-
- if r1 == nil && w1 == nil {
- // Have rb and wb. Execute.
- n := copy(rb, wb)
- wn += n
- wb = wb[n:]
-
- // Finish Read.
- p.r2 <- pipeResult{n, nil}
- r1 = p.r1 // allow another Read
-
- // Maybe finish Write.
- if len(wb) == 0 {
- p.w2 <- pipeResult{wn, nil}
- wn = 0
- w1 = p.w1 // allow another Write
- }
+ if p.data != nil {
+ break
}
+ if p.werr != nil {
+ return 0, p.werr
+ }
+ p.rwait.Wait()
+ }
+ n = copy(b, p.data)
+ p.data = p.data[n:]
+ if len(p.data) == 0 {
+ p.data = nil
+ p.wwait.Signal()
}
+ return
}
-// Read/write halves of the pipe.
-// They are separate structures for two reasons:
-// 1. If one end becomes garbage without being Closed,
-// its finalizer can Close so that the other end
-// does not hang indefinitely.
-// 2. Clients cannot use interface conversions on the
-// read end to find the Write method, and vice versa.
+var zero [0]byte
-type pipeHalf struct {
- c1 chan []byte
- c2 chan pipeResult
- cclose chan os.Error
- done chan int
-
- lock sync.Mutex
- closed bool
+func (p *pipe) write(b []byte) (n int, err os.Error) {
+ // pipe uses nil to mean not available
+ if b == nil {
+ b = zero[:]
+ }
- io sync.Mutex
- ioclosed bool
-}
+ // One writer at a time.
+ p.wl.Lock()
+ defer p.wl.Unlock()
-func (p *pipeHalf) rw(data []byte) (n int, err os.Error) {
- // Run i/o operation.
- // Check ioclosed flag under lock to make sure we're still allowed to do i/o.
- p.io.Lock()
- if p.ioclosed {
- p.io.Unlock()
- return 0, os.EINVAL
+ p.l.Lock()
+ defer p.l.Unlock()
+ p.data = b
+ p.rwait.Signal()
+ for {
+ if p.data == nil {
+ break
+ }
+ if p.rerr != nil {
+ err = p.rerr
+ break
+ }
+ if p.werr != nil {
+ err = os.EINVAL
+ }
+ p.wwait.Wait()
}
- p.io.Unlock()
- p.c1 <- data
- res := <-p.c2
- return res.n, res.err
+ n = len(b) - len(p.data)
+ p.data = nil // in case of rerr or werr
+ return
}
-func (p *pipeHalf) close(err os.Error) os.Error {
- // Close pipe half.
- // Only first call to close does anything.
- p.lock.Lock()
- if p.closed {
- p.lock.Unlock()
- return os.EINVAL
+func (p *pipe) rclose(err os.Error) {
+ if err == nil {
+ err = os.EPIPE
}
- p.closed = true
- p.lock.Unlock()
-
- // First, send the close notification.
- p.cclose <- err
-
- // Runner is now responding to rw operations
- // with os.EINVAL. Cut off future rw operations
- // by setting ioclosed flag.
- p.io.Lock()
- p.ioclosed = true
- p.io.Unlock()
-
- // With ioclosed set, there will be no more rw operations
- // working on the channels.
- // Tell the runner we won't be bothering it anymore.
- p.done <- 1
-
- // Successfully torn down; can disable finalizer.
- runtime.SetFinalizer(p, nil)
-
- return nil
+ p.l.Lock()
+ defer p.l.Unlock()
+ p.rerr = err
+ p.rwait.Signal()
+ p.wwait.Signal()
}
-func (p *pipeHalf) finalizer() {
- p.close(os.EINVAL)
+func (p *pipe) wclose(err os.Error) {
+ if err == nil {
+ err = os.EOF
+ }
+ p.l.Lock()
+ defer p.l.Unlock()
+ p.werr = err
+ p.rwait.Signal()
+ p.wwait.Signal()
}
-
// A PipeReader is the read half of a pipe.
type PipeReader struct {
- pipeHalf
+ p *pipe
}
// Read implements the standard Read interface:
// it reads data from the pipe, blocking until a writer
// arrives or the write end is closed.
// If the write end is closed with an error, that error is
-// returned as err; otherwise err is nil.
+// returned as err; otherwise err is os.EOF.
func (r *PipeReader) Read(data []byte) (n int, err os.Error) {
- return r.rw(data)
+ return r.p.read(data)
}
// Close closes the reader; subsequent writes to the
@@ -231,15 +136,13 @@ func (r *PipeReader) Close() os.Error {
// CloseWithError closes the reader; subsequent writes
// to the write half of the pipe will return the error err.
func (r *PipeReader) CloseWithError(err os.Error) os.Error {
- if err == nil {
- err = os.EPIPE
- }
- return r.close(err)
+ r.p.rclose(err)
+ return nil
}
// A PipeWriter is the write half of a pipe.
type PipeWriter struct {
- pipeHalf
+ p *pipe
}
// Write implements the standard Write interface:
@@ -248,7 +151,7 @@ type PipeWriter struct {
// If the read end is closed with an error, that err is
// returned as err; otherwise err is os.EPIPE.
func (w *PipeWriter) Write(data []byte) (n int, err os.Error) {
- return w.rw(data)
+ return w.p.write(data)
}
// Close closes the writer; subsequent reads from the
@@ -260,10 +163,8 @@ func (w *PipeWriter) Close() os.Error {
// CloseWithError closes the writer; subsequent reads from the
// read half of the pipe will return no bytes and the error err.
func (w *PipeWriter) CloseWithError(err os.Error) os.Error {
- if err == nil {
- err = os.EOF
- }
- return w.close(err)
+ w.p.wclose(err)
+ return nil
}
// Pipe creates a synchronous in-memory pipe.
@@ -272,34 +173,10 @@ func (w *PipeWriter) CloseWithError(err os.Error) os.Error {
// Reads on one end are matched with writes on the other,
// copying data directly between the two; there is no internal buffering.
func Pipe() (*PipeReader, *PipeWriter) {
- p := &pipe{
- r1: make(chan []byte),
- r2: make(chan pipeResult),
- w1: make(chan []byte),
- w2: make(chan pipeResult),
- rclose: make(chan os.Error),
- wclose: make(chan os.Error),
- done: make(chan int),
- }
- go p.run()
-
- // NOTE: Cannot use composite literal here:
- // pipeHalf{c1: p.cr1, c2: p.cr2, cclose: p.crclose, cdone: p.cdone}
- // because this implicitly copies the pipeHalf, which copies the inner mutex.
-
- r := new(PipeReader)
- r.c1 = p.r1
- r.c2 = p.r2
- r.cclose = p.rclose
- r.done = p.done
- runtime.SetFinalizer(r, (*PipeReader).finalizer)
-
- w := new(PipeWriter)
- w.c1 = p.w1
- w.c2 = p.w2
- w.cclose = p.wclose
- w.done = p.done
- runtime.SetFinalizer(w, (*PipeWriter).finalizer)
-
+ p := new(pipe)
+ p.rwait.L = &p.l
+ p.wwait.L = &p.l
+ r := &PipeReader{p}
+ w := &PipeWriter{p}
return r, w
}
generated by cgit v1.2.3 (git 2.46.0) at 2025-11-02 13:02:33 +0000