Add opt-in per-server dial backoff

memcache/backoff.go

@@ -0,0 +1,113 @@
+package memcache
+
+import (
+	"crypto/rand"
+	"encoding/binary"
+	"net"
+	"sync"
+	"time"
+)
+
+// nodeBackoff tracks consecutive failed dials to one server. Once failures
+// reach the threshold, dials are skipped until blockedUntil; after that a
+// single dial is let through to test the server while other callers keep
+// failing fast, so a dead server sees one dial per backoff period per client
+// instead of one per request.
+type nodeBackoff struct {
+	threshold   int
+	baseBackoff time.Duration
+	maxBackoff  time.Duration
+
+	mu            sync.Mutex
+	failures      int
+	backoff       time.Duration
+	blockedUntil  time.Time
+	retryInFlight bool
+}
+
+// allow reports whether a dial may be attempted. Every allow() that returns
+// true must be followed by exactly one record(), or the test dial is never
+// released and the server stays backed off.
+func (n *nodeBackoff) allow() bool {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+
+	if n.failures < n.threshold {
+		return true
+	}
+	if time.Now().Before(n.blockedUntil) || n.retryInFlight {
+		return false
+	}
+	n.retryInFlight = true
+	return true
+}
+
+// record feeds a dial's outcome back into the tracker.
+func (n *nodeBackoff) record(failed bool) {
+	n.mu.Lock()
+	defer n.mu.Unlock()
+
+	n.retryInFlight = false
+	if !failed {
+		n.failures = 0
+		n.backoff = n.baseBackoff
+		return
+	}
+
+	n.failures++
+	// Dials started before the block keep failing in after it; only the
+	// failure that crosses the threshold or a failed test dial extends it.
+	if n.failures < n.threshold || time.Now().Before(n.blockedUntil) {
+		return
+	}
+	n.blockedUntil = time.Now().Add(withJitter(n.backoff))
+	n.backoff *= 2
+	if n.backoff > n.maxBackoff {
+		n.backoff = n.maxBackoff
+	}
+}
+
+// withJitter returns a random duration in [d/2, 3d/2) so clients that backed
+// off a server at the same moment don't all retry it at the same moment.
+func withJitter(d time.Duration) time.Duration {
+	var buf [8]byte
+	if _, err := rand.Read(buf[:]); err != nil {
+		return d
+	}
+	frac := float64(binary.LittleEndian.Uint64(buf[:])>>11) / (1 << 53)
+	return d/2 + time.Duration(frac*float64(d))
+}
+
+// nodeBackoff returns the tracker for addr, creating it on first use.
+func (c *Client) nodeBackoff(addr net.Addr) *nodeBackoff {
+	c.lk.Lock()
+	defer c.lk.Unlock()
+	if c.backoffs == nil {
+		c.backoffs = make(map[string]*nodeBackoff)
+	}
+	n, ok := c.backoffs[addr.String()]
+	if !ok {
+		n = &nodeBackoff{
+			threshold:   c.FailureThreshold,
+			baseBackoff: c.baseBackoff(),
+			maxBackoff:  c.maxBackoff(),
+			backoff:     c.baseBackoff(),
+		}
+		c.backoffs[addr.String()] = n
+	}
+	return n
+}
+
+func (c *Client) baseBackoff() time.Duration {
+	if c.BaseBackoff > 0 {
+		return c.BaseBackoff
+	}
+	return DefaultBaseBackoff
+}
+
+func (c *Client) maxBackoff() time.Duration {
+	if c.MaxBackoff > 0 {
+		return c.MaxBackoff
+	}
+	return DefaultMaxBackoff
+}

memcache/memcache.go

@@ -61,6 +61,10 @@ var (
 
 	// ErrNoServers is returned when no servers are configured or available.
 	ErrNoServers = errors.New("memcache: no servers configured or available")
+
+	// ErrConnectionBroken is returned without dialing when the server has
+	// failed FailureThreshold dials in a row and is being backed off.
+	ErrConnectionBroken = errors.New("memcache: connection to server is broken, backing off")
 )
 
 const (
@@ -70,6 +74,12 @@ const (
 	// DefaultMaxIdleConns is the default maximum number of idle connections
 	// kept for any single address.
 	DefaultMaxIdleConns = 2
+
+	// DefaultBaseBackoff is the default first backoff for a failing server.
+	DefaultBaseBackoff = 1 * time.Second
+
+	// DefaultMaxBackoff is the default cap on a failing server's backoff.
+	DefaultMaxBackoff = 60 * time.Second
 )
 
 const buffered = 8 // arbitrary buffered channel size, for readability
@@ -149,10 +159,24 @@ type Client struct {
 	// be set to a number higher than your peak parallel requests.
 	MaxIdleConns int
 
+	// FailureThreshold is the number of consecutive failed dials to a server
+	// after which new dials to it are skipped (returning ErrConnectionBroken)
+	// for a backoff period. If zero, backoff is disabled.
+	FailureThreshold int
+
+	// BaseBackoff is the first backoff for a failing server; each further
+	// failed dial doubles it. If zero, DefaultBaseBackoff is used.
+	BaseBackoff time.Duration
+
+	// MaxBackoff caps a failing server's backoff. If zero, DefaultMaxBackoff
+	// is used.
+	MaxBackoff time.Duration
+
 	selector ServerSelector
 
 	lk        sync.Mutex
 	freeconn  map[string][]*conn
+	backoffs  map[string]*nodeBackoff
 	TlsConfig *tls.Config
 }
 
@@ -299,7 +323,17 @@ func (c *Client) getConn(addr net.Addr) (*conn, error) {
 		cn.extendDeadline()
 		return cn, nil
 	}
+	var backoff *nodeBackoff
+	if c.FailureThreshold > 0 {
+		backoff = c.nodeBackoff(addr)
+		if !backoff.allow() {
+			return nil, ErrConnectionBroken
+		}
+	}
 	nc, err := c.dial(addr)
+	if backoff != nil {
+		backoff.record(err != nil)
+	}
 	if err != nil {
 		return nil, err
 	}