Merge pull request #28 from whyrusleeping/feat/deadline

Make deadlines interrupt and coalesce writes

Author: Stebalien

LICENSE-BSD

@@ -0,0 +1,27 @@
+Copyright (c) 2009 The Go Authors. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+   * Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+   * Redistributions in binary form must reproduce the above
+copyright notice, this list of conditions and the following disclaimer
+in the documentation and/or other materials provided with the
+distribution.
+   * Neither the name of Google Inc. nor the names of its
+contributors may be used to endorse or promote products derived from
+this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

bench_test.go

@@ -1,6 +1,7 @@
 package yamux
 
 import (
+	"io"
 	"testing"
 )
 
@@ -60,7 +61,7 @@ func BenchmarkSendRecv(b *testing.B) {
 		}
 		defer stream.Close()
 		for i := 0; i < b.N; i++ {
-			if _, err := stream.Read(recvBuf); err != nil {
+			if _, err := io.ReadFull(stream, recvBuf); err != nil {
 				b.Fatalf("err: %v", err)
 			}
 		}
@@ -101,7 +102,7 @@ func BenchmarkSendRecvLarge(b *testing.B) {
 		defer stream.Close()
 		for i := 0; i < b.N; i++ {
 			for j := 0; j < sendSize/recvSize; j++ {
-				if _, err := stream.Read(recvBuf); err != nil {
+				if _, err := io.ReadFull(stream, recvBuf); err != nil {
 					b.Fatalf("err: %v", err)
 				}
 			}

const.go

@@ -6,59 +6,68 @@ import (
 )
 
 type YamuxError struct {
-	msg string
+	msg                string
+	timeout, temporary bool
 }
 
 func (ye YamuxError) Error() string {
 	return ye.msg
 }
 
+func (ye YamuxError) Timeout() bool {
+	return ye.timeout
+}
+
+func (ye YamuxError) Temporary() bool {
+	return ye.temporary
+}
+
 var (
 	// ErrInvalidVersion means we received a frame with an
 	// invalid version
-	ErrInvalidVersion = &YamuxError{"invalid protocol version"}
+	ErrInvalidVersion = &YamuxError{msg: "invalid protocol version"}
 
 	// ErrInvalidMsgType means we received a frame with an
 	// invalid message type
-	ErrInvalidMsgType = &YamuxError{"invalid msg type"}
+	ErrInvalidMsgType = &YamuxError{msg: "invalid msg type"}
 
 	// ErrSessionShutdown is used if there is a shutdown during
 	// an operation
-	ErrSessionShutdown = &YamuxError{"session shutdown"}
+	ErrSessionShutdown = &YamuxError{msg: "session shutdown"}
 
 	// ErrStreamsExhausted is returned if we have no more
 	// stream ids to issue
-	ErrStreamsExhausted = &YamuxError{"streams exhausted"}
+	ErrStreamsExhausted = &YamuxError{msg: "streams exhausted"}
 
 	// ErrDuplicateStream is used if a duplicate stream is
 	// opened inbound
-	ErrDuplicateStream = &YamuxError{"duplicate stream initiated"}
+	ErrDuplicateStream = &YamuxError{msg: "duplicate stream initiated"}
 
 	// ErrReceiveWindowExceeded indicates the window was exceeded
-	ErrRecvWindowExceeded = &YamuxError{"recv window exceeded"}
+	ErrRecvWindowExceeded = &YamuxError{msg: "recv window exceeded"}
 
 	// ErrTimeout is used when we reach an IO deadline
-	ErrTimeout = &YamuxError{"i/o deadline reached"}
+	ErrTimeout = &YamuxError{msg: "i/o deadline reached", timeout: true, temporary: true}
 
 	// ErrStreamClosed is returned when using a closed stream
-	ErrStreamClosed = &YamuxError{"stream closed"}
+	ErrStreamClosed = &YamuxError{msg: "stream closed"}
 
 	// ErrUnexpectedFlag is set when we get an unexpected flag
-	ErrUnexpectedFlag = &YamuxError{"unexpected flag"}
+	ErrUnexpectedFlag = &YamuxError{msg: "unexpected flag"}
 
 	// ErrRemoteGoAway is used when we get a go away from the other side
-	ErrRemoteGoAway = &YamuxError{"remote end is not accepting connections"}
+	ErrRemoteGoAway = &YamuxError{msg: "remote end is not accepting connections"}
 
 	// ErrConnectionReset is sent if a stream is reset. This can happen
 	// if the backlog is exceeded, or if there was a remote GoAway.
-	ErrConnectionReset = &YamuxError{"stream reset"}
+	ErrConnectionReset = &YamuxError{msg: "stream reset"}
 
 	// ErrConnectionWriteTimeout indicates that we hit the "safety valve"
 	// timeout writing to the underlying stream connection.
-	ErrConnectionWriteTimeout = &YamuxError{"connection write timeout"}
+	ErrConnectionWriteTimeout = &YamuxError{msg: "connection write timeout", timeout: true}
 
 	// ErrKeepAliveTimeout is sent if a missed keepalive caused the stream close
-	ErrKeepAliveTimeout = &YamuxError{"keepalive timeout"}
+	ErrKeepAliveTimeout = &YamuxError{msg: "keepalive timeout", timeout: true}
 )
 
 const (
@@ -129,7 +138,7 @@ const (
 		sizeOfStreamID + sizeOfLength
 )
 
-type header []byte
+type header [headerSize]byte
 
 func (h header) Version() uint8 {
 	return h[0]
@@ -156,10 +165,12 @@ func (h header) String() string {
 		h.Version(), h.MsgType(), h.Flags(), h.StreamID(), h.Length())
 }
 
-func (h header) encode(msgType uint8, flags uint16, streamID uint32, length uint32) {
+func encode(msgType uint8, flags uint16, streamID uint32, length uint32) header {
+	var h header
 	h[0] = protoVersion
 	h[1] = msgType
 	binary.BigEndian.PutUint16(h[2:4], flags)
 	binary.BigEndian.PutUint32(h[4:8], streamID)
 	binary.BigEndian.PutUint32(h[8:12], length)
+	return h
 }

const_test.go

@@ -51,8 +51,7 @@ func TestConst(t *testing.T) {
 }
 
 func TestEncodeDecode(t *testing.T) {
-	hdr := header(make([]byte, headerSize))
-	hdr.encode(typeWindowUpdate, flagACK|flagRST, 1234, 4321)
+	hdr := encode(typeWindowUpdate, flagACK|flagRST, 1234, 4321)
 
 	if hdr.Version() != protoVersion {
 		t.Fatalf("bad: %v", hdr)

deadline.go

@@ -0,0 +1,80 @@
+// Copied from the go standard library.
+//
+// Copyright 2010 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE-BSD file.
+
+package yamux
+
+import (
+	"sync"
+	"time"
+)
+
+// pipeDeadline is an abstraction for handling timeouts.
+type pipeDeadline struct {
+	mu     sync.Mutex // Guards timer and cancel
+	timer  *time.Timer
+	cancel chan struct{} // Must be non-nil
+}
+
+func makePipeDeadline() pipeDeadline {
+	return pipeDeadline{cancel: make(chan struct{})}
+}
+
+// set sets the point in time when the deadline will time out.
+// A timeout event is signaled by closing the channel returned by waiter.
+// Once a timeout has occurred, the deadline can be refreshed by specifying a
+// t value in the future.
+//
+// A zero value for t prevents timeout.
+func (d *pipeDeadline) set(t time.Time) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+
+	if d.timer != nil && !d.timer.Stop() {
+		<-d.cancel // Wait for the timer callback to finish and close cancel
+	}
+	d.timer = nil
+
+	// Time is zero, then there is no deadline.
+	closed := isClosedChan(d.cancel)
+	if t.IsZero() {
+		if closed {
+			d.cancel = make(chan struct{})
+		}
+		return
+	}
+
+	// Time in the future, setup a timer to cancel in the future.
+	if dur := time.Until(t); dur > 0 {
+		if closed {
+			d.cancel = make(chan struct{})
+		}
+		d.timer = time.AfterFunc(dur, func() {
+			close(d.cancel)
+		})
+		return
+	}
+
+	// Time in the past, so close immediately.
+	if !closed {
+		close(d.cancel)
+	}
+}
+
+// wait returns a channel that is closed when the deadline is exceeded.
+func (d *pipeDeadline) wait() chan struct{} {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	return d.cancel
+}
+
+func isClosedChan(c <-chan struct{}) bool {
+	select {
+	case <-c:
+		return true
+	default:
+		return false
+	}
+}

go.mod

@@ -0,0 +1,5 @@
+module github.com/whyrusleeping/yamux
+
+go 1.12
+
+require github.com/libp2p/go-buffer-pool v0.0.1

go.sum

@@ -0,0 +1,2 @@
+github.com/libp2p/go-buffer-pool v0.0.1 h1:9Rrn/H46cXjaA2HQ5Y8lyhOS1NhTkZ4yuEs2r3Eechg=
+github.com/libp2p/go-buffer-pool v0.0.1/go.mod h1:xtyIz9PMobb13WaxR6Zo1Pd1zXJKYg0a8KiIvDp3TzQ=

mux.go

@@ -3,6 +3,7 @@ package yamux
 import (
 	"fmt"
 	"io"
+	"net"
 	"os"
 	"time"
 )
@@ -37,6 +38,15 @@ type Config struct {
 	//
 	// Set to 0 to disable it.
 	ReadBufSize int
+
+	// WriteCoalesceDelay is the maximum amount of time we'll delay
+	// coalescing a packet before sending it. This should be on the order of
+	// micro-milliseconds.
+	WriteCoalesceDelay time.Duration
+
+	// MaxMessageSize is the maximum size of a message that we'll send on a
+	// stream. This ensures that a single stream doesn't hog a connection.
+	MaxMessageSize uint32
 }
 
 // DefaultConfig is used to return a default configuration
@@ -49,6 +59,8 @@ func DefaultConfig() *Config {
 		MaxStreamWindowSize:    initialStreamWindow,
 		LogOutput:              os.Stderr,
 		ReadBufSize:            4096,
+		MaxMessageSize:         64 * 1024, // Means 64KiB/10s = 52kbps minimum speed.
+		WriteCoalesceDelay:     10 * time.Microsecond,
 	}
 }
 
@@ -63,13 +75,19 @@ func VerifyConfig(config *Config) error {
 	if config.MaxStreamWindowSize < initialStreamWindow {
 		return fmt.Errorf("MaxStreamWindowSize must be larger than %d", initialStreamWindow)
 	}
+	if config.MaxMessageSize < 1024 {
+		return fmt.Errorf("MaxMessageSize must be greater than a kilobyte")
+	}
+	if config.WriteCoalesceDelay < 0 {
+		return fmt.Errorf("WriteCoalesceDelay must be >= 0")
+	}
 	return nil
 }
 
 // Server is used to initialize a new server-side connection.
 // There must be at most one server-side connection. If a nil config is
 // provided, the DefaultConfiguration will be used.
-func Server(conn io.ReadWriteCloser, config *Config) (*Session, error) {
+func Server(conn net.Conn, config *Config) (*Session, error) {
 	if config == nil {
 		config = DefaultConfig()
 	}
@@ -81,7 +99,7 @@ func Server(conn io.ReadWriteCloser, config *Config) (*Session, error) {
 
 // Client is used to initialize a new client-side connection.
 // There must be at most one client-side connection.
-func Client(conn io.ReadWriteCloser, config *Config) (*Session, error) {
+func Client(conn net.Conn, config *Config) (*Session, error) {
 	if config == nil {
 		config = DefaultConfig()
 	}