yamux_deadlock.go

package main

import (
	"bufio"
	"bytes"
	"fmt"
	"log"
	"net"
	"strings"
	"time"

	"github.com/hashicorp/yamux"
)

const address = "127.0.0.1:7000"

func acceptLoop(listener net.Listener, connChan chan net.Conn) {
	for {
		conn, err := listener.Accept()
		if err != nil {
			close(connChan)
			return
		}
		connChan <- conn
	}
}

func yamuxAcceptLoop(session *yamux.Session) {
	for {
		sConn, err := session.Accept()
		log.Printf("accepted")
		if err != nil {
			log.Fatal(err)
		}
		go yamuxServe(sConn)
	}
}

func yamuxServe(conn net.Conn) {
	defer conn.Close()

	r := bufio.NewReader(conn)
	for {
		line, err := r.ReadString('\n')
		if err != nil {
			log.Printf("server error reading from stream: %v", err)
			return
		}
		line = strings.TrimSpace(line)
		log.Printf("server received message: `%s`", line)
		if line == "5" {
			log.Printf("terminating server side")
			return
		}
	}
}

func main() {
	// set up a yamux.Server and yamux.Client connected over loopback TCP:
	listener, err := net.Listen("tcp", address)
	if err != nil {
		log.Fatal(err)
	}
	connChan := make(chan net.Conn, 1)
	go acceptLoop(listener, connChan)

	clientConn, err := net.Dial("tcp", address)
	if err != nil {
		log.Fatal(err)
	}
	serverConn, ok := <-connChan
	if !ok {
		log.Fatal("could not obtain server connection")
	}

	serverSession, err := yamux.Server(serverConn, nil)
	if err != nil {
		log.Fatal(err)
	}
	clientSession, err := yamux.Client(clientConn, nil)
	if err != nil {
		log.Fatal(err)
	}

	// run server side (accept some messages, then close the stream):
	go yamuxAcceptLoop(serverSession)

	// run client side: send some messages
	cStream, err := clientSession.Open()
	if err != nil {
		log.Fatal(err)
	}

	for i := 0; i < 10; i++ {
		// compose a message to send:
		// "4 [many spaces to exhaust send window] \n"
		var message bytes.Buffer
		fmt.Fprintf(&message, "%d", i)
		for j := 0; j < 100000; j++ {
			fmt.Fprintf(&message, " ")
		}
		fmt.Fprintf(&message, "\n")
		log.Printf("writing %d", i)
		_, err := cStream.Write(message.Bytes())
		log.Printf("wrote %d with error %v", i, err)
		time.Sleep(100 * time.Millisecond)
	}
}