Skip to content

feat: add per-proxy-group bandwidth limiting #3053

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bobbyunknown wants to merge 1 commit into
base: Alpha
Choose a base branch
from
Open

feat: add per-proxy-group bandwidth limiting #3053

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
5 changes: 5 additions & 0 deletions adapter/outbound/base.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,6 +142,11 @@ func (b *Base) Unwrap(metadata *C.Metadata, touch bool) C.Proxy {
return nil
}

// Bandwidth implements C.ProxyAdapter. Standalone proxies have no bandwidth limit.
func (b *Base) Bandwidth() uint64 {
return 0
}

// DialOptions return []dialer.Option from struct
func (b *Base) DialOptions() (opts []dialer.Option) {
if b.iface != "" {
Expand Down
7 changes: 6 additions & 1 deletion adapter/outboundgroup/fallback.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,9 @@ func (f *Fallback) DialContext(ctx context.Context, metadata *C.Metadata) (C.Con
})
}

if err == nil {
c = f.LimitConn(c)
}
return c, err
}

Expand All @@ -57,6 +60,7 @@ func (f *Fallback) ListenPacketContext(ctx context.Context, metadata *C.Metadata
pc, err := proxy.ListenPacketContext(ctx, metadata)
if err == nil {
pc.AppendToChains(f)
pc = f.LimitPacketConn(pc)
}

return pc, err
Expand Down Expand Up @@ -159,7 +163,7 @@ func (f *Fallback) Proxies() []C.Proxy {
return f.GetProxies(false)
}

func NewFallback(option GroupCommonOption, fallbackOption FallbackOption, emptyFallback C.Proxy, providers []P.ProxyProvider) (*Fallback, error) {
func NewFallback(option GroupCommonOption, fallbackOption FallbackOption, emptyFallback C.Proxy, providers []P.ProxyProvider, bandwidth uint64) (*Fallback, error) {
return &Fallback{
GroupBase: NewGroupBase(GroupBaseOption{
Name: option.Name,
Expand All @@ -173,6 +177,7 @@ func NewFallback(option GroupCommonOption, fallbackOption FallbackOption, emptyF
MaxFailedTimes: option.MaxFailedTimes,
EmptyFallback: emptyFallback,
Providers: providers,
Bandwidth: bandwidth,
}),
disableUDP: option.DisableUDP,
testUrl: option.URL,
Expand Down
18 changes: 18 additions & 0 deletions adapter/outboundgroup/groupbase.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (
"github.com/metacubex/mihomo/adapter/outbound"
"github.com/metacubex/mihomo/common/atomic"
"github.com/metacubex/mihomo/common/utils"
"github.com/metacubex/mihomo/component/ratelimit"
C "github.com/metacubex/mihomo/constant"
P "github.com/metacubex/mihomo/constant/provider"
"github.com/metacubex/mihomo/log"
Expand All @@ -34,6 +35,7 @@ type GroupBase struct {
testTimeout int
maxFailedTimes int
emptyFallback C.Proxy
limiter *ratelimit.Limiter // shared group bandwidth limit, nil = unlimited

// for GetProxies
getProxiesMutex sync.Mutex
Expand All @@ -53,6 +55,7 @@ type GroupBaseOption struct {
MaxFailedTimes int
EmptyFallback C.Proxy
Providers []P.ProxyProvider
Bandwidth uint64 // bits per second, 0 = unlimited
}

func NewGroupBase(opt GroupBaseOption) *GroupBase {
Expand Down Expand Up @@ -89,6 +92,7 @@ func NewGroupBase(opt GroupBaseOption) *GroupBase {
testTimeout: opt.TestTimeout,
maxFailedTimes: opt.MaxFailedTimes,
emptyFallback: opt.EmptyFallback,
limiter: ratelimit.NewLimiter(opt.Bandwidth),
}

if gb.testTimeout == 0 {
Expand All @@ -109,6 +113,20 @@ func (gb *GroupBase) Icon() string {
return gb.icon
}

func (gb *GroupBase) Bandwidth() uint64 {
return gb.limiter.Rate()
}

// LimitConn applies the group's shared bandwidth limiter to conn.
func (gb *GroupBase) LimitConn(conn C.Conn) C.Conn {
return gb.limiter.WrapCConn(conn)
}

// LimitPacketConn applies the group's shared bandwidth limiter to a packet conn.
func (gb *GroupBase) LimitPacketConn(pc C.PacketConn) C.PacketConn {
return gb.limiter.WrapPacketConn(pc)
}

func (gb *GroupBase) EmptyFallback() C.Proxy {
return gb.emptyFallback
}
Expand Down
7 changes: 6 additions & 1 deletion adapter/outboundgroup/loadbalance.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -101,6 +101,9 @@ func (lb *LoadBalance) DialContext(ctx context.Context, metadata *C.Metadata) (c
})
}

if err == nil {
c = lb.LimitConn(c)
}
return
}

Expand All @@ -109,6 +112,7 @@ func (lb *LoadBalance) ListenPacketContext(ctx context.Context, metadata *C.Meta
defer func() {
if err == nil {
pc.AppendToChains(lb)
pc = lb.LimitPacketConn(pc)
}
}()

Expand Down Expand Up @@ -247,7 +251,7 @@ func (lb *LoadBalance) Now() string {
return ""
}

func NewLoadBalance(option GroupCommonOption, loadBalanceOption LoadBalanceOption, emptyFallback C.Proxy, providers []P.ProxyProvider) (lb *LoadBalance, err error) {
func NewLoadBalance(option GroupCommonOption, loadBalanceOption LoadBalanceOption, emptyFallback C.Proxy, providers []P.ProxyProvider, bandwidth uint64) (lb *LoadBalance, err error) {
var strategyFn strategyFn
switch loadBalanceOption.Strategy {
case "", "consistent-hashing":
Expand All @@ -272,6 +276,7 @@ func NewLoadBalance(option GroupCommonOption, loadBalanceOption LoadBalanceOptio
MaxFailedTimes: option.MaxFailedTimes,
EmptyFallback: emptyFallback,
Providers: providers,
Bandwidth: bandwidth,
}),
strategyFn: strategyFn,
disableUDP: option.DisableUDP,
Expand Down
12 changes: 8 additions & 4 deletions adapter/outboundgroup/parser.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import (
"github.com/metacubex/mihomo/adapter/provider"
"github.com/metacubex/mihomo/common/structure"
"github.com/metacubex/mihomo/common/utils"
"github.com/metacubex/mihomo/component/ratelimit"
C "github.com/metacubex/mihomo/constant"
P "github.com/metacubex/mihomo/constant/provider"
"github.com/metacubex/mihomo/log"
Expand Down Expand Up @@ -43,6 +44,7 @@ type GroupCommonOption struct {
IncludeAllProviders bool `group:"include-all-providers,omitempty"`
Hidden bool `group:"hidden,omitempty"`
Icon string `group:"icon,omitempty"`
Bandwidth string `group:"bandwidth,omitempty"`
}

func ParseProxyGroup(config map[string]any, proxyMap map[string]C.Proxy, providersMap map[string]P.ProxyProvider, AllProxies []string, AllProviders []string) (ProxyGroup, error) {
Expand All @@ -55,6 +57,8 @@ func ParseProxyGroup(config map[string]any, proxyMap map[string]C.Proxy, provide
return nil, errFormat
}

bandwidth := ratelimit.ParseBandwidth(groupOption.Bandwidth)

if groupOption.Type == "" || groupOption.Name == "" {
return nil, errFormat
}
Expand Down Expand Up @@ -191,28 +195,28 @@ func ParseProxyGroup(config map[string]any, proxyMap map[string]C.Proxy, provide
if err != nil {
return nil, err
}
return NewURLTest(groupOption, opt, emptyFallback, providers)
return NewURLTest(groupOption, opt, emptyFallback, providers, bandwidth)
case "select":
opt := SelectorOption{}
err = decoder.Decode(config, &opt)
if err != nil {
return nil, err
}
return NewSelector(groupOption, opt, emptyFallback, providers)
return NewSelector(groupOption, opt, emptyFallback, providers, bandwidth)
case "fallback":
opt := FallbackOption{}
err = decoder.Decode(config, &opt)
if err != nil {
return nil, err
}
return NewFallback(groupOption, opt, emptyFallback, providers)
return NewFallback(groupOption, opt, emptyFallback, providers, bandwidth)
case "load-balance":
opt := LoadBalanceOption{}
err = decoder.Decode(config, &opt)
if err != nil {
return nil, err
}
return NewLoadBalance(groupOption, opt, emptyFallback, providers)
return NewLoadBalance(groupOption, opt, emptyFallback, providers, bandwidth)
case "relay":
return nil, fmt.Errorf("%w: The group [%s] with relay type was removed, please using dialer-proxy instead", errType, groupName)
default:
Expand Down
5 changes: 4 additions & 1 deletion adapter/outboundgroup/selector.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ func (s *Selector) DialContext(ctx context.Context, metadata *C.Metadata) (C.Con
c, err := s.selectedProxy(true).DialContext(ctx, metadata)
if err == nil {
c.AppendToChains(s)
c = s.LimitConn(c)
}
return c, err
}
Expand All @@ -34,6 +35,7 @@ func (s *Selector) ListenPacketContext(ctx context.Context, metadata *C.Metadata
pc, err := s.selectedProxy(true).ListenPacketContext(ctx, metadata)
if err == nil {
pc.AppendToChains(s)
pc = s.LimitPacketConn(pc)
}
return pc, err
}
Expand Down Expand Up @@ -119,7 +121,7 @@ func (s *Selector) Proxies() []C.Proxy {
return s.GetProxies(false)
}

func NewSelector(option GroupCommonOption, selectorOption SelectorOption, emptyFallback C.Proxy, providers []P.ProxyProvider) (*Selector, error) {
func NewSelector(option GroupCommonOption, selectorOption SelectorOption, emptyFallback C.Proxy, providers []P.ProxyProvider, bandwidth uint64) (*Selector, error) {
return &Selector{
GroupBase: NewGroupBase(GroupBaseOption{
Name: option.Name,
Expand All @@ -133,6 +135,7 @@ func NewSelector(option GroupCommonOption, selectorOption SelectorOption, emptyF
MaxFailedTimes: option.MaxFailedTimes,
EmptyFallback: emptyFallback,
Providers: providers,
Bandwidth: bandwidth,
}),
selected: selectorOption.DefaultSelected,
disableUDP: option.DisableUDP,
Expand Down
7 changes: 6 additions & 1 deletion adapter/outboundgroup/urltest.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,9 @@ func (u *URLTest) DialContext(ctx context.Context, metadata *C.Metadata) (c C.Co
})
}

if err == nil {
c = u.LimitConn(c)
}
return c, err
}

Expand All @@ -82,6 +85,7 @@ func (u *URLTest) ListenPacketContext(ctx context.Context, metadata *C.Metadata)
pc, err := proxy.ListenPacketContext(ctx, metadata)
if err == nil {
pc.AppendToChains(u)
pc = u.LimitPacketConn(pc)
} else {
u.onDialFailed(proxy.Type(), err, u.healthCheck)
}
Expand Down Expand Up @@ -192,7 +196,7 @@ func (u *URLTest) URLTest(ctx context.Context, url string, expectedStatus utils.
return u.GroupBase.URLTest(ctx, u.testUrl, expectedStatus)
}

func NewURLTest(option GroupCommonOption, urlTestOption URLTestOption, emptyFallback C.Proxy, providers []P.ProxyProvider) (*URLTest, error) {
func NewURLTest(option GroupCommonOption, urlTestOption URLTestOption, emptyFallback C.Proxy, providers []P.ProxyProvider, bandwidth uint64) (*URLTest, error) {
if emptyFallback == nil {
return nil, errors.New("empty fallback proxy not exist")
}
Expand All @@ -209,6 +213,7 @@ func NewURLTest(option GroupCommonOption, urlTestOption URLTestOption, emptyFall
MaxFailedTimes: option.MaxFailedTimes,
EmptyFallback: emptyFallback,
Providers: providers,
Bandwidth: bandwidth,
}),
fastSingle: singledo.NewSingle[C.Proxy](time.Second * 10),
disableUDP: option.DisableUDP,
Expand Down
Loading