plugin/forward: erase expired connections by timer (#1782)
* plugin/forward: erase expired connection by timer - in previous implementation, the expired connections resided in cache until new request to the same upstream/protocol came. In case if the upstream was unhealthy new request may come long time later or may not come at all. All this time expired connections held system resources (file descriptors, ephemeral ports). In my fix the expired connections and related resources are released by timer - decreased the complexity of taking connection from cache. The list of connections is treated as stack (LIFO queue), i.e. the connection is taken from the end of queue (the most fresh connection) and returned to the end (as it was implemented before). The remarkable thing is that all connections in the stack appear to be ordered by 'used' field - the cleanup() method finds the first good (not expired) connection in stack with binary search, since all connections are ordered by 'used' field * fix race conditions * minor enhancement * add comments
This commit is contained in:
parent
94ced8255b
commit
833e3ddaf0
3 changed files with 189 additions and 17 deletions
|
@ -2,13 +2,14 @@ package forward
|
|||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/coredns/coredns/plugin/pkg/dnstest"
|
||||
|
||||
"github.com/miekg/dns"
|
||||
)
|
||||
|
||||
func TestPersistent(t *testing.T) {
|
||||
func TestCached(t *testing.T) {
|
||||
s := dnstest.NewServer(func(w dns.ResponseWriter, r *dns.Msg) {
|
||||
ret := new(dns.Msg)
|
||||
ret.SetReply(r)
|
||||
|
@ -17,17 +18,144 @@ func TestPersistent(t *testing.T) {
|
|||
defer s.Close()
|
||||
|
||||
tr := newTransport(s.Addr, nil /* no TLS */)
|
||||
tr.Start()
|
||||
defer tr.Stop()
|
||||
|
||||
c1, cache1, _ := tr.Dial("udp")
|
||||
c2, cache2, _ := tr.Dial("udp")
|
||||
c3, cache3, _ := tr.Dial("udp")
|
||||
|
||||
if cache1 || cache2 || cache3 {
|
||||
if cache1 || cache2 {
|
||||
t.Errorf("Expected non-cached connection")
|
||||
}
|
||||
|
||||
tr.Yield(c1)
|
||||
tr.Yield(c2)
|
||||
c3, cached3, _ := tr.Dial("udp")
|
||||
if !cached3 {
|
||||
t.Error("Expected cached connection (c3)")
|
||||
}
|
||||
if c2 != c3 {
|
||||
t.Error("Expected c2 == c3")
|
||||
}
|
||||
|
||||
tr.Yield(c3)
|
||||
|
||||
// dial another protocol
|
||||
c4, cached4, _ := tr.Dial("tcp")
|
||||
if cached4 {
|
||||
t.Errorf("Expected non-cached connection (c4)")
|
||||
}
|
||||
tr.Yield(c4)
|
||||
}
|
||||
|
||||
func TestCleanupByTimer(t *testing.T) {
|
||||
s := dnstest.NewServer(func(w dns.ResponseWriter, r *dns.Msg) {
|
||||
ret := new(dns.Msg)
|
||||
ret.SetReply(r)
|
||||
w.WriteMsg(ret)
|
||||
})
|
||||
defer s.Close()
|
||||
|
||||
tr := newTransport(s.Addr, nil /* no TLS */)
|
||||
tr.SetExpire(100 * time.Millisecond)
|
||||
tr.Start()
|
||||
defer tr.Stop()
|
||||
|
||||
c1, _, _ := tr.Dial("udp")
|
||||
c2, _, _ := tr.Dial("udp")
|
||||
tr.Yield(c1)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
tr.Yield(c2)
|
||||
|
||||
time.Sleep(120 * time.Millisecond)
|
||||
c3, cached, _ := tr.Dial("udp")
|
||||
if cached {
|
||||
t.Error("Expected non-cached connection (c3)")
|
||||
}
|
||||
tr.Yield(c3)
|
||||
|
||||
time.Sleep(120 * time.Millisecond)
|
||||
c4, cached, _ := tr.Dial("udp")
|
||||
if cached {
|
||||
t.Error("Expected non-cached connection (c4)")
|
||||
}
|
||||
tr.Yield(c4)
|
||||
}
|
||||
|
||||
func TestPartialCleanup(t *testing.T) {
|
||||
s := dnstest.NewServer(func(w dns.ResponseWriter, r *dns.Msg) {
|
||||
ret := new(dns.Msg)
|
||||
ret.SetReply(r)
|
||||
w.WriteMsg(ret)
|
||||
})
|
||||
defer s.Close()
|
||||
|
||||
tr := newTransport(s.Addr, nil /* no TLS */)
|
||||
tr.SetExpire(100 * time.Millisecond)
|
||||
tr.Start()
|
||||
defer tr.Stop()
|
||||
|
||||
c1, _, _ := tr.Dial("udp")
|
||||
c2, _, _ := tr.Dial("udp")
|
||||
c3, _, _ := tr.Dial("udp")
|
||||
c4, _, _ := tr.Dial("udp")
|
||||
c5, _, _ := tr.Dial("udp")
|
||||
|
||||
tr.Yield(c1)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
tr.Yield(c2)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
tr.Yield(c3)
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
tr.Yield(c4)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
tr.Yield(c5)
|
||||
time.Sleep(40 * time.Millisecond)
|
||||
|
||||
c6, _, _ := tr.Dial("udp")
|
||||
if c6 != c5 {
|
||||
t.Errorf("Expected c6 == c5")
|
||||
}
|
||||
c7, _, _ := tr.Dial("udp")
|
||||
if c7 != c4 {
|
||||
t.Errorf("Expected c7 == c4")
|
||||
}
|
||||
c8, cached, _ := tr.Dial("udp")
|
||||
if cached {
|
||||
t.Error("Expected non-cached connection (c8)")
|
||||
}
|
||||
|
||||
tr.Yield(c6)
|
||||
tr.Yield(c7)
|
||||
tr.Yield(c8)
|
||||
}
|
||||
|
||||
func TestCleanupAll(t *testing.T) {
|
||||
s := dnstest.NewServer(func(w dns.ResponseWriter, r *dns.Msg) {
|
||||
ret := new(dns.Msg)
|
||||
ret.SetReply(r)
|
||||
w.WriteMsg(ret)
|
||||
})
|
||||
defer s.Close()
|
||||
|
||||
tr := newTransport(s.Addr, nil /* no TLS */)
|
||||
|
||||
c1, _ := dns.DialTimeout("udp", tr.addr, dialTimeout)
|
||||
c2, _ := dns.DialTimeout("udp", tr.addr, dialTimeout)
|
||||
c3, _ := dns.DialTimeout("udp", tr.addr, dialTimeout)
|
||||
|
||||
tr.conns["udp"] = []*persistConn{
|
||||
{c1, time.Now()},
|
||||
{c2, time.Now()},
|
||||
{c3, time.Now()},
|
||||
}
|
||||
|
||||
if tr.len() != 3 {
|
||||
t.Error("Expected 3 connections")
|
||||
}
|
||||
tr.cleanup(true)
|
||||
|
||||
if tr.len() > 0 {
|
||||
t.Error("Expected no cached connections")
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue