Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -11,5 +11,8 @@
# Output of the go coverage tool, specifically when used with LiteIDE
*.out

# Log files
*.log

# Dependency directories (remove the comment below to include it)
# vendor/
28 changes: 14 additions & 14 deletions main.go
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,13 @@ func main() {

if *t == "proxy_client" ||
*t == "reverse_proxy_client" {
if !(len(fromaddr) == len(toaddr) && len(fromaddr) == len(proxyproto)) {
fmt.Println("[fromaddr] [toaddr] [proxyproto] len must be equal")
fmt.Println()
flag.Usage()
return
}

for i, _ := range proxyproto {
if len(fromaddr[i]) == 0 || len(*server) == 0 || len(toaddr[i]) == 0 {
fmt.Println("[proxy_client] or [reverse_proxy_client] need [server] [fromaddr] [toaddr] [proxyproto]")
Expand All @@ -139,20 +146,20 @@ func main() {
}
}

if !(len(fromaddr) == len(toaddr) && len(fromaddr) == len(proxyproto)) {
fmt.Println("[fromaddr] [toaddr] [proxyproto] len must be equal")
fmt.Println()
flag.Usage()
return
}

if len(protos) == 0 {
protos = append(protos, "tcp")
}
}

if *t == "socks5_client" ||
*t == "reverse_socks5_client" {
if !(len(fromaddr) == len(proxyproto)) {
fmt.Println("[fromaddr] [proxyproto] len must be equal")
fmt.Println()
flag.Usage()
return
}

for i, _ := range proxyproto {
if len(fromaddr[i]) == 0 || len(*server) == 0 {
fmt.Println("[socks5_client] or [reverse_socks5_client] need [server] [fromaddr] [proxyproto]")
Expand All @@ -162,13 +169,6 @@ func main() {
}
}

if !(len(fromaddr) == len(proxyproto)) {
fmt.Println("[fromaddr] [proxyproto] len must be equal")
fmt.Println()
flag.Usage()
return
}

if len(protos) == 0 {
protos = append(protos, "tcp")
}
Expand Down
3 changes: 2 additions & 1 deletion proxy/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@ func NewClient(config *Config, serverproto string, server string, name string, c
proxyproto = append(proxyproto, PROXY_PROTO(p))
}

wg := thread.NewGroup("Clent"+" "+clienttypestr, nil, nil)
wg := thread.NewGroup("Client"+" "+clienttypestr, nil, nil)

c := &Client{
config: config,
Expand Down Expand Up @@ -254,6 +254,7 @@ func (c *Client) processLoginRsp(wg *thread.Group, index int, f *ProxyFrame, sen

err := c.iniService(wg, index, serverconn)
if err != nil {
serverconn.needclose = true
loggo.Error("processLoginRsp iniService fail %s %s", c.server, err)
return
}
Expand Down
38 changes: 19 additions & 19 deletions proxy/common.go
Original file line number Diff line number Diff line change
Expand Up @@ -65,8 +65,8 @@ type ProxyConn struct {
established bool
sendch *common.Channel // *ProxyFrame
recvch *common.Channel // *ProxyFrame
actived int
pinged int
actived int32
pinged int32
id string
needclose bool
}
Expand Down Expand Up @@ -213,7 +213,7 @@ func recvFrom(wg *thread.Group, recvch *common.Channel, conn network.Conn, maxms
}

msglen := binary.LittleEndian.Uint32(bs)
if msglen > uint32(maxmsgsize)+MAX_PROTO_PACK_SIZE || msglen <= 0 {
if msglen > uint32(maxmsgsize)+MAX_PROTO_PACK_SIZE || msglen == 0 {
loggo.Error("recvFrom len fail: %s %d", conn.Info(), msglen)
return errors.New("msg len fail " + strconv.Itoa(int(msglen)))
}
Expand All @@ -233,11 +233,6 @@ func recvFrom(wg *thread.Group, recvch *common.Channel, conn network.Conn, maxms
return err
}

if loggo.IsDebug() {
loggo.Debug("recvFrom start Write %s", conn.Info())
}
recvch.Write(f)

if f.Type != FRAME_TYPE_PING && f.Type != FRAME_TYPE_PONG && loggo.IsDebug() {
loggo.Debug("recvFrom %s %s", conn.Info(), f.Type.String())
if f.Type == FRAME_TYPE_DATA {
Expand All @@ -248,6 +243,11 @@ func recvFrom(wg *thread.Group, recvch *common.Channel, conn network.Conn, maxms
}
}

if loggo.IsDebug() {
loggo.Debug("recvFrom start Write %s", conn.Info())
}
recvch.Write(f)

atomic.AddInt32(&gState.MainRecvNum, 1)
atomic.AddInt64(&gState.MainRecvSize, int64(msglen)+4)
}
Expand All @@ -266,14 +266,14 @@ func sendTo(wg *thread.Group, sendch *common.Channel, conn network.Conn, compres

for !wg.IsExit() {
var f *ProxyFrame
if *pingflag > 0 {
*pingflag = 0
if atomic.LoadInt32(pingflag) > 0 {
atomic.StoreInt32(pingflag, 0)
f = &ProxyFrame{}
f.Type = FRAME_TYPE_PING
f.PingFrame = &PingFrame{}
f.PingFrame.Time = time.Now().UnixNano()
} else if *pongflag > 0 {
*pongflag = 0
} else if atomic.LoadInt32(pongflag) > 0 {
atomic.StoreInt32(pongflag, 0)
Comment on lines +269 to +276
Copy link

Copilot AI Apr 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

pingflag / pongflag clearing uses atomic.LoadInt32(...) > 0 followed by atomic.StoreInt32(..., 0). If another goroutine increments the flag between the load and store, that increment can be lost, causing missed PING/PONG sends. Consider using atomic.SwapInt32(flag, 0) (or a CAS loop) so you clear the flag and observe the previous value atomically.

Copilot uses AI. Check for mistakes.
f = &ProxyFrame{}
f.Type = FRAME_TYPE_PONG
f.PongFrame = &PongFrame{}
Expand Down Expand Up @@ -304,7 +304,7 @@ func sendTo(wg *thread.Group, sendch *common.Channel, conn network.Conn, compres
}

msglen := uint32(len(mb))
if msglen > uint32(maxmsgsize)+MAX_PROTO_PACK_SIZE || msglen <= 0 {
if msglen > uint32(maxmsgsize)+MAX_PROTO_PACK_SIZE || msglen == 0 {
loggo.Error("sendTo len fail: %s %d", conn.Info(), msglen)
return errors.New("msg len fail " + strconv.Itoa(int(msglen)))
}
Expand Down Expand Up @@ -501,14 +501,14 @@ func checkPingActive(wg *thread.Group, sendch *common.Channel, recvch *common.Ch
// 2. 定时触发 Ping 逻辑
case <-pingTicker.C:
// 检查心跳超时逻辑
if proxyconn.pinged > pingintertimeout {
if atomic.LoadInt32(&proxyconn.pinged) > int32(pingintertimeout) {
loggo.Info("checkPingActive ping pong timeout %s", proxyconn.conn.Info())
return errors.New("ping pong timeout")
}

// 发送心跳逻辑
atomic.AddInt32(pingflag, 1)
proxyconn.pinged++
atomic.AddInt32(&proxyconn.pinged, 1)
if showping {
loggo.Info("ping %s", proxyconn.conn.Info())
}
Expand Down Expand Up @@ -556,7 +556,7 @@ func processPing(f *ProxyFrame, sendch *common.Channel, proxyconn *ProxyConn, po

func processPong(f *ProxyFrame, sendch *common.Channel, proxyconn *ProxyConn, showping bool) {
elapse := time.Duration(time.Now().UnixNano() - f.PongFrame.Time)
proxyconn.pinged = 0
atomic.StoreInt32(&proxyconn.pinged, 0)
if showping {
loggo.Info("pong %s %s", proxyconn.conn.Info(), elapse.String())
}
Expand Down Expand Up @@ -597,11 +597,11 @@ func checkSonnyActive(wg *thread.Group, proxyconn *ProxyConn, estimeout int, tim

// 2. 定时触发 Ping 逻辑
case <-activedTicker.C:
if proxyconn.actived == 0 {
if atomic.LoadInt32(&proxyconn.actived) == 0 {
loggo.Error("checkSonnyActive timeout %s", proxyconn.conn.Info())
return errors.New("conn timeout")
}
proxyconn.actived = 0
atomic.StoreInt32(&proxyconn.actived, 0)
}
}

Expand Down Expand Up @@ -633,7 +633,7 @@ func copySonnyRecv(wg *thread.Group, recvch *common.Channel, proxyConn *ProxyCon
}
}
f.DataFrame.Id = proxyConn.id
proxyConn.actived++
atomic.AddInt32(&proxyConn.actived, 1)

father.sendch.Write(f)

Expand Down
2 changes: 1 addition & 1 deletion proxy/inputer.go
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ func (i *Inputer) processDataFrame(f *ProxyFrame) {
sonny.needclose = true
loggo.Error("Inputer processDataFrame timeout sonnny %s %d", f.DataFrame.Id, len(f.DataFrame.Data))
}
sonny.actived++
atomic.AddInt32(&sonny.actived, 1)
loggo.Debug("Inputer processDataFrame %s %d", f.DataFrame.Id, len(f.DataFrame.Data))
}

Expand Down
2 changes: 1 addition & 1 deletion proxy/outputer.go
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ func (o *Outputer) processDataFrame(f *ProxyFrame) {
sonny.needclose = true
loggo.Error("Outputer processDataFrame timeout sonnny %s %d", f.DataFrame.Id, len(f.DataFrame.Data))
}
sonny.actived++
atomic.AddInt32(&sonny.actived, 1)
loggo.Debug("Outputer processDataFrame %s %d", f.DataFrame.Id, len(f.DataFrame.Data))
}

Expand Down
Loading