Etch protocol's client now correctly reuse conn

protocol/ashe/engine.go

@@ -355,17 +355,17 @@ func (c *Client) Dial(ctx *daze.Context, network string, address string) (io.Rea
 	if err != nil {
 		return nil, err
 	}
-	con, err := c.Estab(ctx, srv, network, address)
+	rtc := &daze.ReadWriteCloser{
+		Reader: io.TeeReader(srv, rate.NewLimitsWriter(c.Limits)),
+		Writer: io.MultiWriter(srv, rate.NewLimitsWriter(c.Limits)),
+		Closer: srv,
+	}
+	con, err := c.Estab(ctx, rtc, network, address)
 	if err != nil {
-		srv.Close()
+		rtc.Close()
 		return nil, err
 	}
-	rtc := &daze.ReadWriteCloser{
-		Reader: io.TeeReader(con, rate.NewLimitsWriter(c.Limits)),
-		Writer: io.MultiWriter(con, rate.NewLimitsWriter(c.Limits)),
-		Closer: con,
-	}
-	return rtc, nil
+	return con, nil
 }
 
 // NewClient returns a new Client. Cipher is a password in string form, with no length limit.

protocol/baboon/engine.go

@@ -76,14 +76,14 @@ func (s *Server) ServeDaze(w http.ResponseWriter, r *http.Request) {
 		Writer: cc,
 		Closer: cc,
 	}
-	spy := &ashe.Server{Cipher: s.Cipher}
-	ctx := &daze.Context{Cid: atomic.AddUint32(&s.NextID, 1)}
-	log.Printf("conn: %08x accept remote=%s", ctx.Cid, cc.RemoteAddr())
 	rtc := &daze.ReadWriteCloser{
 		Reader: io.TeeReader(cli, rate.NewLimitsWriter(s.Limits)),
 		Writer: io.MultiWriter(cli, rate.NewLimitsWriter(s.Limits)),
 		Closer: cli,
 	}
+	spy := &ashe.Server{Cipher: s.Cipher}
+	ctx := &daze.Context{Cid: atomic.AddUint32(&s.NextID, 1)}
+	log.Printf("conn: %08x accept remote=%s", ctx.Cid, cc.RemoteAddr())
 	if err := spy.Serve(ctx, rtc); err != nil {
 		log.Printf("conn: %08x  error %s", ctx.Cid, err)
 	}
@@ -176,29 +176,29 @@ func (c *Client) Dial(ctx *daze.Context, network string, address string) (io.Rea
 	if err != nil {
 		return nil, err
 	}
+	rtc := &daze.ReadWriteCloser{
+		Reader: io.TeeReader(srv, rate.NewLimitsWriter(c.Limits)),
+		Writer: io.MultiWriter(srv, rate.NewLimitsWriter(c.Limits)),
+		Closer: srv,
+	}
 	buf = make([]byte, 32)
 	io.ReadFull(&daze.RandomReader{}, buf[:16])
 	copy(buf[16:], c.Cipher[:16])
 	sign := md5.Sum(buf)
 	copy(buf[16:], sign[:])
 	req = doa.Try(http.NewRequest("POST", "http://"+c.Server+"/sync", http.NoBody))
 	req.Header.Set("Authorization", hex.EncodeToString(buf))
-	req.Write(srv)
+	req.Write(rtc)
 	// Discard responded header
 	buf = make([]byte, 147)
-	io.ReadFull(srv, buf)
+	io.ReadFull(rtc, buf)
 	spy := &ashe.Client{Cipher: c.Cipher}
-	con, err := spy.Estab(ctx, srv, network, address)
+	con, err := spy.Estab(ctx, rtc, network, address)
 	if err != nil {
-		srv.Close()
+		rtc.Close()
 		return nil, err
 	}
-	rtc := &daze.ReadWriteCloser{
-		Reader: io.TeeReader(con, rate.NewLimitsWriter(c.Limits)),
-		Writer: io.MultiWriter(con, rate.NewLimitsWriter(c.Limits)),
-		Closer: con,
-	}
-	return rtc, nil
+	return con, nil
 }
 
 // NewClient returns a new Client. Cipher is a password in string form, with no length limit.

protocol/czar/engine.go

@@ -173,12 +173,7 @@ func (c *Client) Dial(ctx *daze.Context, network string, address string) (io.Rea
 			srv.Close()
 			return nil, err
 		}
-		rtc := &daze.ReadWriteCloser{
-			Reader: io.TeeReader(con, rate.NewLimitsWriter(c.Limits)),
-			Writer: io.MultiWriter(con, rate.NewLimitsWriter(c.Limits)),
-			Closer: con,
-		}
-		return rtc, nil
+		return con, nil
 	case <-time.After(daze.Conf.DialerTimeout):
 		return nil, fmt.Errorf("dial tcp: %s: i/o timeout", address)
 	}
@@ -196,6 +191,7 @@ func (c *Client) Run() {
 	var (
 		err error
 		mux *Mux
+		rtc *daze.ReadWriteCloser
 		rtt = 0
 		sid = 0
 		srv net.Conn
@@ -221,7 +217,12 @@ func (c *Client) Run() {
 			}
 		case clientStatusDialSuccess:
 			log.Println("czar: mux init")
-			mux = NewMuxClient(srv)
+			rtc = &daze.ReadWriteCloser{
+				Reader: io.TeeReader(srv, rate.NewLimitsWriter(c.Limits)),
+				Writer: io.MultiWriter(srv, rate.NewLimitsWriter(c.Limits)),
+				Closer: srv,
+			}
+			mux = NewMuxClient(rtc)
 			rtt = 0
 			sid = clientStatusEstablished
 		case clientStatusEstablished:

protocol/etch/engine.go

@@ -8,6 +8,7 @@ import (
 	"crypto/tls"
 	"crypto/x509"
 	"errors"
+	"fmt"
 	"io"
 	"log"
 	"math"
@@ -55,18 +56,14 @@ var (
 // Stream wraps a quic stream as an io.ReadWriteCloser. Writes are flushed immediately so that the ashe handshake,
 // which exchanges very short messages, progresses without waiting for the quic stream buffer to fill.
 type Stream struct {
-	con *quic.Conn
 	rem net.Addr
 	stm *quic.Stream
 }
 
-// Close closes the stream. If the stream owns its connection, the connection is closed as well.
+// Close closes the stream.
 func (s *Stream) Close() error {
 	s.stm.CloseRead()
 	s.stm.CloseWrite()
-	if s.con != nil {
-		return s.con.Close()
-	}
 	return nil
 }
 
@@ -106,7 +103,7 @@ func (s *Server) Serve(ctx *daze.Context, cli io.ReadWriteCloser) error {
 	return spy.Serve(ctx, cli)
 }
 
-// Close listener. Established connections will not be closed.
+// Close closes the listener and aborts established connections managed by this endpoint.
 func (s *Server) Close() error {
 	if s.EpQuic != nil {
 		return s.EpQuic.Close(context.Background())
@@ -139,17 +136,20 @@ func (s *Server) Run() error {
 				for {
 					stm, err := con.AcceptStream(context.Background())
 					if err != nil {
+						if !errors.Is(err, net.ErrClosed) {
+							log.Println("main:", err)
+						}
 						return
 					}
-					cid := atomic.AddUint32(&idx, 1)
-					ctx := &daze.Context{Cid: cid}
 					cli := &Stream{rem: rem, stm: stm}
-					log.Printf("conn: %08x accept remote=%s", ctx.Cid, rem)
 					rtc := &daze.ReadWriteCloser{
 						Reader: io.TeeReader(cli, rate.NewLimitsWriter(s.Limits)),
 						Writer: io.MultiWriter(cli, rate.NewLimitsWriter(s.Limits)),
 						Closer: cli,
 					}
+					cid := atomic.AddUint32(&idx, 1)
+					ctx := &daze.Context{Cid: cid}
+					log.Printf("conn: %08x accept remote=%s", ctx.Cid, rem)
 					go func() {
 						defer rtc.Close()
 						if err := s.Serve(ctx, rtc); err != nil {
@@ -176,59 +176,123 @@ func NewServer(listen string, cipher string) *Server {
 
 // Client implemented the ashe-over-quic protocol.
 type Client struct {
+	Cancel chan struct{}
 	Cipher []byte
 	EpQuic *quic.Endpoint
 	Limits *rate.Limits
+	Mux    chan *quic.Conn
 	Server string
 }
 
-// Close releases the underlying QUIC endpoint.
+// Close the connection. All streams will be closed at the same time.
 func (c *Client) Close() error {
-	if c.EpQuic != nil {
-		return c.EpQuic.Close(context.Background())
-	}
+	close(c.Cancel)
 	return nil
 }
 
 // Dial connects to the address on the named network.
 func (c *Client) Dial(ctx *daze.Context, network string, address string) (io.ReadWriteCloser, error) {
-	cty, end := context.WithTimeout(context.Background(), daze.Conf.DialerTimeout)
-	defer end()
-	con, err := c.EpQuic.Dial(cty, "udp", c.Server, ClientConfig.Do())
-	if err != nil {
-		return nil, err
-	}
-	stm, err := con.NewStream(cty)
-	if err != nil {
-		con.Close()
-		return nil, err
-	}
-	if err := stm.Flush(); err != nil {
-		con.Close()
-		return nil, err
-	}
-	rem := net.UDPAddrFromAddrPort(con.RemoteAddr())
-	srv := &Stream{con: con, rem: rem, stm: stm}
-	spy := &ashe.Client{Cipher: c.Cipher}
-	out, err := spy.Estab(ctx, srv, network, address)
-	if err != nil {
-		srv.Close()
-		return nil, err
+	select {
+	case mux := <-c.Mux:
+		cty, end := context.WithTimeout(context.Background(), daze.Conf.DialerTimeout)
+		stm, err := mux.NewStream(cty)
+		end()
+		if err != nil {
+			return nil, err
+		}
+		rem := net.UDPAddrFromAddrPort(mux.RemoteAddr())
+		srv := &Stream{rem: rem, stm: stm}
+		rtc := &daze.ReadWriteCloser{
+			Reader: io.TeeReader(srv, rate.NewLimitsWriter(c.Limits)),
+			Writer: io.MultiWriter(srv, rate.NewLimitsWriter(c.Limits)),
+			Closer: srv,
+		}
+		spy := &ashe.Client{Cipher: c.Cipher}
+		con, err := spy.Estab(ctx, rtc, network, address)
+		if err != nil {
+			rtc.Close()
+			return nil, err
+		}
+		return con, nil
+	case <-time.After(daze.Conf.DialerTimeout):
+		return nil, fmt.Errorf("dial udp: %s: i/o timeout", address)
 	}
-	rtc := &daze.ReadWriteCloser{
-		Reader: io.TeeReader(out, rate.NewLimitsWriter(c.Limits)),
-		Writer: io.MultiWriter(out, rate.NewLimitsWriter(c.Limits)),
-		Closer: out,
+}
+
+// Run creates and maintains an established connection to the etch server.
+func (c *Client) Run() {
+	const (
+		clientStatusClosed int = iota
+		clientStatusDialFailure
+		clientStatusDialSuccess
+		clientStatusEstablished
+		clientStatusCancel
+	)
+	var (
+		don chan error
+		err error
+		mux *quic.Conn
+		rtt = 0
+		sid = 0
+	)
+	for {
+		switch sid {
+		case clientStatusClosed:
+			cty, end := context.WithTimeout(context.Background(), daze.Conf.DialerTimeout)
+			mux, err = c.EpQuic.Dial(cty, "udp", c.Server, ClientConfig.Do())
+			end()
+			if err != nil {
+				sid = clientStatusDialFailure
+			} else {
+				sid = clientStatusDialSuccess
+			}
+		case clientStatusDialFailure:
+			log.Println("etch:", err)
+			select {
+			case <-time.After(time.Second * time.Duration(math.Pow(2, float64(rtt)))):
+				// A slow start reconnection algorithm.
+				rtt = min(rtt+1, 5)
+				sid = clientStatusClosed
+			case <-c.Cancel:
+				sid = clientStatusCancel
+			}
+		case clientStatusDialSuccess:
+			log.Println("etch: quic init")
+			rtt = 0
+			don = make(chan error, 1)
+			go func(mux *quic.Conn) {
+				don <- mux.Wait(context.Background())
+			}(mux)
+			sid = clientStatusEstablished
+		case clientStatusEstablished:
+			select {
+			case c.Mux <- mux:
+			case err = <-don:
+				log.Println("etch: quic done", err)
+				mux.Close()
+				sid = clientStatusClosed
+			case <-c.Cancel:
+				log.Println("etch: quic done")
+				mux.Close()
+				sid = clientStatusCancel
+			}
+		case clientStatusCancel:
+			c.EpQuic.Close(context.Background())
+			return
+		}
 	}
-	return rtc, nil
 }
 
 // NewClient returns a new Client. Cipher is a password in string form, with no length limit.
 func NewClient(server string, cipher string) *Client {
-	return &Client{
+	client := &Client{
+		Cancel: make(chan struct{}),
 		Cipher: daze.Salt(cipher),
 		EpQuic: doa.Try(quic.Listen("udp", ":0", ClientConfig.Do())),
 		Limits: rate.NewLimits(math.MaxUint32, time.Second),
+		Mux:    make(chan *quic.Conn),
 		Server: server,
 	}
+	go client.Run()
+	return client
 }