2
0
mirror of https://github.com/soheilhy/cmux.git synced 2026-08-31 15:54:31 +08:00

3 Commits

Author SHA1 Message Date
Soheil Hassas Yeganeh
3077b24d47 Fix and optimize travis config
There is no point in running shadow and lints for all version of go.

Also, remove 1.3 and 1.4.
2016-04-24 14:10:41 -04:00
Soheil Hassas Yeganeh
d5924ef0b4 Fix a blocking issue in buffer reader
After sniffing and buffering data, if we try to read from
the socket again, bufio.Reader may block. This breaks HTTP handlers
in go1.5.2+ if one tries on browsers or with curl. Go's HTTP client,
however, is not broken. This issue is also there with TeeReader.

Return immediately with the data in the sniffed buffer.
2016-04-24 12:55:13 -04:00
Tamir Duberstein
59b6f01712 Replace buffer with bufferedReader
bufferedReader is an optimized implementation of io.Reader that behaves
like
```
io.MultiReader(bytes.NewReader(buffer.Bytes()), io.TeeReader(source, buffer))
```
without allocating.

This has a measurable effect on benchmarks:
```
name        old time/op    new time/op    delta
CMuxConn-4    1.09µs ± 4%    0.99µs ±19%   -9.32%  (p=0.000 n=17+19)

name        old alloc/op   new alloc/op   delta
CMuxConn-4      240B ± 0%      260B ± 0%   +8.33%  (p=0.000 n=20+20)

name        old allocs/op  new allocs/op  delta
CMuxConn-4      9.00 ± 0%      5.00 ± 0%  -44.44%  (p=0.000 n=20+20)
```

Note that appropriate test coverage is provided by `TestRead`.
2016-02-28 19:41:37 -05:00
5 changed files with 71 additions and 187 deletions

View File

@@ -1,22 +1,27 @@
language: go language: go
go: go:
- 1.3
- 1.4
- 1.5 - 1.5
- 1.6 - 1.6
- tip
matrix:
allow_failures:
- go: tip
gobuild_args: -race gobuild_args: -race
before_install: before_install:
- go get -u github.com/golang/lint/golint - if [[ $TRAVIS_GO_VERSION == 1.6* ]]; then go get -u github.com/kisielk/errcheck; fi
- if [[ $TRAVIS_GO_VERSION == 1.5* ]]; then go get -u github.com/kisielk/errcheck; fi - if [[ $TRAVIS_GO_VERSION == 1.6* ]]; then go get -u github.com/golang/lint/golint; fi
- go get -u golang.org/x/tools/cmd/vet
before_script: before_script:
- '! gofmt -s -l . | read' - '! gofmt -s -l . | read'
- golint ./...
- echo $TRAVIS_GO_VERSION - echo $TRAVIS_GO_VERSION
- if [[ $TRAVIS_GO_VERSION == 1.5* ]]; then errcheck ./...; fi - if [[ $TRAVIS_GO_VERSION == 1.6* ]]; then golint ./...; fi
- go vet . - if [[ $TRAVIS_GO_VERSION == 1.6* ]]; then errcheck ./...; fi
- go tool vet --shadow . - if [[ $TRAVIS_GO_VERSION == 1.6* ]]; then go tool vet .; fi
- if [[ $TRAVIS_GO_VERSION == 1.6* ]]; then go tool vet --shadow .; fi
script:
- go test -v ./...

View File

@@ -1,57 +1,49 @@
package cmux package cmux
import "io" import (
"bytes"
"io"
)
var _ io.ReadWriter = (*buffer)(nil) // bufferedReader is an optimized implementation of io.Reader that behaves like
// ```
type buffer struct { // io.MultiReader(bytes.NewReader(buffer.Bytes()), io.TeeReader(source, buffer))
read int // ```
data []byte // without allocating.
type bufferedReader struct {
source io.Reader
buffer bytes.Buffer
bufferRead int
bufferSize int
sniffing bool
lastErr error
} }
// From the io.Reader documentation: func (s *bufferedReader) Read(p []byte) (int, error) {
// if s.bufferSize > s.bufferRead {
// When Read encounters an error or end-of-file condition after // If we have already read something from the buffer before, we return the
// successfully reading n > 0 bytes, it returns the number of // same data and the last error if any. We need to immediately return,
// bytes read. It may return the (non-nil) error from the same call // otherwise we may block for ever, if we try to be smart and call
// or return the error (and n == 0) from a subsequent call. // source.Read() seeking a little bit of more data.
// An instance of this general case is that a Reader returning bn := copy(p, s.buffer.Bytes()[s.bufferRead:s.bufferSize])
// a non-zero number of bytes at the end of the input stream may s.bufferRead += bn
// return either err == EOF or err == nil. The next Read should return bn, s.lastErr
// return 0, EOF.
//
// This function implements the latter behaviour, returning the
// (non-nil) error from the same call.
func (b *buffer) Read(p []byte) (int, error) {
var err error
n := copy(p, b.data[b.read:])
b.read += n
if b.read == len(b.data) {
err = io.EOF
} }
return n, err
// If there is nothing more to return in the sniffed buffer, read from the
// source.
sn, sErr := s.source.Read(p)
if sn > 0 && s.sniffing {
s.lastErr = sErr
if wn, wErr := s.buffer.Write(p[:sn]); wErr != nil {
return wn, wErr
}
}
return sn, sErr
} }
func (b *buffer) Len() int { func (s *bufferedReader) reset(snif bool) {
return len(b.data) - b.read s.sniffing = snif
} s.bufferRead = 0
s.bufferSize = s.buffer.Len()
func (b *buffer) resetRead() {
b.read = 0
}
// From the io.Writer documentation:
//
// Write writes len(p) bytes from p to the underlying data stream.
// It returns the number of bytes written from p (0 <= n <= len(p))
// and any error encountered that caused the write to stop early.
// Write must return a non-nil error if it returns n < len(p).
// Write must not modify the slice data, even temporarily.
//
// Implementations must not retain p.
//
// In a previous incarnation, this implementation retained the incoming slice.
func (b *buffer) Write(p []byte) (int, error) {
b.data = append(b.data, p...)
return len(p), nil
} }

View File

@@ -1,113 +0,0 @@
package cmux
import (
"bytes"
"io"
"testing"
)
func TestWriteNoModify(t *testing.T) {
var b buffer
const origWriteByte = 0
const postWriteByte = 1
writeBytes := []byte{origWriteByte}
if _, err := b.Write(writeBytes); err != nil {
t.Fatal(err)
}
writeBytes[0] = postWriteByte
readBytes := make([]byte, 1)
if _, err := b.Read(readBytes); err != io.EOF {
t.Fatal(err)
}
if readBytes[0] != origWriteByte {
t.Fatalf("expected to read %x, but read %x; buffer retained passed-in slice", origWriteByte, postWriteByte)
}
}
const writeString = "deadbeef"
func TestBuffer(t *testing.T) {
writeBytes := []byte(writeString)
const numWrites = 10
var b buffer
for i := 0; i < numWrites; i++ {
n, err := b.Write(writeBytes)
if err != nil && err != io.EOF {
t.Fatal(err)
}
if n != len(writeBytes) {
t.Fatalf("cannot write all the bytes: want=%d got=%d", len(writeBytes), n)
}
}
for j := 0; j < 2; j++ {
readBytes := make([]byte, len(writeBytes))
for i := 0; i < numWrites; i++ {
n, err := b.Read(readBytes)
if i == numWrites-1 {
// The last read should report EOF.
if err != io.EOF {
t.Fatal(err)
}
} else if err != nil {
t.Fatal(err)
}
if n != len(readBytes) {
t.Fatalf("cannot read all the bytes: want=%d got=%d", len(readBytes), n)
}
if !bytes.Equal(writeBytes, readBytes) {
t.Errorf("different bytes read: want=%d got=%d", writeBytes, readBytes)
}
}
n, err := b.Read(readBytes)
if err != io.EOF {
t.Errorf("expected EOF")
}
if n != 0 {
t.Errorf("expected buffer to be empty, but got %d bytes", n)
}
b.resetRead()
}
}
func TestBufferOffset(t *testing.T) {
writeBytes := []byte(writeString)
var b buffer
n, err := b.Write(writeBytes)
if err != nil {
t.Fatal(err)
}
if n != len(writeBytes) {
t.Fatalf("cannot write all the bytes: want=%d got=%d", len(writeBytes), n)
}
const readSize = 2
numReads := len(writeBytes) / readSize
for i := 0; i < numReads; i++ {
readBytes := make([]byte, readSize)
n, err := b.Read(readBytes)
if i == numReads-1 {
// The last read should report EOF.
if err != io.EOF {
t.Fatal(err)
}
} else if err != nil {
t.Fatal(err)
}
if n != readSize {
t.Fatalf("cannot read the bytes: want=%d got=%d", readSize, n)
}
if got := writeBytes[i*readSize : i*readSize+readSize]; !bytes.Equal(got, readBytes) {
t.Fatalf("different bytes read: want=%s got=%s", readBytes, got)
}
}
}

26
cmux.go
View File

@@ -125,9 +125,9 @@ func (m *cMux) serve(c net.Conn, donec <-chan struct{}, wg *sync.WaitGroup) {
muc := newMuxConn(c) muc := newMuxConn(c)
for _, sl := range m.sls { for _, sl := range m.sls {
for _, s := range sl.ss { for _, s := range sl.ss {
matched := s(muc.sniffer()) matched := s(muc.startSniffing())
muc.reset()
if matched { if matched {
muc.doneSniffing()
select { select {
case sl.l.connc <- muc: case sl.l.connc <- muc:
case <-donec: case <-donec:
@@ -177,12 +177,13 @@ func (l muxListener) Accept() (net.Conn, error) {
// MuxConn wraps a net.Conn and provides transparent sniffing of connection data. // MuxConn wraps a net.Conn and provides transparent sniffing of connection data.
type MuxConn struct { type MuxConn struct {
net.Conn net.Conn
buf buffer buf bufferedReader
} }
func newMuxConn(c net.Conn) *MuxConn { func newMuxConn(c net.Conn) *MuxConn {
return &MuxConn{ return &MuxConn{
Conn: c, Conn: c,
buf: bufferedReader{source: c},
} }
} }
@@ -196,22 +197,15 @@ func newMuxConn(c net.Conn) *MuxConn {
// a non-zero number of bytes at the end of the input stream may // a non-zero number of bytes at the end of the input stream may
// return either err == EOF or err == nil. The next Read should // return either err == EOF or err == nil. The next Read should
// return 0, EOF. // return 0, EOF.
//
// This function implements the latter behaviour, returning the
// (non-nil) error from the same call.
func (m *MuxConn) Read(p []byte) (int, error) { func (m *MuxConn) Read(p []byte) (int, error) {
n1, err := m.buf.Read(p) return m.buf.Read(p)
if err != io.EOF {
return n1, err
}
n2, err := m.Conn.Read(p[n1:])
return n1 + n2, err
} }
func (m *MuxConn) sniffer() io.Reader { func (m *MuxConn) startSniffing() io.Reader {
return io.MultiReader(&m.buf, io.TeeReader(m.Conn, &m.buf)) m.buf.reset(true)
return &m.buf
} }
func (m *MuxConn) reset() { func (m *MuxConn) doneSniffing() {
m.buf.resetRead() m.buf.reset(false)
} }

View File

@@ -279,7 +279,13 @@ func TestHTTP2(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
var b [len(http2.ClientPreface)]byte var b [len(http2.ClientPreface)]byte
if _, err := muxedConn.Read(b[:]); err != io.EOF { var n int
// We have the sniffed buffer first...
if n, err = muxedConn.Read(b[:]); err == io.EOF {
t.Fatal(err)
}
// and then we read from the source.
if _, err = muxedConn.Read(b[n:]); err != io.EOF {
t.Fatal(err) t.Fatal(err)
} }
if string(b[:]) != http2.ClientPreface { if string(b[:]) != http2.ClientPreface {