mirror of
https://github.com/fatedier/frp.git
synced 2026-10-23 16:33:27 +08:00
fix(client): use generation-local channels in UDPProxy.InWorkConn (#5585)
InWorkConn published readCh and sendCh into pxy fields while holding pxy.mu, then read those same fields again after unlocking to start the sender, reader and heartbeat goroutines and to call udp.Forwarder. Two work connections installed concurrently made that unlocked read race with the next generation's write to the same fields. Create the channels as locals, publish them under the existing lock, and use only those locals for the goroutines and the forwarder. The fields stay guarded by pxy.mu, so Close still closes the current generation. No behavior change for a single generation. Add TestUDPProxyInWorkConnGenerationChannels, which drives two InWorkConn calls concurrently so the race detector covers the field accesses.
This commit is contained in:
+8
-6
@@ -108,8 +108,10 @@ func (pxy *UDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
|
|||||||
|
|
||||||
pxy.mu.Lock()
|
pxy.mu.Lock()
|
||||||
pxy.workConn = workConn
|
pxy.workConn = workConn
|
||||||
pxy.readCh = make(chan *msg.UDPPacket, 1024)
|
readCh := make(chan *msg.UDPPacket, 1024)
|
||||||
pxy.sendCh = make(chan msg.Message, 1024)
|
sendCh := make(chan msg.Message, 1024)
|
||||||
|
pxy.readCh = readCh
|
||||||
|
pxy.sendCh = sendCh
|
||||||
pxy.closed = false
|
pxy.closed = false
|
||||||
pxy.mu.Unlock()
|
pxy.mu.Unlock()
|
||||||
|
|
||||||
@@ -160,10 +162,10 @@ func (pxy *UDPProxy) InWorkConn(conn net.Conn, _ *msg.StartWorkConn) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
go workConnSenderFn(payloadRW, pxy.sendCh)
|
go workConnSenderFn(payloadRW, sendCh)
|
||||||
go workConnReaderFn(payloadRW, pxy.readCh)
|
go workConnReaderFn(payloadRW, readCh)
|
||||||
go heartbeatFn(pxy.sendCh)
|
go heartbeatFn(sendCh)
|
||||||
|
|
||||||
// Call Forwarder with proxy protocol version (empty string means no proxy protocol)
|
// Call Forwarder with proxy protocol version (empty string means no proxy protocol)
|
||||||
udp.Forwarder(pxy.localAddr, pxy.readCh, pxy.sendCh, int(pxy.clientCfg.UDPPacketSize), pxy.cfg.Transport.ProxyProtocolVersion)
|
udp.Forwarder(pxy.localAddr, readCh, sendCh, int(pxy.clientCfg.UDPPacketSize), pxy.cfg.Transport.ProxyProtocolVersion)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,63 @@
|
|||||||
|
// Copyright 2026 The frp Authors
|
||||||
|
//
|
||||||
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
// you may not use this file except in compliance with the License.
|
||||||
|
// You may obtain a copy of the License at
|
||||||
|
//
|
||||||
|
// http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
//
|
||||||
|
// Unless required by applicable law or agreed to in writing, software
|
||||||
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
// See the License for the specific language governing permissions and
|
||||||
|
// limitations under the License.
|
||||||
|
|
||||||
|
//go:build !frps
|
||||||
|
|
||||||
|
package proxy
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"net"
|
||||||
|
"sync"
|
||||||
|
"testing"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
|
||||||
|
v1 "github.com/fatedier/frp/pkg/config/v1"
|
||||||
|
"github.com/fatedier/frp/pkg/msg"
|
||||||
|
"github.com/fatedier/frp/pkg/util/xlog"
|
||||||
|
)
|
||||||
|
|
||||||
|
func TestUDPProxyInWorkConnGenerationChannels(t *testing.T) {
|
||||||
|
cfg := &v1.UDPProxyConfig{ProxyBaseConfig: v1.ProxyBaseConfig{
|
||||||
|
Name: "udp", Type: "udp",
|
||||||
|
ProxyBackend: v1.ProxyBackend{LocalIP: "127.0.0.1", LocalPort: 9},
|
||||||
|
}}
|
||||||
|
pxy := NewUDPProxy(&BaseProxy{
|
||||||
|
baseCfg: &cfg.ProxyBaseConfig, clientCfg: &v1.ClientCommonConfig{UDPPacketSize: 1500},
|
||||||
|
xl: xlog.New(), ctx: context.Background(),
|
||||||
|
}, cfg).(*UDPProxy)
|
||||||
|
require.NoError(t, pxy.Run())
|
||||||
|
defer pxy.Close()
|
||||||
|
|
||||||
|
conn1, peer1 := net.Pipe()
|
||||||
|
defer conn1.Close()
|
||||||
|
defer peer1.Close()
|
||||||
|
conn2, peer2 := net.Pipe()
|
||||||
|
defer conn2.Close()
|
||||||
|
defer peer2.Close()
|
||||||
|
|
||||||
|
start := make(chan struct{})
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
wg.Add(2)
|
||||||
|
inWorkConn := func(conn net.Conn) {
|
||||||
|
defer wg.Done()
|
||||||
|
<-start
|
||||||
|
pxy.InWorkConn(conn, &msg.StartWorkConn{})
|
||||||
|
}
|
||||||
|
go inWorkConn(conn1)
|
||||||
|
go inWorkConn(conn2)
|
||||||
|
close(start)
|
||||||
|
wg.Wait()
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user