Fix: use the fastest whether the result is successful
This commit is contained in:
parent
6641bf7c07
commit
8b5e511426
5 changed files with 30 additions and 63 deletions
|
@ -17,15 +17,12 @@ type Picker struct {
|
||||||
|
|
||||||
once sync.Once
|
once sync.Once
|
||||||
result interface{}
|
result interface{}
|
||||||
|
|
||||||
firstDone chan struct{}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func newPicker(ctx context.Context, cancel func()) *Picker {
|
func newPicker(ctx context.Context, cancel func()) *Picker {
|
||||||
return &Picker{
|
return &Picker{
|
||||||
ctx: ctx,
|
ctx: ctx,
|
||||||
cancel: cancel,
|
cancel: cancel,
|
||||||
firstDone: make(chan struct{}, 1),
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -42,12 +39,6 @@ func WithTimeout(ctx context.Context, timeout time.Duration) (*Picker, context.C
|
||||||
return newPicker(ctx, cancel), ctx
|
return newPicker(ctx, cancel), ctx
|
||||||
}
|
}
|
||||||
|
|
||||||
// WithoutAutoCancel returns a new Picker and an associated Context derived from ctx,
|
|
||||||
// but it wouldn't cancel context when the first element return.
|
|
||||||
func WithoutAutoCancel(ctx context.Context) *Picker {
|
|
||||||
return newPicker(ctx, nil)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Wait blocks until all function calls from the Go method have returned,
|
// Wait blocks until all function calls from the Go method have returned,
|
||||||
// then returns the first nil error result (if any) from them.
|
// then returns the first nil error result (if any) from them.
|
||||||
func (p *Picker) Wait() interface{} {
|
func (p *Picker) Wait() interface{} {
|
||||||
|
@ -58,17 +49,6 @@ func (p *Picker) Wait() interface{} {
|
||||||
return p.result
|
return p.result
|
||||||
}
|
}
|
||||||
|
|
||||||
// WaitWithoutCancel blocks until the first result return, if timeout will return nil.
|
|
||||||
// The return of this function will not wait for the cancel of context.
|
|
||||||
func (p *Picker) WaitWithoutCancel() interface{} {
|
|
||||||
select {
|
|
||||||
case <-p.firstDone:
|
|
||||||
return p.result
|
|
||||||
case <-p.ctx.Done():
|
|
||||||
return p.result
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Go calls the given function in a new goroutine.
|
// Go calls the given function in a new goroutine.
|
||||||
// The first call to return a nil error cancels the group; its result will be returned by Wait.
|
// The first call to return a nil error cancels the group; its result will be returned by Wait.
|
||||||
func (p *Picker) Go(f func() (interface{}, error)) {
|
func (p *Picker) Go(f func() (interface{}, error)) {
|
||||||
|
@ -80,7 +60,6 @@ func (p *Picker) Go(f func() (interface{}, error)) {
|
||||||
if ret, err := f(); err == nil {
|
if ret, err := f(); err == nil {
|
||||||
p.once.Do(func() {
|
p.once.Do(func() {
|
||||||
p.result = ret
|
p.result = ret
|
||||||
p.firstDone <- struct{}{}
|
|
||||||
if p.cancel != nil {
|
if p.cancel != nil {
|
||||||
p.cancel()
|
p.cancel()
|
||||||
}
|
}
|
||||||
|
|
|
@ -37,30 +37,3 @@ func TestPicker_Timeout(t *testing.T) {
|
||||||
number := picker.Wait()
|
number := picker.Wait()
|
||||||
assert.Nil(t, number)
|
assert.Nil(t, number)
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestPicker_WaitWithoutAutoCancel(t *testing.T) {
|
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*60)
|
|
||||||
defer cancel()
|
|
||||||
picker := WithoutAutoCancel(ctx)
|
|
||||||
|
|
||||||
trigger := false
|
|
||||||
picker.Go(sleepAndSend(ctx, 10, 1))
|
|
||||||
picker.Go(func() (interface{}, error) {
|
|
||||||
timer := time.NewTimer(time.Millisecond * time.Duration(30))
|
|
||||||
select {
|
|
||||||
case <-timer.C:
|
|
||||||
trigger = true
|
|
||||||
return 2, nil
|
|
||||||
case <-ctx.Done():
|
|
||||||
return nil, ctx.Err()
|
|
||||||
}
|
|
||||||
})
|
|
||||||
elm := picker.WaitWithoutCancel()
|
|
||||||
|
|
||||||
assert.NotNil(t, elm)
|
|
||||||
assert.Equal(t, elm.(int), 1)
|
|
||||||
|
|
||||||
elm = picker.Wait()
|
|
||||||
assert.True(t, trigger)
|
|
||||||
assert.Equal(t, elm.(int), 1)
|
|
||||||
}
|
|
||||||
|
|
|
@ -20,7 +20,22 @@ func (c *client) Exchange(m *D.Msg) (msg *D.Msg, err error) {
|
||||||
func (c *client) ExchangeContext(ctx context.Context, m *D.Msg) (msg *D.Msg, err error) {
|
func (c *client) ExchangeContext(ctx context.Context, m *D.Msg) (msg *D.Msg, err error) {
|
||||||
c.Client.Dialer = dialer.Dialer()
|
c.Client.Dialer = dialer.Dialer()
|
||||||
|
|
||||||
// Please note that miekg/dns ExchangeContext doesn't respond to context cancel.
|
// miekg/dns ExchangeContext doesn't respond to context cancel.
|
||||||
msg, _, err = c.Client.ExchangeContext(ctx, m, c.Address)
|
// this is a workaround
|
||||||
return
|
type result struct {
|
||||||
|
msg *D.Msg
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
ch := make(chan result, 1)
|
||||||
|
go func() {
|
||||||
|
msg, _, err := c.Client.ExchangeContext(ctx, m, c.Address)
|
||||||
|
ch <- result{msg, err}
|
||||||
|
}()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-ctx.Done():
|
||||||
|
return nil, ctx.Err()
|
||||||
|
case ret := <-ch:
|
||||||
|
return ret.msg, ret.err
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -39,7 +39,8 @@ func withFakeIP(fakePool *fakeip.Pool) middleware {
|
||||||
msg.Answer = []D.RR{rr}
|
msg.Answer = []D.RR{rr}
|
||||||
|
|
||||||
setMsgTTL(msg, 1)
|
setMsgTTL(msg, 1)
|
||||||
msg.SetReply(r)
|
msg.SetRcode(r, msg.Rcode)
|
||||||
|
msg.Authoritative = true
|
||||||
w.WriteMsg(msg)
|
w.WriteMsg(msg)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
@ -55,7 +56,8 @@ func withResolver(resolver *Resolver) handler {
|
||||||
D.HandleFailed(w, r)
|
D.HandleFailed(w, r)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
msg.SetReply(r)
|
msg.SetRcode(r, msg.Rcode)
|
||||||
|
msg.Authoritative = true
|
||||||
w.WriteMsg(msg)
|
w.WriteMsg(msg)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
|
@ -4,6 +4,7 @@ import (
|
||||||
"context"
|
"context"
|
||||||
"crypto/tls"
|
"crypto/tls"
|
||||||
"errors"
|
"errors"
|
||||||
|
"math/rand"
|
||||||
"net"
|
"net"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
@ -178,15 +179,11 @@ func (r *Resolver) batchExchange(clients []resolver, m *D.Msg) (msg *D.Msg, err
|
||||||
for _, client := range clients {
|
for _, client := range clients {
|
||||||
r := client
|
r := client
|
||||||
fast.Go(func() (interface{}, error) {
|
fast.Go(func() (interface{}, error) {
|
||||||
msg, err := r.ExchangeContext(ctx, m)
|
return r.ExchangeContext(ctx, m)
|
||||||
if err != nil || msg.Rcode != D.RcodeSuccess {
|
|
||||||
return nil, errors.New("resolve error")
|
|
||||||
}
|
|
||||||
return msg, nil
|
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
elm := fast.WaitWithoutCancel()
|
elm := fast.Wait()
|
||||||
if elm == nil {
|
if elm == nil {
|
||||||
return nil, errors.New("All DNS requests failed")
|
return nil, errors.New("All DNS requests failed")
|
||||||
}
|
}
|
||||||
|
@ -239,11 +236,12 @@ func (r *Resolver) resolveIP(host string, dnsType uint16) (ip net.IP, err error)
|
||||||
}
|
}
|
||||||
|
|
||||||
ips := r.msgToIP(msg)
|
ips := r.msgToIP(msg)
|
||||||
if len(ips) == 0 {
|
ipLength := len(ips)
|
||||||
|
if ipLength == 0 {
|
||||||
return nil, errIPNotFound
|
return nil, errIPNotFound
|
||||||
}
|
}
|
||||||
|
|
||||||
ip = ips[0]
|
ip = ips[rand.Intn(ipLength)]
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue