--- /dev/null
+package yamux
+
+import (
+ "bytes"
+ "fmt"
+ "io"
+ "io/ioutil"
+ "log"
+ "reflect"
+ "runtime"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+)
+
+type logCapture struct{ bytes.Buffer }
+
+func (l *logCapture) logs() []string {
+ return strings.Split(strings.TrimSpace(l.String()), "\n")
+}
+
+func (l *logCapture) match(expect []string) bool {
+ return reflect.DeepEqual(l.logs(), expect)
+}
+
+func captureLogs(s *Session) *logCapture {
+ buf := new(logCapture)
+ s.logger = log.New(buf, "", 0)
+ return buf
+}
+
+type pipeConn struct {
+ reader *io.PipeReader
+ writer *io.PipeWriter
+ writeBlocker sync.Mutex
+}
+
+func (p *pipeConn) Read(b []byte) (int, error) {
+ return p.reader.Read(b)
+}
+
+func (p *pipeConn) Write(b []byte) (int, error) {
+ p.writeBlocker.Lock()
+ defer p.writeBlocker.Unlock()
+ return p.writer.Write(b)
+}
+
+func (p *pipeConn) Close() error {
+ p.reader.Close()
+ return p.writer.Close()
+}
+
+func testConn() (io.ReadWriteCloser, io.ReadWriteCloser) {
+ read1, write1 := io.Pipe()
+ read2, write2 := io.Pipe()
+ conn1 := &pipeConn{reader: read1, writer: write2}
+ conn2 := &pipeConn{reader: read2, writer: write1}
+ return conn1, conn2
+}
+
+func testConf() *Config {
+ conf := DefaultConfig()
+ conf.AcceptBacklog = 64
+ conf.KeepAliveInterval = 100 * time.Millisecond
+ conf.ConnectionWriteTimeout = 250 * time.Millisecond
+ return conf
+}
+
+func testConfNoKeepAlive() *Config {
+ conf := testConf()
+ conf.EnableKeepAlive = false
+ return conf
+}
+
+func testClientServer() (*Session, *Session) {
+ return testClientServerConfig(testConf())
+}
+
+func testClientServerConfig(conf *Config) (*Session, *Session) {
+ conn1, conn2 := testConn()
+ client, _ := Client(conn1, conf)
+ server, _ := Server(conn2, conf)
+ return client, server
+}
+
+func TestPing(t *testing.T) {
+ client, server := testClientServer()
+ defer client.Close()
+ defer server.Close()
+
+ rtt, err := client.Ping()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if rtt == 0 {
+ t.Fatalf("bad: %v", rtt)
+ }
+
+ rtt, err = server.Ping()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if rtt == 0 {
+ t.Fatalf("bad: %v", rtt)
+ }
+}
+
+func TestPing_Timeout(t *testing.T) {
+ client, server := testClientServerConfig(testConfNoKeepAlive())
+ defer client.Close()
+ defer server.Close()
+
+ // Prevent the client from responding
+ clientConn := client.conn.(*pipeConn)
+ clientConn.writeBlocker.Lock()
+
+ errCh := make(chan error, 1)
+ go func() {
+ _, err := server.Ping() // Ping via the server session
+ errCh <- err
+ }()
+
+ select {
+ case err := <-errCh:
+ if err != ErrTimeout {
+ t.Fatalf("err: %v", err)
+ }
+ case <-time.After(client.config.ConnectionWriteTimeout * 2):
+ t.Fatalf("failed to timeout within expected %v", client.config.ConnectionWriteTimeout)
+ }
+
+ // Verify that we recover, even if we gave up
+ clientConn.writeBlocker.Unlock()
+
+ go func() {
+ _, err := server.Ping() // Ping via the server session
+ errCh <- err
+ }()
+
+ select {
+ case err := <-errCh:
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ case <-time.After(client.config.ConnectionWriteTimeout):
+ t.Fatalf("timeout")
+ }
+}
+
+func TestCloseBeforeAck(t *testing.T) {
+ cfg := testConf()
+ cfg.AcceptBacklog = 8
+ client, server := testClientServerConfig(cfg)
+
+ defer client.Close()
+ defer server.Close()
+
+ for i := 0; i < 8; i++ {
+ s, err := client.OpenStream()
+ if err != nil {
+ t.Fatal(err)
+ }
+ s.Close()
+ }
+
+ for i := 0; i < 8; i++ {
+ s, err := server.AcceptStream()
+ if err != nil {
+ t.Fatal(err)
+ }
+ s.Close()
+ }
+
+ done := make(chan struct{})
+ go func() {
+ defer close(done)
+ s, err := client.OpenStream()
+ if err != nil {
+ t.Fatal(err)
+ }
+ s.Close()
+ }()
+
+ select {
+ case <-done:
+ case <-time.After(time.Second * 5):
+ t.Fatal("timed out trying to open stream")
+ }
+}
+
+func TestAccept(t *testing.T) {
+ client, server := testClientServer()
+ defer client.Close()
+ defer server.Close()
+
+ if client.NumStreams() != 0 {
+ t.Fatalf("bad")
+ }
+ if server.NumStreams() != 0 {
+ t.Fatalf("bad")
+ }
+
+ wg := &sync.WaitGroup{}
+ wg.Add(4)
+
+ go func() {
+ defer wg.Done()
+ stream, err := server.AcceptStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if id := stream.StreamID(); id != 1 {
+ t.Fatalf("bad: %v", id)
+ }
+ if err := stream.Close(); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ }()
+
+ go func() {
+ defer wg.Done()
+ stream, err := client.AcceptStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if id := stream.StreamID(); id != 2 {
+ t.Fatalf("bad: %v", id)
+ }
+ if err := stream.Close(); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ }()
+
+ go func() {
+ defer wg.Done()
+ stream, err := server.OpenStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if id := stream.StreamID(); id != 2 {
+ t.Fatalf("bad: %v", id)
+ }
+ if err := stream.Close(); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ }()
+
+ go func() {
+ defer wg.Done()
+ stream, err := client.OpenStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if id := stream.StreamID(); id != 1 {
+ t.Fatalf("bad: %v", id)
+ }
+ if err := stream.Close(); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ }()
+
+ doneCh := make(chan struct{})
+ go func() {
+ wg.Wait()
+ close(doneCh)
+ }()
+
+ select {
+ case <-doneCh:
+ case <-time.After(time.Second):
+ panic("timeout")
+ }
+}
+
+func TestNonNilInterface(t *testing.T) {
+ _, server := testClientServer()
+ server.Close()
+
+ conn, err := server.Accept()
+ if err != nil && conn != nil {
+ t.Error("bad: accept should return a connection of nil value")
+ }
+
+ conn, err = server.Open()
+ if err != nil && conn != nil {
+ t.Error("bad: open should return a connection of nil value")
+ }
+}
+
+func TestSendData_Small(t *testing.T) {
+ client, server := testClientServer()
+ defer client.Close()
+ defer server.Close()
+
+ wg := &sync.WaitGroup{}
+ wg.Add(2)
+
+ go func() {
+ defer wg.Done()
+ stream, err := server.AcceptStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+
+ if server.NumStreams() != 1 {
+ t.Fatalf("bad")
+ }
+
+ buf := make([]byte, 4)
+ for i := 0; i < 1000; i++ {
+ n, err := stream.Read(buf)
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if n != 4 {
+ t.Fatalf("short read: %d", n)
+ }
+ if string(buf) != "test" {
+ t.Fatalf("bad: %s", buf)
+ }
+ }
+
+ if err := stream.Close(); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ }()
+
+ go func() {
+ defer wg.Done()
+ stream, err := client.Open()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+
+ if client.NumStreams() != 1 {
+ t.Fatalf("bad")
+ }
+
+ for i := 0; i < 1000; i++ {
+ n, err := stream.Write([]byte("test"))
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if n != 4 {
+ t.Fatalf("short write %d", n)
+ }
+ }
+
+ if err := stream.Close(); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ }()
+
+ doneCh := make(chan struct{})
+ go func() {
+ wg.Wait()
+ close(doneCh)
+ }()
+ select {
+ case <-doneCh:
+ case <-time.After(time.Second):
+ panic("timeout")
+ }
+
+ if client.NumStreams() != 0 {
+ t.Fatalf("bad")
+ }
+ if server.NumStreams() != 0 {
+ t.Fatalf("bad")
+ }
+}
+
+func TestSendData_Large(t *testing.T) {
+ client, server := testClientServer()
+ defer client.Close()
+ defer server.Close()
+
+ const (
+ sendSize = 250 * 1024 * 1024
+ recvSize = 4 * 1024
+ )
+
+ data := make([]byte, sendSize)
+ for idx := range data {
+ data[idx] = byte(idx % 256)
+ }
+
+ wg := &sync.WaitGroup{}
+ wg.Add(2)
+
+ go func() {
+ defer wg.Done()
+ stream, err := server.AcceptStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ var sz int
+ buf := make([]byte, recvSize)
+ for i := 0; i < sendSize/recvSize; i++ {
+ n, err := stream.Read(buf)
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if n != recvSize {
+ t.Fatalf("short read: %d", n)
+ }
+ sz += n
+ for idx := range buf {
+ if buf[idx] != byte(idx%256) {
+ t.Fatalf("bad: %v %v %v", i, idx, buf[idx])
+ }
+ }
+ }
+
+ if err := stream.Close(); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+
+ t.Logf("cap=%d, n=%d\n", stream.recvBuf.Cap(), sz)
+ }()
+
+ go func() {
+ defer wg.Done()
+ stream, err := client.Open()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+
+ n, err := stream.Write(data)
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if n != len(data) {
+ t.Fatalf("short write %d", n)
+ }
+
+ if err := stream.Close(); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ }()
+
+ doneCh := make(chan struct{})
+ go func() {
+ wg.Wait()
+ close(doneCh)
+ }()
+ select {
+ case <-doneCh:
+ case <-time.After(5 * time.Second):
+ panic("timeout")
+ }
+}
+
+func TestGoAway(t *testing.T) {
+ client, server := testClientServer()
+ defer client.Close()
+ defer server.Close()
+
+ if err := server.GoAway(); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+
+ _, err := client.Open()
+ if err != ErrRemoteGoAway {
+ t.Fatalf("err: %v", err)
+ }
+}
+
+func TestManyStreams(t *testing.T) {
+ client, server := testClientServer()
+ defer client.Close()
+ defer server.Close()
+
+ wg := &sync.WaitGroup{}
+
+ acceptor := func(i int) {
+ defer wg.Done()
+ stream, err := server.AcceptStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ buf := make([]byte, 512)
+ for {
+ n, err := stream.Read(buf)
+ if err == io.EOF {
+ return
+ }
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if n == 0 {
+ t.Fatalf("err: %v", err)
+ }
+ }
+ }
+ sender := func(i int) {
+ defer wg.Done()
+ stream, err := client.Open()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ msg := fmt.Sprintf("%08d", i)
+ for i := 0; i < 1000; i++ {
+ n, err := stream.Write([]byte(msg))
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if n != len(msg) {
+ t.Fatalf("short write %d", n)
+ }
+ }
+ }
+
+ for i := 0; i < 50; i++ {
+ wg.Add(2)
+ go acceptor(i)
+ go sender(i)
+ }
+
+ wg.Wait()
+}
+
+func TestManyStreams_PingPong(t *testing.T) {
+ client, server := testClientServer()
+ defer client.Close()
+ defer server.Close()
+
+ wg := &sync.WaitGroup{}
+
+ ping := []byte("ping")
+ pong := []byte("pong")
+
+ acceptor := func(i int) {
+ defer wg.Done()
+ stream, err := server.AcceptStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ buf := make([]byte, 4)
+ for {
+ // Read the 'ping'
+ n, err := stream.Read(buf)
+ if err == io.EOF {
+ return
+ }
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if n != 4 {
+ t.Fatalf("err: %v", err)
+ }
+ if !bytes.Equal(buf, ping) {
+ t.Fatalf("bad: %s", buf)
+ }
+
+ // Shrink the internal buffer!
+ stream.Shrink()
+
+ // Write out the 'pong'
+ n, err = stream.Write(pong)
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if n != 4 {
+ t.Fatalf("err: %v", err)
+ }
+ }
+ }
+ sender := func(i int) {
+ defer wg.Done()
+ stream, err := client.OpenStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ buf := make([]byte, 4)
+ for i := 0; i < 1000; i++ {
+ // Send the 'ping'
+ n, err := stream.Write(ping)
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if n != 4 {
+ t.Fatalf("short write %d", n)
+ }
+
+ // Read the 'pong'
+ n, err = stream.Read(buf)
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if n != 4 {
+ t.Fatalf("err: %v", err)
+ }
+ if !bytes.Equal(buf, pong) {
+ t.Fatalf("bad: %s", buf)
+ }
+
+ // Shrink the buffer
+ stream.Shrink()
+ }
+ }
+
+ for i := 0; i < 50; i++ {
+ wg.Add(2)
+ go acceptor(i)
+ go sender(i)
+ }
+
+ wg.Wait()
+}
+
+func TestHalfClose(t *testing.T) {
+ client, server := testClientServer()
+ defer client.Close()
+ defer server.Close()
+
+ stream, err := client.Open()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if _, err = stream.Write([]byte("a")); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+
+ stream2, err := server.Accept()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ stream2.Close() // Half close
+
+ buf := make([]byte, 4)
+ n, err := stream2.Read(buf)
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if n != 1 {
+ t.Fatalf("bad: %v", n)
+ }
+
+ // Send more
+ if _, err = stream.Write([]byte("bcd")); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ stream.Close()
+
+ // Read after close
+ n, err = stream2.Read(buf)
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if n != 3 {
+ t.Fatalf("bad: %v", n)
+ }
+
+ // EOF after close
+ n, err = stream2.Read(buf)
+ if err != io.EOF {
+ t.Fatalf("err: %v", err)
+ }
+ if n != 0 {
+ t.Fatalf("bad: %v", n)
+ }
+}
+
+func TestReadDeadline(t *testing.T) {
+ client, server := testClientServer()
+ defer client.Close()
+ defer server.Close()
+
+ stream, err := client.Open()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ stream2, err := server.Accept()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream2.Close()
+
+ if err := stream.SetReadDeadline(time.Now().Add(5 * time.Millisecond)); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+
+ buf := make([]byte, 4)
+ if _, err := stream.Read(buf); err != ErrTimeout {
+ t.Fatalf("err: %v", err)
+ }
+}
+
+func TestWriteDeadline(t *testing.T) {
+ client, server := testClientServer()
+ defer client.Close()
+ defer server.Close()
+
+ stream, err := client.Open()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ stream2, err := server.Accept()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream2.Close()
+
+ if err := stream.SetWriteDeadline(time.Now().Add(50 * time.Millisecond)); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+
+ buf := make([]byte, 512)
+ for i := 0; i < int(initialStreamWindow); i++ {
+ _, err := stream.Write(buf)
+ if err != nil && err == ErrTimeout {
+ return
+ } else if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ }
+ t.Fatalf("Expected timeout")
+}
+
+func TestBacklogExceeded(t *testing.T) {
+ client, server := testClientServer()
+ defer client.Close()
+ defer server.Close()
+
+ // Fill the backlog
+ max := client.config.AcceptBacklog
+ for i := 0; i < max; i++ {
+ stream, err := client.Open()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ if _, err := stream.Write([]byte("foo")); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ }
+
+ // Attempt to open a new stream
+ errCh := make(chan error, 1)
+ go func() {
+ _, err := client.Open()
+ errCh <- err
+ }()
+
+ // Shutdown the server
+ go func() {
+ time.Sleep(10 * time.Millisecond)
+ server.Close()
+ }()
+
+ select {
+ case err := <-errCh:
+ if err == nil {
+ t.Fatalf("open should fail")
+ }
+ case <-time.After(time.Second):
+ t.Fatalf("timeout")
+ }
+}
+
+func TestKeepAlive(t *testing.T) {
+ client, server := testClientServer()
+ defer client.Close()
+ defer server.Close()
+
+ time.Sleep(200 * time.Millisecond)
+
+ // Ping value should increase
+ client.pingLock.Lock()
+ defer client.pingLock.Unlock()
+ if client.pingID == 0 {
+ t.Fatalf("should ping")
+ }
+
+ server.pingLock.Lock()
+ defer server.pingLock.Unlock()
+ if server.pingID == 0 {
+ t.Fatalf("should ping")
+ }
+}
+
+func TestKeepAlive_Timeout(t *testing.T) {
+ conn1, conn2 := testConn()
+
+ clientConf := testConf()
+ clientConf.ConnectionWriteTimeout = time.Hour // We're testing keep alives, not connection writes
+ clientConf.EnableKeepAlive = false // Just test one direction, so it's deterministic who hangs up on whom
+ client, _ := Client(conn1, clientConf)
+ defer client.Close()
+
+ server, _ := Server(conn2, testConf())
+ defer server.Close()
+
+ _ = captureLogs(client) // Client logs aren't part of the test
+ serverLogs := captureLogs(server)
+
+ errCh := make(chan error, 1)
+ go func() {
+ _, err := server.Accept() // Wait until server closes
+ errCh <- err
+ }()
+
+ // Prevent the client from responding
+ clientConn := client.conn.(*pipeConn)
+ clientConn.writeBlocker.Lock()
+
+ select {
+ case err := <-errCh:
+ if err != ErrKeepAliveTimeout {
+ t.Fatalf("unexpected error: %v", err)
+ }
+ case <-time.After(1 * time.Second):
+ t.Fatalf("timeout waiting for timeout")
+ }
+
+ if !server.IsClosed() {
+ t.Fatalf("server should have closed")
+ }
+
+ if !serverLogs.match([]string{"[ERR] yamux: keepalive failed: i/o deadline reached"}) {
+ t.Fatalf("server log incorect: %v", serverLogs.logs())
+ }
+}
+
+func TestLargeWindow(t *testing.T) {
+ conf := DefaultConfig()
+ conf.MaxStreamWindowSize *= 2
+
+ client, server := testClientServerConfig(conf)
+ defer client.Close()
+ defer server.Close()
+
+ stream, err := client.Open()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ stream2, err := server.Accept()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream2.Close()
+
+ stream.SetWriteDeadline(time.Now().Add(10 * time.Millisecond))
+ buf := make([]byte, conf.MaxStreamWindowSize)
+ n, err := stream.Write(buf)
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if n != len(buf) {
+ t.Fatalf("short write: %d", n)
+ }
+}
+
+type UnlimitedReader struct{}
+
+func (u *UnlimitedReader) Read(p []byte) (int, error) {
+ runtime.Gosched()
+ return len(p), nil
+}
+
+func TestSendData_VeryLarge(t *testing.T) {
+ client, server := testClientServer()
+ defer client.Close()
+ defer server.Close()
+
+ var n int64 = 1 * 1024 * 1024 * 1024
+ var workers int = 16
+
+ wg := &sync.WaitGroup{}
+ wg.Add(workers * 2)
+
+ for i := 0; i < workers; i++ {
+ go func() {
+ defer wg.Done()
+ stream, err := server.AcceptStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ buf := make([]byte, 4)
+ _, err = stream.Read(buf)
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if !bytes.Equal(buf, []byte{0, 1, 2, 3}) {
+ t.Fatalf("bad header")
+ }
+
+ recv, err := io.Copy(ioutil.Discard, stream)
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if recv != n {
+ t.Fatalf("bad: %v", recv)
+ }
+ }()
+ }
+ for i := 0; i < workers; i++ {
+ go func() {
+ defer wg.Done()
+ stream, err := client.Open()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ _, err = stream.Write([]byte{0, 1, 2, 3})
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+
+ unlimited := &UnlimitedReader{}
+ sent, err := io.Copy(stream, io.LimitReader(unlimited, n))
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if sent != n {
+ t.Fatalf("bad: %v", sent)
+ }
+ }()
+ }
+
+ doneCh := make(chan struct{})
+ go func() {
+ wg.Wait()
+ close(doneCh)
+ }()
+ select {
+ case <-doneCh:
+ case <-time.After(20 * time.Second):
+ panic("timeout")
+ }
+}
+
+func TestBacklogExceeded_Accept(t *testing.T) {
+ client, server := testClientServer()
+ defer client.Close()
+ defer server.Close()
+
+ max := 5 * client.config.AcceptBacklog
+ go func() {
+ for i := 0; i < max; i++ {
+ stream, err := server.Accept()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+ }
+ }()
+
+ // Fill the backlog
+ for i := 0; i < max; i++ {
+ stream, err := client.Open()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ if _, err := stream.Write([]byte("foo")); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ }
+}
+
+func TestSession_WindowUpdateWriteDuringRead(t *testing.T) {
+ client, server := testClientServerConfig(testConfNoKeepAlive())
+ defer client.Close()
+ defer server.Close()
+
+ var wg sync.WaitGroup
+ wg.Add(2)
+
+ // Choose a huge flood size that we know will result in a window update.
+ flood := int64(client.config.MaxStreamWindowSize) - 1
+
+ // The server will accept a new stream and then flood data to it.
+ go func() {
+ defer wg.Done()
+
+ stream, err := server.AcceptStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ n, err := stream.Write(make([]byte, flood))
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if int64(n) != flood {
+ t.Fatalf("short write: %d", n)
+ }
+ }()
+
+ // The client will open a stream, block outbound writes, and then
+ // listen to the flood from the server, which should time out since
+ // it won't be able to send the window update.
+ go func() {
+ defer wg.Done()
+
+ stream, err := client.OpenStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ conn := client.conn.(*pipeConn)
+ conn.writeBlocker.Lock()
+
+ _, err = stream.Read(make([]byte, flood))
+ if err != ErrConnectionWriteTimeout {
+ t.Fatalf("err: %v", err)
+ }
+ }()
+
+ wg.Wait()
+}
+
+func TestSession_PartialReadWindowUpdate(t *testing.T) {
+ client, server := testClientServerConfig(testConfNoKeepAlive())
+ defer client.Close()
+ defer server.Close()
+
+ var wg sync.WaitGroup
+ wg.Add(1)
+
+ // Choose a huge flood size that we know will result in a window update.
+ flood := int64(client.config.MaxStreamWindowSize)
+ var wr *Stream
+
+ // The server will accept a new stream and then flood data to it.
+ go func() {
+ defer wg.Done()
+
+ var err error
+ wr, err = server.AcceptStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer wr.Close()
+
+ if wr.sendWindow != client.config.MaxStreamWindowSize {
+ t.Fatalf("sendWindow: exp=%d, got=%d", client.config.MaxStreamWindowSize, wr.sendWindow)
+ }
+
+ n, err := wr.Write(make([]byte, flood))
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ if int64(n) != flood {
+ t.Fatalf("short write: %d", n)
+ }
+ if wr.sendWindow != 0 {
+ t.Fatalf("sendWindow: exp=%d, got=%d", 0, wr.sendWindow)
+ }
+ }()
+
+ stream, err := client.OpenStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ wg.Wait()
+
+ _, err = stream.Read(make([]byte, flood/2+1))
+
+ if exp := uint32(flood/2 + 1); wr.sendWindow != exp {
+ t.Errorf("sendWindow: exp=%d, got=%d", exp, wr.sendWindow)
+ }
+}
+
+func TestSession_sendNoWait_Timeout(t *testing.T) {
+ client, server := testClientServerConfig(testConfNoKeepAlive())
+ defer client.Close()
+ defer server.Close()
+
+ var wg sync.WaitGroup
+ wg.Add(2)
+
+ go func() {
+ defer wg.Done()
+
+ stream, err := server.AcceptStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+ }()
+
+ // The client will open the stream and then block outbound writes, we'll
+ // probe sendNoWait once it gets into that state.
+ go func() {
+ defer wg.Done()
+
+ stream, err := client.OpenStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ conn := client.conn.(*pipeConn)
+ conn.writeBlocker.Lock()
+
+ hdr := header(make([]byte, headerSize))
+ hdr.encode(typePing, flagACK, 0, 0)
+ for {
+ err = client.sendNoWait(hdr)
+ if err == nil {
+ continue
+ } else if err == ErrConnectionWriteTimeout {
+ break
+ } else {
+ t.Fatalf("err: %v", err)
+ }
+ }
+ }()
+
+ wg.Wait()
+}
+
+func TestSession_PingOfDeath(t *testing.T) {
+ client, server := testClientServerConfig(testConfNoKeepAlive())
+ defer client.Close()
+ defer server.Close()
+
+ var wg sync.WaitGroup
+ wg.Add(2)
+
+ var doPingOfDeath sync.Mutex
+ doPingOfDeath.Lock()
+
+ // This is used later to block outbound writes.
+ conn := server.conn.(*pipeConn)
+
+ // The server will accept a stream, block outbound writes, and then
+ // flood its send channel so that no more headers can be queued.
+ go func() {
+ defer wg.Done()
+
+ stream, err := server.AcceptStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ conn.writeBlocker.Lock()
+ for {
+ hdr := header(make([]byte, headerSize))
+ hdr.encode(typePing, 0, 0, 0)
+ err = server.sendNoWait(hdr)
+ if err == nil {
+ continue
+ } else if err == ErrConnectionWriteTimeout {
+ break
+ } else {
+ t.Fatalf("err: %v", err)
+ }
+ }
+
+ doPingOfDeath.Unlock()
+ }()
+
+ // The client will open a stream and then send the server a ping once it
+ // can no longer write. This makes sure the server doesn't deadlock reads
+ // while trying to reply to the ping with no ability to write.
+ go func() {
+ defer wg.Done()
+
+ stream, err := client.OpenStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ // This ping will never unblock because the ping id will never
+ // show up in a response.
+ doPingOfDeath.Lock()
+ go func() { client.Ping() }()
+
+ // Wait for a while to make sure the previous ping times out,
+ // then turn writes back on and make sure a ping works again.
+ time.Sleep(2 * server.config.ConnectionWriteTimeout)
+ conn.writeBlocker.Unlock()
+ if _, err = client.Ping(); err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ }()
+
+ wg.Wait()
+}
+
+func TestSession_ConnectionWriteTimeout(t *testing.T) {
+ client, server := testClientServerConfig(testConfNoKeepAlive())
+ defer client.Close()
+ defer server.Close()
+
+ var wg sync.WaitGroup
+ wg.Add(2)
+
+ go func() {
+ defer wg.Done()
+
+ stream, err := server.AcceptStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+ }()
+
+ // The client will open the stream and then block outbound writes, we'll
+ // tee up a write and make sure it eventually times out.
+ go func() {
+ defer wg.Done()
+
+ stream, err := client.OpenStream()
+ if err != nil {
+ t.Fatalf("err: %v", err)
+ }
+ defer stream.Close()
+
+ conn := client.conn.(*pipeConn)
+ conn.writeBlocker.Lock()
+
+ // Since the write goroutine is blocked then this will return a
+ // timeout since it can't get feedback about whether the write
+ // worked.
+ n, err := stream.Write([]byte("hello"))
+ if err != ErrConnectionWriteTimeout {
+ t.Fatalf("err: %v", err)
+ }
+ if n != 0 {
+ t.Fatalf("lied about writes: %d", n)
+ }
+ }()
+
+ wg.Wait()
+}
OSDN Git Service
