io: reimplement Pipe

No known bugs in the current pipe,
but this one is simpler and easier to
understand.

R=iant
CC=golang-dev
https://golang.org/cl/4252057

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.
+	rl    sync.Mutex // gates readers one at a time
-	// Writer does the same on cw1, cw2.
+	wl    sync.Mutex // gates writers one at a time
-	r1, w1 chan []byte
+	l     sync.Mutex // protects remaining fields
-	r2, w2 chan pipeResult
+	data  []byte     // data remaining in pending write
-
+	rwait sync.Cond  // waiting reader
-	rclose chan os.Error // read close; error to return to writers
+	wwait sync.Cond  // waiting writer
-	wclose chan os.Error // write close; error to return to readers
+	rerr  os.Error   // if reader closed, error to give writes
-
+	werr  os.Error   // if writer closed, error to give reads
-	done chan int // read or write half is done
 }
 
-func (p *pipe) run() {
+func (p *pipe) read(b []byte) (n int, err os.Error) {
-	var (
+	// One reader at a time.
-		rb    []byte      // pending Read
+	p.rl.Lock()
-		wb    []byte      // pending Write
+	defer p.rl.Unlock()
-		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
 
+	p.l.Lock()
+	defer p.l.Unlock()
 	for {
-		select {
+		if p.rerr != nil {
-		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 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
-			}
-		}
-	}
-}
-
-// 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.
-
-type pipeHalf struct {
-	c1     chan []byte
-	c2     chan pipeResult
-	cclose chan os.Error
-	done   chan int
-
-	lock   sync.Mutex
-	closed bool
-
-	io       sync.Mutex
-	ioclosed bool
-}
-
-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.io.Unlock()
+		if p.data != nil {
-	p.c1 <- data
+			break
-	res := <-p.c2
-	return res.n, res.err
-}
-
-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
 		}
-	p.closed = true
+		if p.werr != nil {
-	p.lock.Unlock()
+			return 0, p.werr
-
+		}
-	// First, send the close notification.
+		p.rwait.Wait()
-	p.cclose <- err
+	}
-
+	n = copy(b, p.data)
-	// Runner is now responding to rw operations
+	p.data = p.data[n:]
-	// with os.EINVAL.  Cut off future rw operations
+	if len(p.data) == 0 {
-	// by setting ioclosed flag.
+		p.data = nil
-	p.io.Lock()
+		p.wwait.Signal()
-	p.ioclosed = true
+	}
-	p.io.Unlock()
+	return
-
-	// 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
 }
 
-func (p *pipeHalf) finalizer() {
+var zero [0]byte
-	p.close(os.EINVAL)
+
+func (p *pipe) write(b []byte) (n int, err os.Error) {
+	// pipe uses nil to mean not available
+	if b == nil {
+		b = zero[:]
+	}
+
+	// One writer at a time.
+	p.wl.Lock()
+	defer p.wl.Unlock()
+
+	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()
+	}
+	n = len(b) - len(p.data)
+	p.data = nil // in case of rerr or werr
+	return
 }
 
+func (p *pipe) rclose(err os.Error) {
+	if err == nil {
+		err = os.EPIPE
+	}
+	p.l.Lock()
+	defer p.l.Unlock()
+	p.rerr = err
+	p.rwait.Signal()
+	p.wwait.Signal()
+}
+
+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 {
+	r.p.rclose(err)
-		err = os.EPIPE
+	return nil
-	}
-	return r.close(err)
 }
 
 // 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 {
+	w.p.wclose(err)
-		err = os.EOF
+	return nil
-	}
-	return w.close(err)
 }
 
 // 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{
+	p := new(pipe)
-		r1:     make(chan []byte),
+	p.rwait.L = &p.l
-		r2:     make(chan pipeResult),
+	p.wwait.L = &p.l
-		w1:     make(chan []byte),
+	r := &PipeReader{p}
-		w2:     make(chan pipeResult),
+	w := &PipeWriter{p}
-		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)
-
 	return r, w
 }