mirror of
https://github.com/go-gost/x.git
synced 2026-10-19 16:16:32 +00:00
feat(p2p): move tunnel data onto the gRPC Tunnel stream
TunnelProvider becomes OpenTunnelStream: OpenTunnel authorizes the tunnel and issues a single-use id, the Tunnel stream carries the data. The new x/p2p/streamconn conn wraps the stream as a net.Conn with real Set*Deadline support (bounded Recv pump, idempotent Close, framed mode for udp). The tunnel base dialer collapses to the stream; its failure path closes the conn nil-safely instead of leaking.
This commit is contained in:
ginuerzh
+59
-23
@@ -6,14 +6,22 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net"
|
||||
"time"
|
||||
|
||||
"github.com/go-gost/core/logger"
|
||||
"github.com/go-gost/plugin/p2p/proto"
|
||||
xp2p "github.com/go-gost/x/p2p"
|
||||
"github.com/go-gost/x/internal/plugin"
|
||||
xp2p "github.com/go-gost/x/p2p"
|
||||
"github.com/go-gost/x/p2p/streamconn"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/metadata"
|
||||
)
|
||||
|
||||
// rpcTimeout caps the control RPC (OpenTunnel) with a fresh context so a
|
||||
// wedged plugin cannot stall a dial.
|
||||
const rpcTimeout = 3 * time.Second
|
||||
|
||||
type grpcPlugin struct {
|
||||
conn grpc.ClientConnInterface
|
||||
client proto.P2PClient
|
||||
@@ -23,8 +31,8 @@ type grpcPlugin struct {
|
||||
// NewGRPCPlugin creates a p2p.TunnelProvider backed by a gRPC plugin. The
|
||||
// constructor mirrors the recorder plugin's relaxed shape (connection
|
||||
// failures are logged, a non-nil provider is always returned), but the
|
||||
// runtime semantics are fail-closed: OpenTunnel/CloseTunnel on an
|
||||
// unavailable plugin return errors, never no-ops.
|
||||
// runtime semantics are fail-closed: OpenTunnelStream on an unavailable
|
||||
// plugin returns an error, never a no-op.
|
||||
func NewGRPCPlugin(name string, addr string, opts ...plugin.Option) xp2p.TunnelProvider {
|
||||
var options plugin.Options
|
||||
for _, opt := range opts {
|
||||
@@ -50,36 +58,64 @@ func NewGRPCPlugin(name string, addr string, opts ...plugin.Option) xp2p.TunnelP
|
||||
return p
|
||||
}
|
||||
|
||||
func (p *grpcPlugin) OpenTunnel(ctx context.Context, network, peer string) (id, endpoint string, err error) {
|
||||
// OpenTunnelStream runs the two-RPC tunnel setup: OpenTunnel authorizes the
|
||||
// tunnel and issues its id (the stream credential), then the Tunnel bidi
|
||||
// stream — bound to the id via the "id" metadata key — carries the data. The
|
||||
// returned conn's Close ends the stream, which the host treats as the tunnel
|
||||
// teardown.
|
||||
func (p *grpcPlugin) OpenTunnelStream(ctx context.Context, network, peer string) (net.Conn, error) {
|
||||
if p.client == nil {
|
||||
return "", "", errors.New("p2p: plugin unavailable")
|
||||
return nil, errors.New("p2p: plugin unavailable")
|
||||
}
|
||||
reply, err := p.client.OpenTunnel(ctx, &proto.OpenTunnelRequest{Peer: peer, Network: network})
|
||||
|
||||
octx, cancel := context.WithTimeout(ctx, rpcTimeout)
|
||||
reply, err := p.client.OpenTunnel(octx, &proto.OpenTunnelRequest{Peer: peer, Network: network})
|
||||
cancel()
|
||||
if err != nil {
|
||||
p.log.Error(err)
|
||||
return "", "", err
|
||||
return nil, err
|
||||
}
|
||||
if !reply.Ok {
|
||||
return "", "", errors.New(reply.Error)
|
||||
return nil, errors.New(reply.Error)
|
||||
}
|
||||
return reply.Id, reply.Endpoint, nil
|
||||
|
||||
// The stream outlives Dial: derive its context from Background, not the
|
||||
// dial ctx. Its cancel is the conn's abort — cancelling the client stream
|
||||
// unblocks a parked Recv/Send and tells the host to tear the tunnel down.
|
||||
sctx, scancel := context.WithCancel(context.Background())
|
||||
sctx = metadata.AppendToOutgoingContext(sctx, "id", reply.Id)
|
||||
|
||||
// Bound the establishment: a wedged plugin must not hang Dial (which
|
||||
// holds the base dialer's lock) waiting for a transport that never comes.
|
||||
estTimer := time.AfterFunc(rpcTimeout, scancel)
|
||||
stream, err := p.client.Tunnel(sctx)
|
||||
if !estTimer.Stop() && err == nil {
|
||||
// The timer fired while the stream was being established: scancel has
|
||||
// run, so the stream is canceled and unusable — fail instead of
|
||||
// returning a dead conn.
|
||||
err = errors.New("p2p: tunnel stream establishment timed out")
|
||||
}
|
||||
if err != nil {
|
||||
scancel()
|
||||
p.log.Error(err)
|
||||
return nil, err
|
||||
}
|
||||
return streamconn.New(stream, scancel, network,
|
||||
streamAddr{network: network, addr: "p2p"},
|
||||
streamAddr{network: network, addr: peer}), nil
|
||||
}
|
||||
|
||||
func (p *grpcPlugin) CloseTunnel(ctx context.Context, id string) error {
|
||||
if p.client == nil {
|
||||
return errors.New("p2p: plugin unavailable")
|
||||
}
|
||||
reply, err := p.client.CloseTunnel(ctx, &proto.CloseTunnelRequest{Id: id})
|
||||
if err != nil {
|
||||
p.log.Error(err)
|
||||
return err
|
||||
}
|
||||
if !reply.Ok {
|
||||
return errors.New("p2p: close tunnel failed")
|
||||
}
|
||||
return nil
|
||||
// streamAddr is the synthetic address of a gRPC-carried tunnel. The stream
|
||||
// has no socket, but callers read LocalAddr/RemoteAddr (e.g. the udp dialer's
|
||||
// ReadFrom), so they must be non-nil.
|
||||
type streamAddr struct {
|
||||
network string
|
||||
addr string
|
||||
}
|
||||
|
||||
func (a streamAddr) Network() string { return a.network }
|
||||
func (a streamAddr) String() string { return a.addr }
|
||||
|
||||
func (p *grpcPlugin) Close() error {
|
||||
if p.conn == nil {
|
||||
return nil
|
||||
@@ -89,4 +125,4 @@ func (p *grpcPlugin) Close() error {
|
||||
return closer.Close()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
+182
-173
@@ -19,8 +19,10 @@ import (
|
||||
"github.com/go-gost/x/dialer/udp"
|
||||
xlogger "github.com/go-gost/x/logger"
|
||||
xp2p "github.com/go-gost/x/p2p"
|
||||
"github.com/go-gost/x/p2p/streamconn"
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc/codes"
|
||||
"google.golang.org/grpc/metadata"
|
||||
"google.golang.org/grpc/status"
|
||||
)
|
||||
|
||||
@@ -33,25 +35,27 @@ func TestMain(m *testing.M) {
|
||||
}
|
||||
|
||||
// fakeServer implements the P2P control contract: OpenTunnel validates the
|
||||
// peer, creates a local listener bridged to it, CloseTunnel tears it down.
|
||||
// Flags flip the failure paths: deadEndpoint returns an endpoint whose port
|
||||
// is already closed, bizFail answers ok:false body with a gRPC status OK.
|
||||
// peer and allocates the tunnel id (no endpoint — the data plane is the
|
||||
// Tunnel stream); Tunnel serves that stream, bridged to the target (tcp) or
|
||||
// framed-echoed (udp). A record is removed when its Tunnel handler returns:
|
||||
// stream end IS the tunnel teardown. bizFail answers ok:false with a gRPC
|
||||
// status OK.
|
||||
type fakeServer struct {
|
||||
proto.UnimplementedP2PServer
|
||||
mu sync.Mutex
|
||||
seq atomic.Int64
|
||||
tunnels map[string]*fakeTunnel
|
||||
networks []string // network of every OpenTunnel request, in order
|
||||
deadEndpoint bool
|
||||
bizFail bool
|
||||
mu sync.Mutex
|
||||
seq atomic.Int64
|
||||
tunnels map[string]fakeTunnel
|
||||
networks []string // network of every OpenTunnel request, in order
|
||||
bizFail bool
|
||||
}
|
||||
|
||||
type fakeTunnel struct {
|
||||
target string
|
||||
ln net.Listener
|
||||
pc net.PacketConn // udp tunnels only
|
||||
mu sync.Mutex
|
||||
conns map[net.Conn]struct{}
|
||||
network string
|
||||
target string
|
||||
}
|
||||
|
||||
func newFakeServer() *fakeServer {
|
||||
return &fakeServer{tunnels: make(map[string]fakeTunnel)}
|
||||
}
|
||||
|
||||
func (s *fakeServer) OpenTunnel(ctx context.Context, req *proto.OpenTunnelRequest) (*proto.OpenTunnelReply, error) {
|
||||
@@ -65,46 +69,75 @@ func (s *fakeServer) OpenTunnel(ctx context.Context, req *proto.OpenTunnelReques
|
||||
if s.bizFail {
|
||||
return &proto.OpenTunnelReply{Ok: false, Error: "biz boom"}, nil
|
||||
}
|
||||
if req.Network == "udp" {
|
||||
pc, err := net.ListenPacket("udp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
return &proto.OpenTunnelReply{Ok: false, Error: err.Error()}, nil
|
||||
}
|
||||
t := &fakeTunnel{target: req.Peer, pc: pc, conns: make(map[net.Conn]struct{})}
|
||||
id := fmt.Sprintf("t-%d", s.seq.Add(1))
|
||||
s.mu.Lock()
|
||||
s.tunnels[id] = t
|
||||
s.mu.Unlock()
|
||||
return &proto.OpenTunnelReply{Ok: true, Id: id, Endpoint: pc.LocalAddr().String()}, nil
|
||||
network := req.Network
|
||||
if network == "" {
|
||||
network = "tcp"
|
||||
}
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
return &proto.OpenTunnelReply{Ok: false, Error: err.Error()}, nil
|
||||
}
|
||||
endpoint := ln.Addr().String()
|
||||
if s.deadEndpoint {
|
||||
ln.Close() // endpoint is dead; the tunnel is still tracked for cleanup
|
||||
}
|
||||
t := &fakeTunnel{target: req.Peer, ln: ln, conns: make(map[net.Conn]struct{})}
|
||||
id := fmt.Sprintf("t-%d", s.seq.Add(1))
|
||||
s.mu.Lock()
|
||||
s.tunnels[id] = t
|
||||
s.tunnels[id] = fakeTunnel{network: network, target: req.Peer}
|
||||
s.mu.Unlock()
|
||||
if !s.deadEndpoint {
|
||||
go t.serve()
|
||||
}
|
||||
return &proto.OpenTunnelReply{Ok: true, Id: id, Endpoint: endpoint}, nil
|
||||
return &proto.OpenTunnelReply{Ok: true, Id: id}, nil
|
||||
}
|
||||
|
||||
func (s *fakeServer) CloseTunnel(ctx context.Context, req *proto.CloseTunnelRequest) (*proto.CloseTunnelReply, error) {
|
||||
s.mu.Lock()
|
||||
t, ok := s.tunnels[req.Id]
|
||||
delete(s.tunnels, req.Id)
|
||||
s.mu.Unlock()
|
||||
if ok {
|
||||
t.close()
|
||||
// Tunnel is the data plane: look the tunnel up by the "id" metadata key
|
||||
// (unknown → NotFound), then serve the stream. The host side of the stream
|
||||
// stays raw; in udp mode the framing is carried end to end as bytes (the
|
||||
// fake plays the framed peer).
|
||||
func (s *fakeServer) Tunnel(stream proto.P2P_TunnelServer) error {
|
||||
md, _ := metadata.FromIncomingContext(stream.Context())
|
||||
var id string
|
||||
if v := md.Get("id"); len(v) > 0 {
|
||||
id = v[0]
|
||||
}
|
||||
return &proto.CloseTunnelReply{Ok: true}, nil
|
||||
s.mu.Lock()
|
||||
t, ok := s.tunnels[id]
|
||||
s.mu.Unlock()
|
||||
if !ok {
|
||||
return status.Errorf(codes.NotFound, "unknown tunnel id")
|
||||
}
|
||||
defer func() {
|
||||
s.mu.Lock()
|
||||
delete(s.tunnels, id)
|
||||
s.mu.Unlock()
|
||||
}()
|
||||
|
||||
if t.network == "udp" {
|
||||
c := streamconn.New(stream, nil, "udp", nil, nil)
|
||||
defer c.Close()
|
||||
buf := make([]byte, streamconn.MaxFrame)
|
||||
for {
|
||||
n, err := c.Read(buf)
|
||||
if err != nil {
|
||||
return nil // client closed: normal teardown
|
||||
}
|
||||
if _, err := c.Write(buf[:n]); err != nil {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
up, err := net.DialTimeout("tcp", t.target, 5*time.Second)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer up.Close()
|
||||
down := streamconn.New(stream, nil, "tcp", nil, nil)
|
||||
defer down.Close()
|
||||
done := make(chan struct{}, 2)
|
||||
go func() {
|
||||
io.Copy(up, down)
|
||||
up.Close()
|
||||
done <- struct{}{}
|
||||
}()
|
||||
go func() {
|
||||
io.Copy(down, up)
|
||||
down.Close()
|
||||
done <- struct{}{}
|
||||
}()
|
||||
<-done
|
||||
<-done
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *fakeServer) Status(ctx context.Context, req *proto.StatusRequest) (*proto.StatusReply, error) {
|
||||
@@ -129,59 +162,16 @@ func (s *fakeServer) networkOf(i int) string {
|
||||
return s.networks[i]
|
||||
}
|
||||
|
||||
func (t *fakeTunnel) serve() {
|
||||
for {
|
||||
conn, err := t.ln.Accept()
|
||||
if err != nil {
|
||||
func waitTunnelCount(t *testing.T, s *fakeServer, want int) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(3 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if s.tunnelCount() == want {
|
||||
return
|
||||
}
|
||||
t.mu.Lock()
|
||||
t.conns[conn] = struct{}{}
|
||||
t.mu.Unlock()
|
||||
go t.bridge(conn)
|
||||
}
|
||||
}
|
||||
|
||||
func (t *fakeTunnel) bridge(conn net.Conn) {
|
||||
defer func() {
|
||||
t.mu.Lock()
|
||||
delete(t.conns, conn)
|
||||
t.mu.Unlock()
|
||||
}()
|
||||
up, err := net.DialTimeout("tcp", t.target, 5*time.Second)
|
||||
if err != nil {
|
||||
conn.Close()
|
||||
return
|
||||
}
|
||||
defer func() {
|
||||
up.Close()
|
||||
conn.Close()
|
||||
}()
|
||||
done := make(chan struct{}, 2)
|
||||
go func() {
|
||||
io.Copy(up, conn)
|
||||
done <- struct{}{}
|
||||
}()
|
||||
go func() {
|
||||
io.Copy(conn, up)
|
||||
done <- struct{}{}
|
||||
}()
|
||||
<-done
|
||||
<-done
|
||||
}
|
||||
|
||||
func (t *fakeTunnel) close() {
|
||||
if t.ln != nil {
|
||||
t.ln.Close()
|
||||
}
|
||||
if t.pc != nil {
|
||||
t.pc.Close()
|
||||
}
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
for conn := range t.conns {
|
||||
conn.Close()
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("tunnel count = %d, want %d", s.tunnelCount(), want)
|
||||
}
|
||||
|
||||
// startFake runs the fake control server and returns its address; the
|
||||
@@ -245,7 +235,7 @@ func newTestDialer(t *testing.T, s *fakeServer) (dialer.Dialer, xp2p.TunnelProvi
|
||||
}
|
||||
|
||||
func TestDialRoundTrip(t *testing.T) {
|
||||
s := &fakeServer{tunnels: make(map[string]*fakeTunnel)}
|
||||
s := newFakeServer()
|
||||
d, _ := newTestDialer(t, s)
|
||||
target := startEcho(t)
|
||||
|
||||
@@ -261,10 +251,9 @@ func TestDialRoundTrip(t *testing.T) {
|
||||
if err := conn.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n := s.tunnelCount(); n != 0 {
|
||||
t.Fatalf("tunnel count after close = %d, want 0", n)
|
||||
}
|
||||
// second Close must be a no-op (sync.Once), not an error or panic
|
||||
// The record is removed when the host-side handler returns (stream end).
|
||||
waitTunnelCount(t, s, 0)
|
||||
// second Close must be a no-op (idempotent), not an error or panic
|
||||
if err := conn.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -273,6 +262,38 @@ func TestDialRoundTrip(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// TestDialReadDeadline proves the tunnel conn honors SetReadDeadline (the
|
||||
// reason the conn exists in this shape: inner tls/ws handshakes set deadlines
|
||||
// and ignore the error, so a silent peer must not hang a Read forever).
|
||||
func TestDialReadDeadline(t *testing.T) {
|
||||
s := newFakeServer()
|
||||
d, _ := newTestDialer(t, s)
|
||||
target := startEcho(t)
|
||||
|
||||
conn, err := d.Dial(context.Background(), target)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
if err := conn.SetReadDeadline(time.Now().Add(50 * time.Millisecond)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
start := time.Now()
|
||||
if _, err := conn.Read(make([]byte, 4)); !errors.Is(err, os.ErrDeadlineExceeded) {
|
||||
t.Fatalf("Read err = %v, want os.ErrDeadlineExceeded", err)
|
||||
}
|
||||
if elapsed := time.Since(start); elapsed < 30*time.Millisecond {
|
||||
t.Fatalf("Read returned after %v, want the deadline to have applied", elapsed)
|
||||
}
|
||||
|
||||
// Clearing the deadline restores normal reads.
|
||||
if err := conn.SetReadDeadline(time.Time{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
echoOnce(t, conn)
|
||||
}
|
||||
|
||||
func TestPluginDown(t *testing.T) {
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
@@ -289,20 +310,8 @@ func TestPluginDown(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestLocalEndpointDead(t *testing.T) {
|
||||
s := &fakeServer{tunnels: make(map[string]*fakeTunnel), deadEndpoint: true}
|
||||
d, _ := newTestDialer(t, s)
|
||||
|
||||
if _, err := d.Dial(context.Background(), "127.0.0.1:9999"); err == nil {
|
||||
t.Fatal("Dial succeeded, want error on dead endpoint")
|
||||
}
|
||||
if n := s.tunnelCount(); n != 0 {
|
||||
t.Fatalf("tunnel count after failed dial = %d, want 0 (cleanup ran)", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestDialConcurrent(t *testing.T) {
|
||||
s := &fakeServer{tunnels: make(map[string]*fakeTunnel)}
|
||||
s := newFakeServer()
|
||||
d, _ := newTestDialer(t, s)
|
||||
target := startEcho(t)
|
||||
|
||||
@@ -319,13 +328,11 @@ func TestDialConcurrent(t *testing.T) {
|
||||
})
|
||||
}
|
||||
wg.Wait()
|
||||
if n := s.tunnelCount(); n != 0 {
|
||||
t.Fatalf("tunnel count after concurrent dials = %d, want 0", n)
|
||||
}
|
||||
waitTunnelCount(t, s, 0)
|
||||
}
|
||||
|
||||
func TestOpenTunnelInvalidPeer(t *testing.T) {
|
||||
s := &fakeServer{tunnels: make(map[string]*fakeTunnel)}
|
||||
s := newFakeServer()
|
||||
d, _ := newTestDialer(t, s)
|
||||
|
||||
if _, err := d.Dial(context.Background(), "foo"); err == nil {
|
||||
@@ -337,7 +344,8 @@ func TestOpenTunnelInvalidPeer(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestOpenTunnelBizFail(t *testing.T) {
|
||||
s := &fakeServer{tunnels: make(map[string]*fakeTunnel), bizFail: true}
|
||||
s := newFakeServer()
|
||||
s.bizFail = true
|
||||
d, _ := newTestDialer(t, s)
|
||||
|
||||
_, err := d.Dial(context.Background(), "127.0.0.1:9999")
|
||||
@@ -365,7 +373,7 @@ func (d *handshakeDialer) Handshake(ctx context.Context, conn net.Conn, opts ...
|
||||
}
|
||||
|
||||
func TestHandshakeForward(t *testing.T) {
|
||||
s := &fakeServer{tunnels: make(map[string]*fakeTunnel)}
|
||||
s := newFakeServer()
|
||||
provider := NewGRPCPlugin("t", startFake(t, s))
|
||||
inner := &handshakeDialer{Dialer: tcp.NewDialer(dialer.LoggerOption(logger.Default()))}
|
||||
d := xp2p.NewTunnelDialer(inner, provider)
|
||||
@@ -401,11 +409,12 @@ func TestSupportedDialer(t *testing.T) {
|
||||
}
|
||||
|
||||
// TestUDPNetworkPassthrough covers the datagram shape end to end over the
|
||||
// gRPC seam: a udp inner dialer must make the plugin return a datagram
|
||||
// endpoint (network carried on the request), and the wrapper must dial that
|
||||
// endpoint as udp — a TCP endpoint would coalesce packets.
|
||||
// gRPC seam: a udp inner dialer must make the plugin open a udp tunnel
|
||||
// (network carried on the request), the wrapped conn must preserve datagram
|
||||
// boundaries, and the dial result must expose the PacketConn shape the tun
|
||||
// handler's transport expects.
|
||||
func TestUDPNetworkPassthrough(t *testing.T) {
|
||||
s := &fakeServer{tunnels: make(map[string]*fakeTunnel)}
|
||||
s := newFakeServer()
|
||||
provider := NewGRPCPlugin("t", startFake(t, s))
|
||||
inner := udp.NewDialer(dialer.LoggerOption(logger.Default()))
|
||||
d := xp2p.NewTunnelDialer(inner, provider)
|
||||
@@ -419,14 +428,26 @@ func TestUDPNetworkPassthrough(t *testing.T) {
|
||||
if got := s.networkOf(0); got != "udp" {
|
||||
t.Fatalf("OpenTunnel network = %q, want udp", got)
|
||||
}
|
||||
if got := conn.LocalAddr().Network(); got != "udp" {
|
||||
t.Fatalf("endpoint dial network = %q, want udp", got)
|
||||
}
|
||||
// A datagram conn must expose the connected-socket ReadFrom/WriteTo shape
|
||||
// the tun handler's transport expects.
|
||||
if _, ok := conn.(net.PacketConn); !ok {
|
||||
t.Fatal("udp dial result is not a net.PacketConn")
|
||||
}
|
||||
conn.SetDeadline(time.Now().Add(3 * time.Second))
|
||||
if _, err := conn.Write([]byte("ping")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
buf := make([]byte, 8)
|
||||
if n, err := conn.Read(buf); err != nil || string(buf[:n]) != "ping" {
|
||||
t.Fatalf("datagram round trip = %q, %v; want ping", buf[:n], err)
|
||||
}
|
||||
// Boundaries preserved: consecutive datagrams never coalesce.
|
||||
conn.Write([]byte("a"))
|
||||
conn.Write([]byte("bb"))
|
||||
if n, err := conn.Read(buf); err != nil || n != 1 || buf[0] != 'a' {
|
||||
t.Fatalf("Read = %q, %v; want single-byte datagram", buf[:n], err)
|
||||
}
|
||||
if n, err := conn.Read(buf); err != nil || string(buf[:n]) != "bb" {
|
||||
t.Fatalf("Read = %q, %v; want bb (no coalescing)", buf[:n], err)
|
||||
}
|
||||
}
|
||||
|
||||
// muxDialer implements dialer.Multiplexer to exercise the wrapper's
|
||||
@@ -461,51 +482,39 @@ func TestMultiplexDelegation(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// countingProvider is a TunnelProvider recording open/close calls and the
|
||||
// network each open asked for.
|
||||
// countingProvider is a TunnelProvider recording opens/closes and the network
|
||||
// each open asked for. Each open returns an in-memory conn.
|
||||
type countingProvider struct {
|
||||
ln net.Listener
|
||||
pc net.PacketConn
|
||||
mu sync.Mutex
|
||||
openCalls int
|
||||
closeCall int
|
||||
networks []string
|
||||
closes []string
|
||||
}
|
||||
|
||||
func (p *countingProvider) OpenTunnel(ctx context.Context, network, peer string) (id, endpoint string, err error) {
|
||||
func (p *countingProvider) OpenTunnelStream(ctx context.Context, network, peer string) (net.Conn, error) {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
p.openCalls++
|
||||
p.networks = append(p.networks, network)
|
||||
if network == "udp" {
|
||||
pc, err := net.ListenPacket("udp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
p.pc = pc
|
||||
return fmt.Sprintf("t-%d", p.openCalls), pc.LocalAddr().String(), nil
|
||||
}
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
p.ln = ln
|
||||
return fmt.Sprintf("t-%d", p.openCalls), ln.Addr().String(), nil
|
||||
p.mu.Unlock()
|
||||
c1, c2 := net.Pipe()
|
||||
go c1.Close()
|
||||
return &countingConn{Conn: c2, onClose: func() {
|
||||
p.mu.Lock()
|
||||
p.closeCall++
|
||||
p.mu.Unlock()
|
||||
}}, nil
|
||||
}
|
||||
|
||||
func (p *countingProvider) CloseTunnel(ctx context.Context, id string) error {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
p.closes = append(p.closes, id)
|
||||
if p.ln != nil {
|
||||
p.ln.Close()
|
||||
p.ln = nil
|
||||
}
|
||||
if p.pc != nil {
|
||||
p.pc.Close()
|
||||
p.pc = nil
|
||||
}
|
||||
return nil
|
||||
// countingConn observes the tunnel conn's close (the tunnel teardown).
|
||||
type countingConn struct {
|
||||
net.Conn
|
||||
onClose func()
|
||||
once sync.Once
|
||||
}
|
||||
|
||||
func (c *countingConn) Close() error {
|
||||
c.once.Do(c.onClose)
|
||||
return c.Conn.Close()
|
||||
}
|
||||
|
||||
func (p *countingProvider) networkOf(i int) string {
|
||||
@@ -528,7 +537,7 @@ func (p *countingProvider) openCount() int {
|
||||
func (p *countingProvider) closeCount() int {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
return len(p.closes)
|
||||
return p.closeCall
|
||||
}
|
||||
|
||||
// noBaseDialer never touches its base net dialer — the mux session cache hit
|
||||
@@ -574,11 +583,11 @@ func TestLazyOpenTunnel(t *testing.T) {
|
||||
}
|
||||
conn.Close()
|
||||
if n := p.openCount(); n != 0 {
|
||||
t.Fatalf("OpenTunnel calls = %d, want 0 (cache hit must not leak a tunnel)", n)
|
||||
t.Fatalf("OpenTunnelStream calls = %d, want 0 (cache hit must not leak a tunnel)", n)
|
||||
}
|
||||
|
||||
// Happy path: the inner dials its base exactly once; closing the conn
|
||||
// releases the tunnel.
|
||||
// tears the tunnel down (the stream ending IS the teardown).
|
||||
p2 := &countingProvider{}
|
||||
d2 := xp2p.NewTunnelDialer(&baseDialer{}, p2)
|
||||
conn2, err := d2.Dial(context.Background(), "127.0.0.1:9999")
|
||||
@@ -586,17 +595,17 @@ func TestLazyOpenTunnel(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if n := p2.openCount(); n != 1 {
|
||||
t.Fatalf("OpenTunnel calls = %d, want 1", n)
|
||||
t.Fatalf("OpenTunnelStream calls = %d, want 1", n)
|
||||
}
|
||||
if got := p2.networkOf(0); got != "tcp" {
|
||||
t.Fatalf("OpenTunnel network = %q, want tcp for a stream inner", got)
|
||||
}
|
||||
if n := p2.closeCount(); n != 0 {
|
||||
t.Fatalf("CloseTunnel calls before conn close = %d, want 0", n)
|
||||
t.Fatalf("conn closes before conn2 close = %d, want 0", n)
|
||||
}
|
||||
conn2.Close()
|
||||
if n := p2.closeCount(); n != 1 {
|
||||
t.Fatalf("CloseTunnel calls after conn close = %d, want 1", n)
|
||||
t.Fatalf("conn closes after conn2 close = %d, want 1", n)
|
||||
}
|
||||
|
||||
// Failure path: inner.Dial fails after the base dial opened the tunnel;
|
||||
@@ -607,9 +616,9 @@ func TestLazyOpenTunnel(t *testing.T) {
|
||||
t.Fatal("Dial succeeded, want inner error")
|
||||
}
|
||||
if n := p3.openCount(); n != 1 {
|
||||
t.Fatalf("OpenTunnel calls = %d, want 1", n)
|
||||
t.Fatalf("OpenTunnelStream calls = %d, want 1", n)
|
||||
}
|
||||
if n := p3.closeCount(); n != 1 {
|
||||
t.Fatalf("CloseTunnel calls after failed dial = %d, want 1 (fail closed)", n)
|
||||
t.Fatalf("conn closes after failed dial = %d, want 1 (fail closed)", n)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,400 @@
|
||||
// Package streamconn adapts a gRPC bidirectional Tunnel byte stream (the p2p
|
||||
// plugin's data plane) to a net.Conn for the GOST side. The p2p host keeps its
|
||||
// own minimal raw conn (p2p/streamconn.go) instead of sharing this one: the
|
||||
// plugin module holds contracts only, and the host cannot import the x module.
|
||||
// The two sides couple through the proto (Chunk messages), not through code.
|
||||
//
|
||||
// Two modes:
|
||||
// - raw ("tcp"): Read/Write move the stream's bytes; writes are split into
|
||||
// chunks below gRPC's default message limit.
|
||||
// - framed ("udp"): Read returns exactly one length-prefixed datagram and
|
||||
// Write sends one, preserving datagram boundaries through the byte stream
|
||||
// (the same 2-byte BE prefix the p2p datagram channel uses — the framing
|
||||
// helpers below are the single implementation).
|
||||
//
|
||||
// # Backpressure
|
||||
//
|
||||
// A single Recv pump goroutine hands chunks to Read through a one-chunk
|
||||
// channel. A stalled consumer blocks the pump, which stops calling Recv and
|
||||
// lets gRPC's per-stream flow control throttle the sender — memory stays
|
||||
// bounded at about two chunks.
|
||||
//
|
||||
// # Deadlines
|
||||
//
|
||||
// Read deadlines are enforced by Read itself. A write deadline must interrupt
|
||||
// a Send parked in flow control, and only the client side can do that (by
|
||||
// cancelling the stream), so a Conn built without an abort func (i.e. from a
|
||||
// server stream) rejects write deadlines. Close never blocks: on a server
|
||||
// stream it only marks the conn closed — the parked Recv unblocks when the
|
||||
// handler returns, which ends the RPC and cancels the stream (grpc-go
|
||||
// finishStream calls the stream's cancel for exactly this reason).
|
||||
package streamconn
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"errors"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/go-gost/plugin/p2p/proto"
|
||||
)
|
||||
|
||||
// Stream is the data-carrying half of the P2P Tunnel RPC; it is satisfied by
|
||||
// both proto.P2P_TunnelClient and proto.P2P_TunnelServer.
|
||||
type Stream interface {
|
||||
Send(*proto.Chunk) error
|
||||
Recv() (*proto.Chunk, error)
|
||||
Context() context.Context
|
||||
}
|
||||
|
||||
// writeChunkSize bounds one Send; it stays well under gRPC's 4 MiB default
|
||||
// receive limit.
|
||||
const writeChunkSize = 32 * 1024
|
||||
|
||||
// MaxFrame is the largest payload that fits the 2-byte length prefix.
|
||||
const MaxFrame = 65535
|
||||
|
||||
// WriteFrame writes p as one length-prefixed frame.
|
||||
func WriteFrame(w io.Writer, p []byte) error {
|
||||
var hdr [2]byte
|
||||
binary.BigEndian.PutUint16(hdr[:], uint16(len(p)))
|
||||
bufs := net.Buffers{hdr[:], p}
|
||||
_, err := bufs.WriteTo(w)
|
||||
return err
|
||||
}
|
||||
|
||||
// ReadFrame reads one length-prefixed frame. A zero-length frame returns
|
||||
// (nil, nil); the caller should skip it.
|
||||
func ReadFrame(r io.Reader) ([]byte, error) {
|
||||
var hdr [2]byte
|
||||
if _, err := io.ReadFull(r, hdr[:]); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
n := int(binary.BigEndian.Uint16(hdr[:]))
|
||||
if n == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
p := make([]byte, n)
|
||||
if _, err := io.ReadFull(r, p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return p, nil
|
||||
}
|
||||
|
||||
var _ net.Conn = (*Conn)(nil)
|
||||
|
||||
// Conn is a net.Conn over a Stream. abort, when non-nil, cancels the
|
||||
// underlying stream (the client side's context.CancelFunc); the host side
|
||||
// passes nil because a server handler cannot abort its stream.
|
||||
func New(s Stream, abort func(), network string, local, remote net.Addr) *Conn {
|
||||
c := &Conn{
|
||||
s: s,
|
||||
abort: abort,
|
||||
framed: network == "udp",
|
||||
local: local,
|
||||
remote: remote,
|
||||
chunks: make(chan *proto.Chunk, 1),
|
||||
done: make(chan struct{}),
|
||||
closed: make(chan struct{}),
|
||||
rdWake: make(chan struct{}),
|
||||
}
|
||||
go c.pump()
|
||||
return c
|
||||
}
|
||||
|
||||
type Conn struct {
|
||||
s Stream
|
||||
abort func()
|
||||
framed bool
|
||||
local, remote net.Addr
|
||||
|
||||
chunks chan *proto.Chunk // cap 1: the pump blocks here when the consumer is behind
|
||||
done chan struct{} // closed by the pump when Recv ends
|
||||
closed chan struct{} // closed by Close
|
||||
rdWake chan struct{} // closed and replaced by SetReadDeadline to wake parked Reads
|
||||
|
||||
closeOnce sync.Once
|
||||
sendMu sync.Mutex // serializes Send (gRPC allows one sender)
|
||||
|
||||
mu sync.Mutex // guards err, rb, rd, wd
|
||||
err error // Recv's terminal error
|
||||
rb []byte // chunk bytes not yet consumed by Read
|
||||
rd, wd time.Time
|
||||
}
|
||||
|
||||
// pump moves chunks from Recv to Read. It blocks on the one-chunk handoff, so
|
||||
// Recv is never called while a chunk is unconsumed — that is the backpressure.
|
||||
// Every exit path funnels through run's return value: the terminal error is
|
||||
// stored and done closed, so a reader parked in Read always wakes. That
|
||||
// matters most when the transport dies under a server-side pump (the client
|
||||
// aborts the stream): nothing else would ever wake the host's pipe, hanging
|
||||
// the handler and leaking the record.
|
||||
func (c *Conn) pump() {
|
||||
err := c.run()
|
||||
c.mu.Lock()
|
||||
if c.err == nil {
|
||||
c.err = err
|
||||
}
|
||||
c.mu.Unlock()
|
||||
close(c.done)
|
||||
}
|
||||
|
||||
func (c *Conn) run() error {
|
||||
for {
|
||||
chunk, err := c.s.Recv()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
select {
|
||||
case c.chunks <- chunk:
|
||||
case <-c.closed:
|
||||
return net.ErrClosed
|
||||
case <-c.s.Context().Done():
|
||||
// The stream died (client abort/RST on a server stream, or a
|
||||
// canceled client context).
|
||||
return c.s.Context().Err()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// nextChunk returns the next chunk payload, honoring closed, the pump's
|
||||
// terminal state, and the read deadline. A deadline set while parked wakes
|
||||
// the select (rdWake) and re-evaluates; a chunk delivered before the terminal
|
||||
// state is returned first, the error on the following call — otherwise a
|
||||
// random select could drop the last chunk of a clean EOF (silent truncation).
|
||||
func (c *Conn) nextChunk() ([]byte, error) {
|
||||
select {
|
||||
case <-c.closed:
|
||||
return nil, net.ErrClosed
|
||||
default:
|
||||
}
|
||||
|
||||
for {
|
||||
c.mu.Lock()
|
||||
dl := c.rd
|
||||
wake := c.rdWake
|
||||
c.mu.Unlock()
|
||||
|
||||
var tch <-chan time.Time
|
||||
var timer *time.Timer
|
||||
if !dl.IsZero() {
|
||||
d := time.Until(dl)
|
||||
if d <= 0 {
|
||||
return nil, os.ErrDeadlineExceeded
|
||||
}
|
||||
timer = time.NewTimer(d)
|
||||
tch = timer.C
|
||||
}
|
||||
|
||||
select {
|
||||
case chunk := <-c.chunks:
|
||||
if timer != nil {
|
||||
timer.Stop()
|
||||
}
|
||||
return chunk.GetData(), nil
|
||||
case <-c.closed:
|
||||
if timer != nil {
|
||||
timer.Stop()
|
||||
}
|
||||
return nil, net.ErrClosed
|
||||
case <-c.done:
|
||||
if timer != nil {
|
||||
timer.Stop()
|
||||
}
|
||||
select {
|
||||
case chunk := <-c.chunks:
|
||||
return chunk.GetData(), nil
|
||||
default:
|
||||
}
|
||||
return nil, c.termErr()
|
||||
case <-tch:
|
||||
return nil, os.ErrDeadlineExceeded
|
||||
case <-wake:
|
||||
if timer != nil {
|
||||
timer.Stop()
|
||||
}
|
||||
// The read deadline changed while parked: re-evaluate it.
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Conn) termErr() error {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
if c.err != nil {
|
||||
return c.err
|
||||
}
|
||||
return net.ErrClosed
|
||||
}
|
||||
|
||||
func (c *Conn) Read(b []byte) (int, error) {
|
||||
if len(b) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
if c.framed {
|
||||
p, err := c.readDatagram()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
// Datagram semantics: bytes beyond b are discarded, as on a UDP socket.
|
||||
return copy(b, p), nil
|
||||
}
|
||||
for {
|
||||
c.mu.Lock()
|
||||
if len(c.rb) > 0 {
|
||||
n := copy(b, c.rb)
|
||||
c.rb = c.rb[n:]
|
||||
c.mu.Unlock()
|
||||
return n, nil
|
||||
}
|
||||
c.mu.Unlock()
|
||||
|
||||
data, err := c.nextChunk()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
c.mu.Lock()
|
||||
c.rb = data
|
||||
c.mu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
// readDatagram assembles one complete frame across chunks. c.rb only ever
|
||||
// accumulates up to MaxFrame+one chunk: a frame header bounds the wait.
|
||||
func (c *Conn) readDatagram() ([]byte, error) {
|
||||
for {
|
||||
c.mu.Lock()
|
||||
rb := c.rb
|
||||
if len(rb) >= 2 {
|
||||
n := int(binary.BigEndian.Uint16(rb))
|
||||
if len(rb) >= 2+n {
|
||||
p := rb[2 : 2+n]
|
||||
c.rb = c.rb[2+n:]
|
||||
c.mu.Unlock()
|
||||
return p, nil
|
||||
}
|
||||
}
|
||||
c.mu.Unlock()
|
||||
|
||||
data, err := c.nextChunk()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
c.mu.Lock()
|
||||
c.rb = append(c.rb, data...)
|
||||
c.mu.Unlock()
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Conn) Write(b []byte) (int, error) {
|
||||
select {
|
||||
case <-c.closed:
|
||||
return 0, net.ErrClosed
|
||||
default:
|
||||
}
|
||||
|
||||
if c.framed {
|
||||
if len(b) > MaxFrame {
|
||||
return 0, errors.New("streamconn: datagram exceeds MaxFrame")
|
||||
}
|
||||
var buf bytes.Buffer
|
||||
if err := WriteFrame(&buf, b); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if err := c.send(buf.Bytes()); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return len(b), nil // datagram: the payload length, not the frame
|
||||
}
|
||||
|
||||
if err := c.send(b); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return len(b), nil
|
||||
}
|
||||
|
||||
func (c *Conn) send(b []byte) error {
|
||||
c.sendMu.Lock()
|
||||
defer c.sendMu.Unlock()
|
||||
for len(b) > 0 {
|
||||
n := min(len(b), writeChunkSize)
|
||||
if err := c.sendChunk(b[:n]); err != nil {
|
||||
return err
|
||||
}
|
||||
b = b[n:]
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Conn) sendChunk(p []byte) error {
|
||||
// A write deadline parks the Send in flow control; only a stream cancel
|
||||
// unblocks it, so the watchdog requires an abort func.
|
||||
if c.abort != nil {
|
||||
if dl := c.writeDeadline(); !dl.IsZero() {
|
||||
d := time.Until(dl)
|
||||
if d <= 0 {
|
||||
return os.ErrDeadlineExceeded
|
||||
}
|
||||
t := time.AfterFunc(d, c.abort)
|
||||
defer t.Stop()
|
||||
}
|
||||
}
|
||||
return c.s.Send(&proto.Chunk{Data: p})
|
||||
}
|
||||
|
||||
// Close marks the conn closed and aborts the stream when possible. It is
|
||||
// idempotent and never blocks on the pump.
|
||||
func (c *Conn) Close() error {
|
||||
c.closeOnce.Do(func() {
|
||||
c.mu.Lock()
|
||||
if c.err == nil {
|
||||
c.err = net.ErrClosed
|
||||
}
|
||||
c.mu.Unlock()
|
||||
close(c.closed)
|
||||
if c.abort != nil {
|
||||
c.abort()
|
||||
}
|
||||
})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Conn) LocalAddr() net.Addr { return c.local }
|
||||
func (c *Conn) RemoteAddr() net.Addr { return c.remote }
|
||||
|
||||
func (c *Conn) SetDeadline(t time.Time) error {
|
||||
if err := c.SetReadDeadline(t); err != nil {
|
||||
return err
|
||||
}
|
||||
return c.SetWriteDeadline(t)
|
||||
}
|
||||
|
||||
func (c *Conn) SetReadDeadline(t time.Time) error {
|
||||
c.mu.Lock()
|
||||
c.rd = t
|
||||
close(c.rdWake) // interrupt a parked Read; net.Conn deadlines apply to in-flight reads
|
||||
c.rdWake = make(chan struct{})
|
||||
c.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Conn) SetWriteDeadline(t time.Time) error {
|
||||
if c.abort == nil {
|
||||
return &net.OpError{Op: "set", Net: "p2p", Err: errors.New("write deadline not supported on a server stream")}
|
||||
}
|
||||
c.mu.Lock()
|
||||
c.wd = t
|
||||
c.mu.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Conn) writeDeadline() time.Time {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return c.wd
|
||||
}
|
||||
@@ -0,0 +1,358 @@
|
||||
package streamconn
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/go-gost/plugin/p2p/proto"
|
||||
)
|
||||
|
||||
// fakeStream is an in-memory Stream: recvCh feeds Recv, sendCh collects Send.
|
||||
type fakeStream struct {
|
||||
ctx context.Context
|
||||
recvCh chan *proto.Chunk
|
||||
sendCh chan *proto.Chunk
|
||||
sends atomic.Int32
|
||||
recvs atomic.Int32
|
||||
}
|
||||
|
||||
func newFakeStream(ctx context.Context) *fakeStream {
|
||||
return &fakeStream{
|
||||
ctx: ctx,
|
||||
recvCh: make(chan *proto.Chunk, 64),
|
||||
sendCh: make(chan *proto.Chunk, 64),
|
||||
}
|
||||
}
|
||||
|
||||
func (f *fakeStream) Send(c *proto.Chunk) error {
|
||||
f.sends.Add(1)
|
||||
f.sendCh <- c
|
||||
return nil
|
||||
}
|
||||
|
||||
func (f *fakeStream) Recv() (*proto.Chunk, error) {
|
||||
f.recvs.Add(1)
|
||||
select {
|
||||
case c := <-f.recvCh:
|
||||
return c, nil
|
||||
case <-f.ctx.Done():
|
||||
return nil, io.EOF
|
||||
}
|
||||
}
|
||||
|
||||
func (f *fakeStream) Context() context.Context { return f.ctx }
|
||||
|
||||
func waitFor(t *testing.T, what string, ok func() bool) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(2 * time.Second)
|
||||
for time.Now().Before(deadline) {
|
||||
if ok() {
|
||||
return
|
||||
}
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("condition not met in time: %s", what)
|
||||
}
|
||||
|
||||
func TestFrameRoundTrip(t *testing.T) {
|
||||
for _, n := range []int{0, 1, 2, 255, 256, 65535} {
|
||||
payload := bytes.Repeat([]byte{0xab}, n)
|
||||
var buf bytes.Buffer
|
||||
if err := WriteFrame(&buf, payload); err != nil {
|
||||
t.Fatalf("n=%d WriteFrame: %v", n, err)
|
||||
}
|
||||
got, err := ReadFrame(&buf)
|
||||
if err != nil {
|
||||
t.Fatalf("n=%d ReadFrame: %v", n, err)
|
||||
}
|
||||
if n == 0 {
|
||||
if got != nil {
|
||||
t.Fatalf("n=0: want nil, got %v", got)
|
||||
}
|
||||
continue
|
||||
}
|
||||
if !bytes.Equal(got, payload) {
|
||||
t.Fatalf("n=%d: payload mismatch", n)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestReadFrameFragmented feeds a frame header + payload split across writes
|
||||
// (and TCP-style arrivals) to prove io.ReadFull reassembly.
|
||||
func TestReadFrameFragmented(t *testing.T) {
|
||||
c1, c2 := net.Pipe()
|
||||
defer c2.Close()
|
||||
go func() {
|
||||
c1.Write([]byte{0x00, 0x03, 'a'})
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
c1.Write([]byte("b"))
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
c1.Write([]byte("c"))
|
||||
c1.Close()
|
||||
}()
|
||||
got, err := ReadFrame(c2)
|
||||
if err != nil {
|
||||
t.Fatalf("ReadFrame: %v", err)
|
||||
}
|
||||
if string(got) != "abc" {
|
||||
t.Fatalf("got %q want %q", got, "abc")
|
||||
}
|
||||
}
|
||||
|
||||
func TestReadDeadline(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
fs := newFakeStream(ctx)
|
||||
c := New(fs, func() {}, "tcp", nil, nil)
|
||||
defer c.Close()
|
||||
|
||||
if err := c.SetReadDeadline(time.Now().Add(50 * time.Millisecond)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
buf := make([]byte, 16)
|
||||
start := time.Now()
|
||||
if _, err := c.Read(buf); !errors.Is(err, os.ErrDeadlineExceeded) {
|
||||
t.Fatalf("Read err = %v, want os.ErrDeadlineExceeded", err)
|
||||
}
|
||||
if elapsed := time.Since(start); elapsed < 30*time.Millisecond {
|
||||
t.Fatalf("Read returned after %v, want the deadline to have applied", elapsed)
|
||||
}
|
||||
|
||||
// Clearing the deadline lets the next Read consume data again.
|
||||
if err := c.SetReadDeadline(time.Time{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
go func() { fs.recvCh <- &proto.Chunk{Data: []byte("ping")} }()
|
||||
if n, err := c.Read(buf); err != nil || string(buf[:n]) != "ping" {
|
||||
t.Fatalf("Read after deadline clear = %q, %v; want ping", buf[:n], err)
|
||||
}
|
||||
}
|
||||
|
||||
// TestBackpressure proves the one-chunk handoff throttles Recv: while the
|
||||
// consumer is idle the pump must not fetch a third chunk.
|
||||
func TestBackpressure(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
fs := newFakeStream(ctx)
|
||||
for i := range 5 {
|
||||
fs.recvCh <- &proto.Chunk{Data: []byte{byte(i)}}
|
||||
}
|
||||
c := New(fs, func() {}, "tcp", nil, nil)
|
||||
defer c.Close()
|
||||
|
||||
// Pump: fetch 1 -> deliver, fetch 2 -> blocked sending. No fetch 3.
|
||||
waitFor(t, "pump fetched two chunks", func() bool { return fs.recvs.Load() == 2 })
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
if n := fs.recvs.Load(); n != 2 {
|
||||
t.Fatalf("Recv calls = %d while consumer idle, want 2 (backpressure)", n)
|
||||
}
|
||||
|
||||
// One Read frees the slot; the pump advances by exactly one fetch.
|
||||
buf := make([]byte, 8)
|
||||
if n, err := c.Read(buf); err != nil || n != 1 || buf[0] != 0 {
|
||||
t.Fatalf("Read = %d, %v, %v; want the first chunk", n, buf[:n], err)
|
||||
}
|
||||
waitFor(t, "pump advanced by one", func() bool { return fs.recvs.Load() == 3 })
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
if n := fs.recvs.Load(); n != 3 {
|
||||
t.Fatalf("Recv calls = %d after one Read, want 3 (one slot freed, one fetch)", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFramedDatagrams(t *testing.T) {
|
||||
ctxA, cancelA := context.WithCancel(context.Background())
|
||||
defer cancelA()
|
||||
ctxB, cancelB := context.WithCancel(context.Background())
|
||||
defer cancelB()
|
||||
a2b := make(chan *proto.Chunk, 64)
|
||||
b2a := make(chan *proto.Chunk, 64)
|
||||
connA := New(&fakeStream{ctx: ctxA, recvCh: b2a, sendCh: a2b}, cancelA, "udp", nil, nil)
|
||||
connB := New(&fakeStream{ctx: ctxB, recvCh: a2b, sendCh: b2a}, cancelB, "udp", nil, nil)
|
||||
defer connA.Close()
|
||||
defer connB.Close()
|
||||
|
||||
// One datagram per Write/Read, boundaries preserved.
|
||||
connA.Write([]byte("hello"))
|
||||
buf := make([]byte, 64)
|
||||
if n, err := connB.Read(buf); err != nil || string(buf[:n]) != "hello" {
|
||||
t.Fatalf("Read = %q, %v; want hello", buf[:n], err)
|
||||
}
|
||||
connA.Write([]byte("a"))
|
||||
connA.Write([]byte("bb"))
|
||||
if n, err := connB.Read(buf); err != nil || n != 1 || buf[0] != 'a' {
|
||||
t.Fatalf("Read = %q, %v; want single-byte datagram", buf[:n], err)
|
||||
}
|
||||
if n, err := connB.Read(buf); err != nil || n != 2 || string(buf[:n]) != "bb" {
|
||||
t.Fatalf("Read = %q, %v; want bb (no coalescing)", buf[:n], err)
|
||||
}
|
||||
|
||||
// An empty datagram reads as 0 bytes, not EOF.
|
||||
connA.Write(nil)
|
||||
if n, err := connB.Read(buf); err != nil || n != 0 {
|
||||
t.Fatalf("Read = %d, %v; want empty datagram (0, nil)", n, err)
|
||||
}
|
||||
|
||||
// Oversized relative to the read buffer: truncated, remainder dropped,
|
||||
// and the next Read gets the next datagram — not the leftover bytes.
|
||||
connA.Write(bytes.Repeat([]byte{7}, 10))
|
||||
if n, err := connB.Read(make([]byte, 4)); err != nil || n != 4 {
|
||||
t.Fatalf("Read = %d, %v; want truncation to 4", n, err)
|
||||
}
|
||||
connA.Write([]byte("z"))
|
||||
if n, err := connB.Read(buf); err != nil || string(buf[:n]) != "z" {
|
||||
t.Fatalf("Read = %q, %v; want z (remainder dropped)", buf[:n], err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWriteSplitsChunks(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
fs := newFakeStream(ctx)
|
||||
c := New(fs, func() {}, "tcp", nil, nil)
|
||||
defer c.Close()
|
||||
|
||||
payload := bytes.Repeat([]byte{0x5a}, writeChunkSize+16)
|
||||
n, err := c.Write(payload)
|
||||
if err != nil || n != len(payload) {
|
||||
t.Fatalf("Write = %d, %v; want %d", n, err, len(payload))
|
||||
}
|
||||
c1 := <-fs.sendCh
|
||||
c2 := <-fs.sendCh
|
||||
if len(c1.Data) != writeChunkSize || len(c2.Data) != 16 {
|
||||
t.Fatalf("chunk sizes = %d, %d; want %d, 16", len(c1.Data), len(c2.Data), writeChunkSize)
|
||||
}
|
||||
if !bytes.Equal(append(c1.Data, c2.Data...), payload) {
|
||||
t.Fatal("split payload mismatch")
|
||||
}
|
||||
select {
|
||||
case extra := <-fs.sendCh:
|
||||
t.Fatalf("unexpected third Send: %d bytes", len(extra.Data))
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// TestPumpCancelUnblocksRead: when the transport dies while the pump is parked
|
||||
// on the handoff (no Close, no Recv error yet — e.g. the client aborts a
|
||||
// server stream), the reader must still get a terminal error. A pump that
|
||||
// exited silently here would hang the host-side pipe forever.
|
||||
func TestPumpCancelUnblocksRead(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
fs := newFakeStream(ctx)
|
||||
fs.recvCh <- &proto.Chunk{Data: []byte{1}}
|
||||
fs.recvCh <- &proto.Chunk{Data: []byte{2}}
|
||||
c := New(fs, func() {}, "tcp", nil, nil)
|
||||
defer c.Close()
|
||||
|
||||
waitFor(t, "pump parked on the handoff", func() bool { return fs.recvs.Load() == 2 })
|
||||
cancel()
|
||||
|
||||
buf := make([]byte, 4)
|
||||
if n, err := c.Read(buf); err != nil || n != 1 {
|
||||
t.Fatalf("Read = %d, %v; want the buffered chunk first", n, err)
|
||||
}
|
||||
c.SetReadDeadline(time.Now().Add(2 * time.Second)) // hang guard
|
||||
if _, err := c.Read(buf); err == nil {
|
||||
t.Fatal("Read after the pump died = nil error, want the terminal error")
|
||||
}
|
||||
}
|
||||
|
||||
// TestTerminalEOFKeepsBufferedChunk: a chunk already delivered when the
|
||||
// terminal state arrives (final chunk + clean EOF together) must still be
|
||||
// readable — the select must prefer the chunk, not drop it for the error.
|
||||
func TestTerminalEOFKeepsBufferedChunk(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
fs := newFakeStream(ctx)
|
||||
fs.recvCh <- &proto.Chunk{Data: []byte("last")}
|
||||
c := New(fs, func() {}, "tcp", nil, nil)
|
||||
defer c.Close()
|
||||
|
||||
waitFor(t, "chunk delivered then parked", func() bool { return fs.recvs.Load() == 2 })
|
||||
cancel() // Recv reports EOF: the chunk and the terminal state are both pending
|
||||
|
||||
buf := make([]byte, 8)
|
||||
if n, err := c.Read(buf); err != nil || string(buf[:n]) != "last" {
|
||||
t.Fatalf("Read = %q, %v; want the buffered last chunk", buf[:n], err)
|
||||
}
|
||||
c.SetReadDeadline(time.Now().Add(2 * time.Second)) // hang guard
|
||||
if _, err := c.Read(buf); err == nil {
|
||||
t.Fatal("Read after the last chunk = nil error, want io.EOF")
|
||||
}
|
||||
}
|
||||
|
||||
// TestReadDeadlineUpdateWakes covers the net.Conn contract that a deadline set
|
||||
// while a Read is blocked applies to that Read (the usual interrupt idiom).
|
||||
func TestReadDeadlineUpdateWakes(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
c := New(newFakeStream(ctx), func() {}, "tcp", nil, nil)
|
||||
defer c.Close()
|
||||
|
||||
errCh := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := c.Read(make([]byte, 4))
|
||||
errCh <- err
|
||||
}()
|
||||
time.Sleep(30 * time.Millisecond) // let the Read park
|
||||
c.SetReadDeadline(time.Now())
|
||||
select {
|
||||
case err := <-errCh:
|
||||
if !errors.Is(err, os.ErrDeadlineExceeded) {
|
||||
t.Fatalf("Read err = %v, want os.ErrDeadlineExceeded", err)
|
||||
}
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("SetReadDeadline did not wake the parked Read")
|
||||
}
|
||||
}
|
||||
|
||||
func TestCloseUnblocksRead(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
c := New(newFakeStream(ctx), func() {}, "tcp", nil, nil)
|
||||
|
||||
go func() {
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
c.Close()
|
||||
}()
|
||||
if _, err := c.Read(make([]byte, 4)); !errors.Is(err, net.ErrClosed) {
|
||||
t.Fatalf("Read err = %v, want net.ErrClosed", err)
|
||||
}
|
||||
// Idempotent close.
|
||||
if err := c.Close(); err != nil {
|
||||
t.Fatalf("second Close = %v, want nil", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWriteDeadlineSupport(t *testing.T) {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
|
||||
// Server side (no abort): a parked Send cannot be interrupted, so write
|
||||
// deadlines are rejected.
|
||||
c := New(newFakeStream(ctx), nil, "tcp", nil, nil)
|
||||
defer c.Close()
|
||||
if err := c.SetWriteDeadline(time.Now().Add(time.Second)); err == nil {
|
||||
t.Fatal("SetWriteDeadline on a server stream = nil, want not-supported error")
|
||||
}
|
||||
if err := c.SetDeadline(time.Now().Add(time.Second)); err == nil {
|
||||
t.Fatal("SetDeadline on a server stream = nil, want the write side surfaced")
|
||||
}
|
||||
|
||||
// Client side: accepted; an expired deadline fails the write fast.
|
||||
ca := New(newFakeStream(ctx), func() {}, "tcp", nil, nil)
|
||||
defer ca.Close()
|
||||
if err := ca.SetWriteDeadline(time.Now().Add(-time.Millisecond)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := ca.Write([]byte("x")); !errors.Is(err, os.ErrDeadlineExceeded) {
|
||||
t.Fatalf("Write err = %v, want os.ErrDeadlineExceeded", err)
|
||||
}
|
||||
}
|
||||
+36
-90
@@ -1,31 +1,26 @@
|
||||
// Package p2p adapts any dialer to run on top of a tunnel supplied by a p2p
|
||||
// plugin. The inner dialer's base connection (options.Dialer) is replaced
|
||||
// with "open a tunnel to the peer, dial the returned local endpoint". The
|
||||
// peer-facing protocol stays owned by the inner dialer — p2p only supplies
|
||||
// the tunnel base.
|
||||
// with the tunnel's data stream — a net.Conn carried over the GOST↔host gRPC
|
||||
// connection. The peer-facing protocol stays owned by the inner dialer — p2p
|
||||
// only supplies the tunnel base.
|
||||
package p2p
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/go-gost/core/dialer"
|
||||
"github.com/go-gost/core/metadata"
|
||||
)
|
||||
|
||||
// rpcTimeout caps each control RPC (OpenTunnel/CloseTunnel) with a fresh
|
||||
// context so a wedged plugin cannot stall a dial or leak a finalizer.
|
||||
const rpcTimeout = 3 * time.Second
|
||||
|
||||
// TunnelProvider opens and closes tunnels to a peer. It is the wrapper's
|
||||
// only contact with the p2p plugin; peer and endpoint are opaque strings.
|
||||
// network is "tcp" or "udp": it selects the endpoint's semantics, so a
|
||||
// datagram dialer gets a datagram endpoint instead of a byte stream.
|
||||
// TunnelProvider opens tunnels to a peer over the plugin control protocol.
|
||||
// It is the wrapper's only contact with the p2p plugin; peer is an opaque
|
||||
// string (the plugin defines its semantics). network is "tcp" or "udp" and
|
||||
// selects the data stream's semantics: "udp" preserves datagram boundaries,
|
||||
// so a datagram dialer gets a datagram conn instead of a byte stream.
|
||||
type TunnelProvider interface {
|
||||
OpenTunnel(ctx context.Context, network, peer string) (id, endpoint string, err error)
|
||||
CloseTunnel(ctx context.Context, id string) error
|
||||
OpenTunnelStream(ctx context.Context, network, peer string) (net.Conn, error)
|
||||
Close() error
|
||||
}
|
||||
|
||||
@@ -44,7 +39,7 @@ func NewTunnelDialer(inner dialer.Dialer, pr TunnelProvider) dialer.Dialer {
|
||||
// protocol on top of a p2p tunnel. The wrapper's implicit contract is
|
||||
// "inner dials one plain conn via options.Dialer and uses it as its base" —
|
||||
// a byte stream for the stream dialers (mux inners dial it once per session
|
||||
// and multiplex streams over it), a datagram endpoint for udp. kcp/quic
|
||||
// and multiplex streams over it), a datagram stream for udp. kcp/quic
|
||||
// bases, http2 probes and destroys tunnels. Fail closed: new dialers are
|
||||
// unsupported until verified.
|
||||
func SupportedDialer(name string) bool {
|
||||
@@ -57,7 +52,7 @@ func SupportedDialer(name string) bool {
|
||||
}
|
||||
|
||||
// tunnelNetwork normalizes a dialer network to the two the tunnel protocol
|
||||
// defines, so a udp4/udp6 dialer still asks for a datagram endpoint.
|
||||
// defines, so a udp4/udp6 dialer still asks for a datagram stream.
|
||||
func tunnelNetwork(network string) string {
|
||||
switch network {
|
||||
case "udp", "udp4", "udp6":
|
||||
@@ -86,12 +81,10 @@ func (d *tunnelDialer) Dial(ctx context.Context, addr string, opts ...dialer.Dia
|
||||
base := &tunnelBaseDialer{peer: addr, pr: d.provider}
|
||||
conn, err := d.inner.Dial(ctx, addr, append(opts, dialer.NetDialerDialOption(base))...)
|
||||
if err != nil {
|
||||
// The tunnel is unusable: close it and fail closed.
|
||||
if id := base.tunnelID(); id != "" {
|
||||
cctx, cancel := context.WithTimeout(context.Background(), rpcTimeout)
|
||||
defer cancel()
|
||||
d.provider.CloseTunnel(cctx, id)
|
||||
}
|
||||
// The tunnel is unusable: close it and fail closed. closeTunnel is
|
||||
// nil-safe — OpenTunnelStream may have failed before any conn existed
|
||||
// (the host's pending GC then reclaims the record).
|
||||
base.closeTunnel()
|
||||
return nil, err
|
||||
}
|
||||
return conn, nil
|
||||
@@ -116,89 +109,42 @@ func (d *tunnelDialer) Multiplex() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// tunnelBaseDialer lazily opens a tunnel on first use, then dials the tunnel
|
||||
// endpoint and wraps the resulting conn with tunnel teardown. If the inner
|
||||
// dialer never dials its base (mux session cache hit), no tunnel is opened.
|
||||
// tunnelBaseDialer lazily opens a tunnel on first use and returns its data
|
||||
// conn as the inner dialer's base. The conn's Close IS the tunnel teardown:
|
||||
// the stream ending is the tunnel ending. If the inner dialer never dials its
|
||||
// base (mux session cache hit), no tunnel is opened.
|
||||
type tunnelBaseDialer struct {
|
||||
peer string
|
||||
pr TunnelProvider
|
||||
|
||||
mu sync.Mutex
|
||||
id string
|
||||
endpoint string
|
||||
mu sync.Mutex
|
||||
conn net.Conn
|
||||
}
|
||||
|
||||
// tunnelID returns the opened tunnel's id, or "" if none was opened.
|
||||
func (d *tunnelBaseDialer) tunnelID() string {
|
||||
// closeTunnel closes the opened conn, if any. Nil-safe: OpenTunnelStream can
|
||||
// fail before any conn exists, and closing a nil net.Conn interface panics.
|
||||
func (d *tunnelBaseDialer) closeTunnel() {
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
return d.id
|
||||
conn := d.conn
|
||||
d.conn = nil
|
||||
d.mu.Unlock()
|
||||
if conn != nil {
|
||||
conn.Close()
|
||||
}
|
||||
}
|
||||
|
||||
func (d *tunnelBaseDialer) Dial(ctx context.Context, network, addr string) (net.Conn, error) {
|
||||
// Only the control RPC is time-capped; the endpoint dial runs on the
|
||||
// caller's ctx so a slow path to the peer isn't killed by this budget.
|
||||
// Only the control RPC is time-capped (inside OpenTunnelStream); the
|
||||
// stream runs on its own context so the conn outlives this dial.
|
||||
network = tunnelNetwork(network)
|
||||
d.mu.Lock()
|
||||
defer d.mu.Unlock()
|
||||
if d.endpoint == "" {
|
||||
octx, cancel := context.WithTimeout(ctx, rpcTimeout)
|
||||
defer cancel()
|
||||
id, endpoint, err := d.pr.OpenTunnel(octx, network, d.peer)
|
||||
if d.conn == nil {
|
||||
conn, err := d.pr.OpenTunnelStream(ctx, network, d.peer)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
d.id, d.endpoint = id, endpoint
|
||||
d.conn = conn
|
||||
}
|
||||
conn, err := (&net.Dialer{}).DialContext(ctx, network, d.endpoint)
|
||||
if err != nil {
|
||||
// The tunnel is unusable: release it and forget it, so a retry
|
||||
// on the same base dialer opens a fresh one.
|
||||
cctx, cancel := context.WithTimeout(context.Background(), rpcTimeout)
|
||||
defer cancel()
|
||||
d.pr.CloseTunnel(cctx, d.id)
|
||||
d.id, d.endpoint = "", ""
|
||||
return nil, err
|
||||
}
|
||||
if network == "udp" {
|
||||
// Announce the client address with an empty datagram. A datagram
|
||||
// endpoint learns where to write peer packets only from a packet it
|
||||
// receives, so without this a tunnel whose peer speaks first would
|
||||
// drop every packet until this side happened to send one.
|
||||
conn.Write(nil)
|
||||
}
|
||||
return &tunnelConn{Conn: conn, id: d.id, pr: d.pr, onClosed: func() {
|
||||
// The inner may close the conn before this base dialer ever sees an
|
||||
// error; clear the state so tunnelDialer's fail path doesn't issue a
|
||||
// second CloseTunnel.
|
||||
d.mu.Lock()
|
||||
d.id, d.endpoint = "", ""
|
||||
d.mu.Unlock()
|
||||
}}, nil
|
||||
return d.conn, nil
|
||||
}
|
||||
|
||||
// tunnelConn is the tunnel handle: closing it closes the underlying conn and
|
||||
// releases the tunnel (CloseTunnel) exactly once.
|
||||
type tunnelConn struct {
|
||||
net.Conn
|
||||
id string
|
||||
pr TunnelProvider
|
||||
onClosed func()
|
||||
closeOne sync.Once
|
||||
}
|
||||
|
||||
func (c *tunnelConn) Close() error {
|
||||
var err error
|
||||
c.closeOne.Do(func() {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), rpcTimeout)
|
||||
defer cancel()
|
||||
err = c.pr.CloseTunnel(ctx, c.id)
|
||||
if cerr := c.Conn.Close(); err == nil {
|
||||
err = cerr
|
||||
}
|
||||
if c.onClosed != nil {
|
||||
c.onClosed()
|
||||
}
|
||||
})
|
||||
return err
|
||||
}
|
||||
Reference in New Issue
Block a user