pipe.go

Documentation: golang.org/x/net/http2

     1  // Copyright 2014 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  package http2
     6  
     7  import (
     8  	"errors"
     9  	"io"
    10  	"sync"
    11  )
    12  
    13  // pipe is a goroutine-safe io.Reader/io.Writer pair. It's like
    14  // io.Pipe except there are no PipeReader/PipeWriter halves, and the
    15  // underlying buffer is an interface. (io.Pipe is always unbuffered)
    16  type pipe struct {
    17  	mu       sync.Mutex
    18  	c        sync.Cond     // c.L lazily initialized to &p.mu
    19  	b        pipeBuffer    // nil when done reading
    20  	unread   int           // bytes unread when done
    21  	err      error         // read error once empty. non-nil means closed.
    22  	breakErr error         // immediate read error (caller doesn't see rest of b)
    23  	donec    chan struct{} // closed on error
    24  	readFn   func()        // optional code to run in Read before error
    25  }
    26  
    27  type pipeBuffer interface {
    28  	Len() int
    29  	io.Writer
    30  	io.Reader
    31  }
    32  
    33  // setBuffer initializes the pipe buffer.
    34  // It has no effect if the pipe is already closed.
    35  func (p *pipe) setBuffer(b pipeBuffer) {
    36  	p.mu.Lock()
    37  	defer p.mu.Unlock()
    38  	if p.err != nil || p.breakErr != nil {
    39  		return
    40  	}
    41  	p.b = b
    42  }
    43  
    44  func (p *pipe) Len() int {
    45  	p.mu.Lock()
    46  	defer p.mu.Unlock()
    47  	if p.b == nil {
    48  		return p.unread
    49  	}
    50  	return p.b.Len()
    51  }
    52  
    53  // Read waits until data is available and copies bytes
    54  // from the buffer into p.
    55  func (p *pipe) Read(d []byte) (n int, err error) {
    56  	p.mu.Lock()
    57  	defer p.mu.Unlock()
    58  	if p.c.L == nil {
    59  		p.c.L = &p.mu
    60  	}
    61  	for {
    62  		if p.breakErr != nil {
    63  			return 0, p.breakErr
    64  		}
    65  		if p.b != nil && p.b.Len() > 0 {
    66  			return p.b.Read(d)
    67  		}
    68  		if p.err != nil {
    69  			if p.readFn != nil {
    70  				p.readFn()     // e.g. copy trailers
    71  				p.readFn = nil // not sticky like p.err
    72  			}
    73  			p.b = nil
    74  			return 0, p.err
    75  		}
    76  		p.c.Wait()
    77  	}
    78  }
    79  
    80  var (
    81  	errClosedPipeWrite        = errors.New("write on closed buffer")
    82  	errUninitializedPipeWrite = errors.New("write on uninitialized buffer")
    83  )
    84  
    85  // Write copies bytes from p into the buffer and wakes a reader.
    86  // It is an error to write more data than the buffer can hold.
    87  func (p *pipe) Write(d []byte) (n int, err error) {
    88  	p.mu.Lock()
    89  	defer p.mu.Unlock()
    90  	if p.c.L == nil {
    91  		p.c.L = &p.mu
    92  	}
    93  	defer p.c.Signal()
    94  	if p.err != nil || p.breakErr != nil {
    95  		return 0, errClosedPipeWrite
    96  	}
    97  	// pipe.setBuffer is never invoked, leaving the buffer uninitialized.
    98  	// We shouldn't try to write to an uninitialized pipe,
    99  	// but returning an error is better than panicking.
   100  	if p.b == nil {
   101  		return 0, errUninitializedPipeWrite
   102  	}
   103  	return p.b.Write(d)
   104  }
   105  
   106  // CloseWithError causes the next Read (waking up a current blocked
   107  // Read if needed) to return the provided err after all data has been
   108  // read.
   109  //
   110  // The error must be non-nil.
   111  func (p *pipe) CloseWithError(err error) { p.closeWithError(&p.err, err, nil) }
   112  
   113  // BreakWithError causes the next Read (waking up a current blocked
   114  // Read if needed) to return the provided err immediately, without
   115  // waiting for unread data.
   116  func (p *pipe) BreakWithError(err error) { p.closeWithError(&p.breakErr, err, nil) }
   117  
   118  // closeWithErrorAndCode is like CloseWithError but also sets some code to run
   119  // in the caller's goroutine before returning the error.
   120  func (p *pipe) closeWithErrorAndCode(err error, fn func()) { p.closeWithError(&p.err, err, fn) }
   121  
   122  func (p *pipe) closeWithError(dst *error, err error, fn func()) {
   123  	if err == nil {
   124  		panic("err must be non-nil")
   125  	}
   126  	p.mu.Lock()
   127  	defer p.mu.Unlock()
   128  	if p.c.L == nil {
   129  		p.c.L = &p.mu
   130  	}
   131  	defer p.c.Signal()
   132  	if *dst != nil {
   133  		// Already been done.
   134  		return
   135  	}
   136  	p.readFn = fn
   137  	if dst == &p.breakErr {
   138  		if p.b != nil {
   139  			p.unread += p.b.Len()
   140  		}
   141  		p.b = nil
   142  	}
   143  	*dst = err
   144  	p.closeDoneLocked()
   145  }
   146  
   147  // requires p.mu be held.
   148  func (p *pipe) closeDoneLocked() {
   149  	if p.donec == nil {
   150  		return
   151  	}
   152  	// Close if unclosed. This isn't racy since we always
   153  	// hold p.mu while closing.
   154  	select {
   155  	case <-p.donec:
   156  	default:
   157  		close(p.donec)
   158  	}
   159  }
   160  
   161  // Err returns the error (if any) first set by BreakWithError or CloseWithError.
   162  func (p *pipe) Err() error {
   163  	p.mu.Lock()
   164  	defer p.mu.Unlock()
   165  	if p.breakErr != nil {
   166  		return p.breakErr
   167  	}
   168  	return p.err
   169  }
   170  
   171  // Done returns a channel which is closed if and when this pipe is closed
   172  // with CloseWithError.
   173  func (p *pipe) Done() <-chan struct{} {
   174  	p.mu.Lock()
   175  	defer p.mu.Unlock()
   176  	if p.donec == nil {
   177  		p.donec = make(chan struct{})
   178  		if p.err != nil || p.breakErr != nil {
   179  			// Already hit an error.
   180  			p.closeDoneLocked()
   181  		}
   182  	}
   183  	return p.donec
   184  }
   185
View as plain text