buffer.go raw

   1  // Copyright 2012 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 ssh
   6  
   7  import (
   8  	"io"
   9  	"sync"
  10  )
  11  
  12  // buffer provides a linked list buffer for data exchange
  13  // between producer and consumer. Theoretically the buffer is
  14  // of unlimited capacity as it does no allocation of its own.
  15  type buffer struct {
  16  	// protects concurrent access to head, tail and closed
  17  	*sync.Cond
  18  
  19  	head *element // the buffer that will be read first
  20  	tail *element // the buffer that will be read last
  21  
  22  	closed bool
  23  }
  24  
  25  // An element represents a single link in a linked list.
  26  type element struct {
  27  	buf  []byte
  28  	next *element
  29  }
  30  
  31  // newBuffer returns an empty buffer that is not closed.
  32  func newBuffer() *buffer {
  33  	e := new(element)
  34  	b := &buffer{
  35  		Cond: newCond(),
  36  		head: e,
  37  		tail: e,
  38  	}
  39  	return b
  40  }
  41  
  42  // write makes buf available for Read to receive.
  43  // buf must not be modified after the call to write.
  44  func (b *buffer) write(buf []byte) {
  45  	b.Cond.L.Lock()
  46  	e := &element{buf: buf}
  47  	b.tail.next = e
  48  	b.tail = e
  49  	b.Cond.Signal()
  50  	b.Cond.L.Unlock()
  51  }
  52  
  53  // eof closes the buffer. Reads from the buffer once all
  54  // the data has been consumed will receive io.EOF.
  55  func (b *buffer) eof() {
  56  	b.Cond.L.Lock()
  57  	b.closed = true
  58  	b.Cond.Signal()
  59  	b.Cond.L.Unlock()
  60  }
  61  
  62  // Read reads data from the internal buffer in buf.  Reads will block
  63  // if no data is available, or until the buffer is closed.
  64  func (b *buffer) Read(buf []byte) (n int, err error) {
  65  	b.Cond.L.Lock()
  66  	defer b.Cond.L.Unlock()
  67  
  68  	for len(buf) > 0 {
  69  		// if there is data in b.head, copy it
  70  		if len(b.head.buf) > 0 {
  71  			r := copy(buf, b.head.buf)
  72  			buf, b.head.buf = buf[r:], b.head.buf[r:]
  73  			n += r
  74  			continue
  75  		}
  76  		// if there is a next buffer, make it the head
  77  		if len(b.head.buf) == 0 && b.head != b.tail {
  78  			b.head = b.head.next
  79  			continue
  80  		}
  81  
  82  		// if at least one byte has been copied, return
  83  		if n > 0 {
  84  			break
  85  		}
  86  
  87  		// if nothing was read, and there is nothing outstanding
  88  		// check to see if the buffer is closed.
  89  		if b.closed {
  90  			err = io.EOF
  91  			break
  92  		}
  93  		// out of buffers, wait for producer
  94  		b.Cond.Wait()
  95  	}
  96  	return
  97  }
  98