+++ /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
