1
0
mirror of https://github.com/v2fly/v2ray-core.git synced 2025-01-04 16:37:12 -05:00

add exponential backoff as retry logic

This commit is contained in:
Darien Raymond 2016-11-20 21:47:51 +01:00
parent 56fb8c478c
commit 30cd9e929d
No known key found for this signature in database
GPG Key ID: 7251FFA14BB18169
7 changed files with 58 additions and 27 deletions

View File

@ -16,34 +16,43 @@ type Strategy interface {
} }
type retryer struct { type retryer struct {
NextDelay func(int) int totalAttempt int
nextDelay func() uint32
} }
// On implements Strategy.On. // On implements Strategy.On.
func (r *retryer) On(method func() error) error { func (r *retryer) On(method func() error) error {
attempt := 0 attempt := 0
for { for attempt < r.totalAttempt {
err := method() err := method()
if err == nil { if err == nil {
return nil return nil
} }
delay := r.NextDelay(attempt) delay := r.nextDelay()
if delay < 0 {
return ErrRetryFailed
}
<-time.After(time.Duration(delay) * time.Millisecond) <-time.After(time.Duration(delay) * time.Millisecond)
attempt++ attempt++
} }
return ErrRetryFailed
} }
// Timed returns a retry strategy with fixed interval. // Timed returns a retry strategy with fixed interval.
func Timed(attempts int, delay int) Strategy { func Timed(attempts int, delay uint32) Strategy {
return &retryer{ return &retryer{
NextDelay: func(attempt int) int { totalAttempt: attempts,
if attempt >= attempts { nextDelay: func() uint32 {
return -1
}
return delay return delay
}, },
} }
} }
func ExponentialBackoff(attempts int, delay uint32) Strategy {
nextDelay := uint32(0)
return &retryer{
totalAttempt: attempts,
nextDelay: func() uint32 {
r := nextDelay
nextDelay += delay
return r
},
}
}

View File

@ -68,14 +68,26 @@ func TestRetryExhausted(t *testing.T) {
startTime := time.Now() startTime := time.Now()
called := 0 called := 0
err := Timed(2, 1000).On(func() error { err := Timed(2, 1000).On(func() error {
if called < 5 { called++
called++ return errorTestOnly
return errorTestOnly
}
return nil
}) })
duration := time.Since(startTime) duration := time.Since(startTime)
assert.Error(err).Equals(ErrRetryFailed) assert.Error(err).Equals(ErrRetryFailed)
assert.Int64(int64(duration / time.Millisecond)).AtLeast(1900) assert.Int64(int64(duration / time.Millisecond)).AtLeast(1900)
} }
func TestExponentialBackoff(t *testing.T) {
assert := assert.On(t)
startTime := time.Now()
called := 0
err := ExponentialBackoff(10, 100).On(func() error {
called++
return errorTestOnly
})
duration := time.Since(startTime)
assert.Error(err).Equals(ErrRetryFailed)
assert.Int64(int64(duration / time.Millisecond)).AtLeast(4000)
}

View File

@ -58,7 +58,7 @@ func (this *InboundDetourHandlerAlways) Close() {
// Starts the inbound connection handler. // Starts the inbound connection handler.
func (this *InboundDetourHandlerAlways) Start() error { func (this *InboundDetourHandlerAlways) Start() error {
for _, ich := range this.ich { for _, ich := range this.ich {
err := retry.Timed(100 /* times */, 100 /* ms */).On(func() error { err := retry.ExponentialBackoff(10 /* times */, 200 /* ms */).On(func() error {
err := ich.Start() err := ich.Start()
if err != nil { if err != nil {
log.Error("Failed to start inbound detour:", err) log.Error("Failed to start inbound detour:", err)

View File

@ -79,7 +79,7 @@ func (this *FreedomConnection) Dispatch(destination v2net.Destination, payload *
if this.domainStrategy == Config_USE_IP && destination.Address.Family().IsDomain() { if this.domainStrategy == Config_USE_IP && destination.Address.Family().IsDomain() {
destination = this.ResolveIP(destination) destination = this.ResolveIP(destination)
} }
err := retry.Timed(5, 100).On(func() error { err := retry.ExponentialBackoff(5, 100).On(func() error {
rawConn, err := internet.Dial(this.meta.Address, destination, this.meta.GetDialerOptions()) rawConn, err := internet.Dial(this.meta.Address, destination, this.meta.GetDialerOptions())
if err != nil { if err != nil {
return err return err

View File

@ -44,7 +44,7 @@ func (this *Client) Dispatch(destination v2net.Destination, payload *alloc.Buffe
var server *protocol.ServerSpec var server *protocol.ServerSpec
var conn internet.Connection var conn internet.Connection
err := retry.Timed(5, 100).On(func() error { err := retry.ExponentialBackoff(5, 100).On(func() error {
server = this.serverPicker.PickServer() server = this.serverPicker.PickServer()
dest := server.Destination() dest := server.Destination()
dest.Network = network dest.Network = network

View File

@ -33,7 +33,7 @@ func (this *VMessOutboundHandler) Dispatch(target v2net.Destination, payload *al
var rec *protocol.ServerSpec var rec *protocol.ServerSpec
var conn internet.Connection var conn internet.Connection
err := retry.Timed(5, 100).On(func() error { err := retry.ExponentialBackoff(5, 100).On(func() error {
rec = this.serverPicker.PickServer() rec = this.serverPicker.PickServer()
rawConn, err := internet.Dial(this.meta.Address, rec.Destination(), this.meta.GetDialerOptions()) rawConn, err := internet.Dial(this.meta.Address, rec.Destination(), this.meta.GetDialerOptions())
if err != nil { if err != nil {

View File

@ -7,6 +7,7 @@ import (
"v2ray.com/core/common/log" "v2ray.com/core/common/log"
v2net "v2ray.com/core/common/net" v2net "v2ray.com/core/common/net"
"v2ray.com/core/common/retry"
) )
var ( var (
@ -78,14 +79,23 @@ func (this *TCPHub) Close() {
func (this *TCPHub) start() { func (this *TCPHub) start() {
this.accepting = true this.accepting = true
for this.accepting { for this.accepting {
conn, err := this.listener.Accept() var newConn Connection
err := retry.ExponentialBackoff(10, 200).On(func() error {
if err != nil { if !this.accepting {
if this.accepting { return nil
log.Warning("Internet|Listener: Failed to accept new TCP connection: ", err)
} }
continue conn, err := this.listener.Accept()
if err != nil {
if this.accepting {
log.Warning("Internet|Listener: Failed to accept new TCP connection: ", err)
}
return err
}
newConn = conn
return nil
})
if err == nil && newConn != nil {
go this.connCallback(newConn)
} }
go this.connCallback(conn)
} }
} }