From cec5e71d4e99b3a359a88d05317f18e18463b9c7 Mon Sep 17 00:00:00 2001 From: mzz <2017@duck.com> Date: Fri, 20 Sep 2024 01:40:05 +0800 Subject: [PATCH] fix: crash on openwrt (#640) Co-authored-by: dae-prow[bot] <136105375+dae-prow[bot]@users.noreply.github.com> --- control/udp.go | 11 +++-- control/udp_task_pool.go | 84 ++++++++++++++++++++++------------- control/udp_task_pool_test.go | 31 +++++++++++++ go.mod | 9 ++++ go.sum | 17 +++++++ 5 files changed, 118 insertions(+), 34 deletions(-) create mode 100644 control/udp_task_pool_test.go diff --git a/control/udp.go b/control/udp.go index 5f3b459..814224f 100644 --- a/control/udp.go +++ b/control/udp.go @@ -22,11 +22,14 @@ import ( "github.com/sirupsen/logrus" ) -const ( +var ( DefaultNatTimeout = 3 * time.Minute - DnsNatTimeout = 17 * time.Second // RFC 5452 - AnyfromTimeout = 5 * time.Second // Do not cache too long. - MaxRetry = 2 +) + +const ( + DnsNatTimeout = 17 * time.Second // RFC 5452 + AnyfromTimeout = 5 * time.Second // Do not cache too long. + MaxRetry = 2 ) type DialOption struct { diff --git a/control/udp_task_pool.go b/control/udp_task_pool.go index c19d145..c79a31f 100644 --- a/control/udp_task_pool.go +++ b/control/udp_task_pool.go @@ -2,19 +2,26 @@ package control import ( "sync" + "sync/atomic" "time" + + "github.com/panjf2000/ants" ) +var isTest = false + const UdpTaskQueueLength = 128 type UdpTask = func() +// UdpTaskQueue make sure packets with the same key (4 tuples) will be sent in order. type UdpTaskQueue struct { key string + p *UdpTaskPool ch chan UdpTask timer *time.Timer agingTime time.Duration - closed chan struct{} + closed atomic.Bool freed chan struct{} } @@ -23,6 +30,30 @@ func (q *UdpTaskQueue) Push(task UdpTask) { q.ch <- task } +func (q *UdpTaskQueue) convoy() { + for { + if q.closed.Load() { + clearloop: + for { + select { + case t := <-q.ch: + // Emit it back due to closed q. + ReemitWorkers.Submit(func() { + q.p.EmitTask(q.key, t) + }) + default: + break clearloop + } + } + close(q.freed) + return + } else { + t := <-q.ch + t() + } + } +} + type UdpTaskPool struct { queueChPool sync.Pool // mu protects m @@ -41,28 +72,7 @@ func NewUdpTaskPool() *UdpTaskPool { return p } -func (p *UdpTaskPool) convoy(q *UdpTaskQueue) { - for { - select { - case <-q.closed: - clearloop: - for { - select { - case t := <-q.ch: - // Emit it back due to closed q. - p.EmitTask(q.key, t) - default: - break clearloop - } - } - close(q.freed) - return - case t := <-q.ch: - t() - } - } -} - +// EmitTask: Make sure packets with the same key (4 tuples) will be sent in order. func (p *UdpTaskPool) EmitTask(key string, task UdpTask) { p.mu.Lock() q, ok := p.m[key] @@ -70,33 +80,47 @@ func (p *UdpTaskPool) EmitTask(key string, task UdpTask) { ch := p.queueChPool.Get().(chan UdpTask) q = &UdpTaskQueue{ key: key, + p: p, ch: ch, timer: nil, agingTime: DefaultNatTimeout, - closed: make(chan struct{}), + closed: atomic.Bool{}, freed: make(chan struct{}), } q.timer = time.AfterFunc(q.agingTime, func() { // This func may be invoked twice due to concurrent Reset. - select { - case <-q.closed: + if !q.closed.CompareAndSwap(false, true) { return - default: + } + if isTest { + time.Sleep(3 * time.Microsecond) } p.mu.Lock() defer p.mu.Unlock() if p.m[key] == q { delete(p.m, key) } - close(q.closed) + // Trigger next loop in func convoy + q.ch <- func() {} <-q.freed p.queueChPool.Put(ch) }) p.m[key] = q - go p.convoy(q) + go q.convoy() } p.mu.Unlock() q.Push(task) } -var DefaultUdpTaskPool = NewUdpTaskPool() +var ( + DefaultUdpTaskPool = NewUdpTaskPool() + ReemitWorkers *ants.Pool +) + +func init() { + var err error + ReemitWorkers, err = ants.NewPool(UdpTaskQueueLength/2, ants.WithExpiryDuration(AnyfromTimeout)) + if err != nil { + panic(err) + } +} diff --git a/control/udp_task_pool_test.go b/control/udp_task_pool_test.go new file mode 100644 index 0000000..b2a8689 --- /dev/null +++ b/control/udp_task_pool_test.go @@ -0,0 +1,31 @@ +/* + * SPDX-License-Identifier: AGPL-3.0-only + * Copyright (c) 2022-2024, daeuniverse Organization + */ + +package control + +import ( + "testing" + "time" + + "github.com/shirou/gopsutil/v4/cpu" + "github.com/stretchr/testify/require" +) + +func TestUdpTaskPool(t *testing.T) { + isTest = true + c, err := cpu.Times(false) + require.NoError(t, err) + t.Log(c) + DefaultNatTimeout = 1000 * time.Microsecond + for i := 0; i < 100; i++ { + DefaultUdpTaskPool.EmitTask("testkey", func() { + }) + time.Sleep(99 * time.Microsecond) + } + time.Sleep(5 * time.Second) + c, err = cpu.Times(false) + require.NoError(t, err) + t.Log(c) +} diff --git a/go.mod b/go.mod index efac4b5..573ed27 100644 --- a/go.mod +++ b/go.mod @@ -19,6 +19,7 @@ require ( github.com/shirou/gopsutil/v4 v4.24.5 github.com/sirupsen/logrus v1.9.3 github.com/spf13/cobra v1.7.0 + github.com/stretchr/testify v1.9.0 github.com/v2rayA/ahocorasick-domain v0.0.0-20231231085011-99ceb8ef3208 github.com/vishvananda/netlink v1.1.0 github.com/vishvananda/netns v0.0.4 @@ -37,6 +38,7 @@ require ( github.com/awnumar/memguard v0.19.1 // indirect github.com/cloudflare/circl v1.3.7 // indirect github.com/daeuniverse/quic-go v0.0.0-20240413031024-943f218e0810 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect github.com/dsnet/compress v0.0.2-0.20210315054119-f66993602bf5 // indirect github.com/go-ole/go-ole v1.2.6 // indirect github.com/go-task/slim-sprig v0.0.0-20230315185526-52ccab3ef572 // indirect @@ -45,11 +47,17 @@ require ( github.com/gorilla/websocket v1.5.0 // indirect github.com/klauspost/compress v1.17.4 // indirect github.com/klauspost/pgzip v1.2.5 // indirect + github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect github.com/nwaples/rardecode v1.1.0 // indirect github.com/onsi/ginkgo/v2 v2.11.0 // indirect + github.com/panjf2000/ants v1.3.0 // indirect github.com/pierrec/lz4/v4 v4.1.2 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect github.com/quic-go/qpack v0.4.0 // indirect + github.com/shoenig/go-m1cpu v0.1.6 // indirect + github.com/tklauser/go-sysconf v0.3.12 // indirect + github.com/tklauser/numcpus v0.6.1 // indirect github.com/ulikunitz/xz v0.5.9 // indirect github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect github.com/yusufpapurcu/wmi v1.2.4 // indirect @@ -58,6 +66,7 @@ require ( golang.org/x/net v0.22.0 // indirect golang.org/x/tools v0.18.0 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20230807174057-1744710a1577 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect ) require ( diff --git a/go.sum b/go.sum index 9ce53b8..e9f0383 100644 --- a/go.sum +++ b/go.sum @@ -77,6 +77,7 @@ github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMyw github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= @@ -102,6 +103,8 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4= +github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I= github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA= github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg= github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM= @@ -139,6 +142,8 @@ github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7J github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo= github.com/onsi/gomega v1.27.8 h1:gegWiwZjBsf2DgiSbf5hpokZ98JVDMcWkUiigk6/KXc= github.com/onsi/gomega v1.27.8/go.mod h1:2J8vzI/s+2shY9XHRApDkdgPo1TKT7P2u6fXeJKFnNQ= +github.com/panjf2000/ants v1.3.0 h1:8pQ+8leaLc9lys2viEEr8md0U4RN6uOSUCE9bOYjQ9M= +github.com/panjf2000/ants v1.3.0/go.mod h1:AaACblRPzq35m1g3enqYcxspbbiOJJYaxU2wMpm1cXY= github.com/pierrec/lz4/v4 v4.1.2 h1:qvY3YFXRQE/XB8MlLzJH7mSzBs74eA2gg52YTk6jUPM= github.com/pierrec/lz4/v4 v4.1.2/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= @@ -158,6 +163,10 @@ github.com/seiflotfy/cuckoofilter v0.0.0-20220411075957-e3b120b3f5fb h1:XfLJSPIO github.com/seiflotfy/cuckoofilter v0.0.0-20220411075957-e3b120b3f5fb/go.mod h1:bR6DqgcAl1zTcOX8/pE2Qkj9XO00eCNqmKb7lXP8EAg= github.com/shirou/gopsutil/v4 v4.24.5 h1:gGsArG5K6vmsh5hcFOHaPm87UD003CaDMkAOweSQjhM= github.com/shirou/gopsutil/v4 v4.24.5/go.mod h1:aoebb2vxetJ/yIDZISmduFvVNPHqXQ9SEJwRXxkf0RA= +github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFtM= +github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ= +github.com/shoenig/test v0.6.4 h1:kVTaSd7WLz5WZ2IaoM0RSzRsUD+m8wRR+5qvntpn4LU= +github.com/shoenig/test v0.6.4/go.mod h1:byHiCGXqrVaflBLAMq/srcZIHynQPQgeyvkvXnjqq0k= github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ= github.com/sirupsen/logrus v1.9.3/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= github.com/spf13/cobra v1.7.0 h1:hyqWnYt1ZQShIddO5kBpj3vu05/++x6tJ6dg8EC572I= @@ -172,6 +181,10 @@ github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/ github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFAEVmqU= +github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI= +github.com/tklauser/numcpus v0.6.1 h1:ng9scYS7az0Bk4OZLvrNXNSAO2Pxr1XXRAPyjhIx+Fk= +github.com/tklauser/numcpus v0.6.1/go.mod h1:1XfjsgE2zo8GVw7POkMbHENHzVg3GzmoZ9fESEdAacY= github.com/twmb/murmur3 v1.1.6 h1:mqrRot1BRxm+Yct+vavLMou2/iJt0tNVTTC0QoIjaZg= github.com/twmb/murmur3 v1.1.6/go.mod h1:Qq/R7NUyOfr65zD+6Q5IHKsJLwP7exErjN6lyyq3OSQ= github.com/ulikunitz/xz v0.5.8/go.mod h1:nbz6k7qbPmH4IRqmfOplQw/tblSgqTqBwxkY0oWt/14= @@ -237,6 +250,8 @@ golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.20.0 h1:Od9JTbYCk261bKm4M/mw7AklTlFYIa0bIp9BgSm1S8Y= golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/term v0.18.0 h1:FcHjZXDMxI8mM3nwhX9HlKop4C0YQvCVCdwYl2wOtE8= @@ -271,6 +286,8 @@ google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQ google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI= google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys= gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc= gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc=