From 94bfdd575c5956b1e470d8b170bcf4749333d15a Mon Sep 17 00:00:00 2001 From: mohanson Date: Wed, 15 Jul 2026 11:49:50 +0800 Subject: [PATCH] Etch protocol's client now correctly reuse conn --- protocol/ashe/engine.go | 16 ++--- protocol/baboon/engine.go | 26 +++---- protocol/czar/engine.go | 15 ++-- protocol/etch/engine.go | 146 +++++++++++++++++++++++++++----------- 4 files changed, 134 insertions(+), 69 deletions(-) diff --git a/protocol/ashe/engine.go b/protocol/ashe/engine.go index 94cc88c..928f6c0 100644 --- a/protocol/ashe/engine.go +++ b/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. diff --git a/protocol/baboon/engine.go b/protocol/baboon/engine.go index a9c82e2..6e0db85 100644 --- a/protocol/baboon/engine.go +++ b/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,6 +176,11 @@ 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]) @@ -183,22 +188,17 @@ func (c *Client) Dial(ctx *daze.Context, network string, address string) (io.Rea 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. diff --git a/protocol/czar/engine.go b/protocol/czar/engine.go index 35b744b..4560b23 100644 --- a/protocol/czar/engine.go +++ b/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: diff --git a/protocol/etch/engine.go b/protocol/etch/engine.go index 2bdcbb4..7d86ad0 100644 --- a/protocol/etch/engine.go +++ b/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 }