libgo/go/io/pipe.go

   1 // Copyright 2009 The Go Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style
   3 // license that can be found in the LICENSE file.
   4
   5 // Pipe adapter to connect code expecting an io.Reader
   6 // with code expecting an io.Writer.
   7
   8 package io
   9
  10 import (
  11         "errors"
  12         "sync"
  13 )
  14
  15 // onceError is an object that will only store an error once.
  16 type onceError struct {
  17         sync.Mutex // guards following
  18         err        error
  19 }
  20
  21 func (a *onceError) Store(err error) {
  22         a.Lock()
  23         defer a.Unlock()
  24         if a.err != nil {
  25                 return
  26         }
  27         a.err = err
  28 }
  29 func (a *onceError) Load() error {
  30         a.Lock()
  31         defer a.Unlock()
  32         return a.err
  33 }
  34
  35 // ErrClosedPipe is the error used for read or write operations on a closed pipe.
  36 var ErrClosedPipe = errors.New("io: read/write on closed pipe")
  37
  38 // A pipe is the shared pipe structure underlying PipeReader and PipeWriter.
  39 type pipe struct {
  40         wrMu sync.Mutex // Serializes Write operations
  41         wrCh chan []byte
  42         rdCh chan int
  43
  44         once sync.Once // Protects closing done
  45         done chan struct{}
  46         rerr onceError
  47         werr onceError
  48 }
  49
  50 func (p *pipe) Read(b []byte) (n int, err error) {
  51         select {
  52         case <-p.done:
  53                 return 0, p.readCloseError()
  54         default:
  55         }
  56
  57         select {
  58         case bw := <-p.wrCh:
  59                 nr := copy(b, bw)
  60                 p.rdCh <- nr
  61                 return nr, nil
  62         case <-p.done:
  63                 return 0, p.readCloseError()
  64         }
  65 }
  66
  67 func (p *pipe) readCloseError() error {
  68         rerr := p.rerr.Load()
  69         if werr := p.werr.Load(); rerr == nil && werr != nil {
  70                 return werr
  71         }
  72         return ErrClosedPipe
  73 }
  74
  75 func (p *pipe) CloseRead(err error) error {
  76         if err == nil {
  77                 err = ErrClosedPipe
  78         }
  79         p.rerr.Store(err)
  80         p.once.Do(func() { close(p.done) })
  81         return nil
  82 }
  83
  84 func (p *pipe) Write(b []byte) (n int, err error) {
  85         select {
  86         case <-p.done:
  87                 return 0, p.writeCloseError()
  88         default:
  89                 p.wrMu.Lock()
  90                 defer p.wrMu.Unlock()
  91         }
  92
  93         for once := true; once || len(b) > 0; once = false {
  94                 select {
  95                 case p.wrCh <- b:
  96                         nw := <-p.rdCh
  97                         b = b[nw:]
  98                         n += nw
  99                 case <-p.done:
 100                         return n, p.writeCloseError()
 101                 }
 102         }
 103         return n, nil
 104 }
 105
 106 func (p *pipe) writeCloseError() error {
 107         werr := p.werr.Load()
 108         if rerr := p.rerr.Load(); werr == nil && rerr != nil {
 109                 return rerr
 110         }
 111         return ErrClosedPipe
 112 }
 113
 114 func (p *pipe) CloseWrite(err error) error {
 115         if err == nil {
 116                 err = EOF
 117         }
 118         p.werr.Store(err)
 119         p.once.Do(func() { close(p.done) })
 120         return nil
 121 }
 122
 123 // A PipeReader is the read half of a pipe.
 124 type PipeReader struct {
 125         p *pipe
 126 }
 127
 128 // Read implements the standard Read interface:
 129 // it reads data from the pipe, blocking until a writer
 130 // arrives or the write end is closed.
 131 // If the write end is closed with an error, that error is
 132 // returned as err; otherwise err is EOF.
 133 func (r *PipeReader) Read(data []byte) (n int, err error) {
 134         return r.p.Read(data)
 135 }
 136
 137 // Close closes the reader; subsequent writes to the
 138 // write half of the pipe will return the error ErrClosedPipe.
 139 func (r *PipeReader) Close() error {
 140         return r.CloseWithError(nil)
 141 }
 142
 143 // CloseWithError closes the reader; subsequent writes
 144 // to the write half of the pipe will return the error err.
 145 //
 146 // CloseWithError never overwrites the previous error if it exists
 147 // and always returns nil.
 148 func (r *PipeReader) CloseWithError(err error) error {
 149         return r.p.CloseRead(err)
 150 }
 151
 152 // A PipeWriter is the write half of a pipe.
 153 type PipeWriter struct {
 154         p *pipe
 155 }
 156
 157 // Write implements the standard Write interface:
 158 // it writes data to the pipe, blocking until one or more readers
 159 // have consumed all the data or the read end is closed.
 160 // If the read end is closed with an error, that err is
 161 // returned as err; otherwise err is ErrClosedPipe.
 162 func (w *PipeWriter) Write(data []byte) (n int, err error) {
 163         return w.p.Write(data)
 164 }
 165
 166 // Close closes the writer; subsequent reads from the
 167 // read half of the pipe will return no bytes and EOF.
 168 func (w *PipeWriter) Close() error {
 169         return w.CloseWithError(nil)
 170 }
 171
 172 // CloseWithError closes the writer; subsequent reads from the
 173 // read half of the pipe will return no bytes and the error err,
 174 // or EOF if err is nil.
 175 //
 176 // CloseWithError never overwrites the previous error if it exists
 177 // and always returns nil.
 178 func (w *PipeWriter) CloseWithError(err error) error {
 179         return w.p.CloseWrite(err)
 180 }
 181
 182 // Pipe creates a synchronous in-memory pipe.
 183 // It can be used to connect code expecting an io.Reader
 184 // with code expecting an io.Writer.
 185 //
 186 // Reads and Writes on the pipe are matched one to one
 187 // except when multiple Reads are needed to consume a single Write.
 188 // That is, each Write to the PipeWriter blocks until it has satisfied
 189 // one or more Reads from the PipeReader that fully consume
 190 // the written data.
 191 // The data is copied directly from the Write to the corresponding
 192 // Read (or Reads); there is no internal buffering.
 193 //
 194 // It is safe to call Read and Write in parallel with each other or with Close.
 195 // Parallel calls to Read and parallel calls to Write are also safe:
 196 // the individual calls will be gated sequentially.
 197 func Pipe() (*PipeReader, *PipeWriter) {
 198         p := &pipe{
 199                 wrCh: make(chan []byte),
 200                 rdCh: make(chan int),
 201                 done: make(chan struct{}),
 202         }
 203         return &PipeReader{p}, &PipeWriter{p}
 204 }