Upstream changes

caching/rate_limited_storage.go

@@ -27,7 +27,10 @@ func (s *RateLimitedStorage) wait() {
 }
 
 func (s *RateLimitedStorage) signal() {
-	<-s.semaphore
+	select {
+	case <-s.semaphore:
+	default:
+	}
 }
 
 // See Storage for documentation.

container/concurrent/lrucache.go

@@ -51,8 +51,10 @@ func (p *concurrentLruCacheImp) Get(key string) (v interface{}, found bool) {
 }
 
 func (p *concurrentLruCacheImp) GetMultiple(keys []string) []CacheResult {
-	p.lock.RLock()
-	defer p.lock.RUnlock()
+	// the LRU cache get alters the cache. Therefore, we should
+	// acquire the write lock and not the read lock
+	p.lock.Lock()
+	defer p.lock.Unlock()
 
 	res := make([]CacheResult, len(keys))
 

database/sqltypes/sqltypes.go

@@ -253,6 +253,7 @@ func BuildValue(goval interface{}) (v Value, err error) {
 	}
 	return v, nil
 }
+
 // ConverAssignRowNullable is the same as ConvertAssignRow except that it allows
 // nil as a value for the row or any of the row values. In thoses cases, the
 // corresponding values are ignored.
@@ -281,6 +282,7 @@ func ConvertAssignRowNullable(row []Value, dest ...interface{}) error {
 
 	return nil
 }
+
 // ConvertAssignRow copies a row of values in the list of destinations.  An
 // error is returned if any one of the row's element coping is done between
 // incompatible value and dest types.  The list of destinations must contain

gocheck2/checkers.go

@@ -99,6 +99,45 @@ func (b *bytesEquals) Info() *CheckerInfo {
 //     c.Assert(nil, DeepEquals, []byte("")) // fails
 var BytesEquals = &bytesEquals{}
 
+// -----------------------------------------------------------------------
+// AlmostEqual checker.
+// Meant to compare floats with some margin of error which might arrise
+// from rounding errors.
+
+type almostEqualChecker struct{}
+
+func (ae *almostEqualChecker) Info() *CheckerInfo {
+	return &CheckerInfo{
+		Name:   "AlmostEqual",
+		Params: []string{"obtained", "expected", "margin"},
+	}
+}
+
+func (ae *almostEqualChecker) Check(params []interface{}, names []string) (bool, string) {
+	if len(params) != 3 {
+		return false, "AlmostEqual takes exactly 3 arguments"
+	}
+	obtained, ok1 := params[0].(float64)
+	expected, ok2 := params[1].(float64)
+	margin, ok3 := params[2].(float64)
+
+	if !(ok1 && ok2 && ok3) {
+		return false, "All arguments to AlmostEqual must be float64"
+	}
+
+	if margin < 0 {
+		return false, "Margin must be non-negative"
+	}
+
+	if obtained < (expected-margin) || obtained > (expected+margin) {
+		return false, fmt.Sprintf("Obtained %f different from expected %f by more than %f margin",
+			obtained, expected, margin)
+	}
+	return true, ""
+}
+
+var AlmostEqual = &almostEqualChecker{}
+
 // -----------------------------------------------------------------------
 // HasKey checker.
 
@@ -161,7 +200,7 @@ func (c noErr) Check(params []interface{}, names []string) (bool, string) {
 
 func (c noErr) Info() *CheckerInfo {
 	return &CheckerInfo{
-		Name: "NoErr",
+		Name:   "NoErr",
 		Params: []string{"error"},
 	}
 }

gocheck2/checkers_test.go

@@ -56,7 +56,7 @@ func (s *CheckersSuite) TestNoErr(c *C) {
 	params := []interface{}{errors.New("1\n2\n3")}
 	text := params[0].(error).Error()
 	NoErr.Check(params, nil)
-	c.Assert(fmt.Sprintf("%#v", params[0]), Equals, "\n" + text)
+	c.Assert(fmt.Sprintf("%#v", params[0]), Equals, "\n"+text)
 }
 
 func (s *CheckersSuite) TestErrorMatches(c *C) {
@@ -65,3 +65,16 @@ func (s *CheckersSuite) TestErrorMatches(c *C) {
 	test(c, MultilineErrorMatches, false, "",
 		errors.Newf("Oh damn, this stinks"), "skinks")
 }
+
+func (s *CheckersSuite) TestAlmostEqual(c *C) {
+	// Test margins.
+	test(c, AlmostEqual, true, "", 5.0, 5.0, 0.0)
+	test(c, AlmostEqual, true, "", 5.0, 5.0, 0.1)
+	test(c, AlmostEqual, true, "", 5.0, 4.995, 0.01)
+	test(c, AlmostEqual, false, "Obtained 5.000000 different from expected 4.995000 by more than 0.001000 margin", 5.0, 4.995, 0.001)
+
+	// Test invalid args.
+	test(c, AlmostEqual, false, "AlmostEqual takes exactly 3 arguments", 5.0, 4.99)
+	test(c, AlmostEqual, false, "All arguments to AlmostEqual must be float64", "5.0", 5.0, 0.1)
+	test(c, AlmostEqual, false, "Margin must be non-negative", 5.0, 5.0, -0.1)
+}

memcache/base_shard_manager.go

@@ -1,6 +1,7 @@
 package memcache
 
 import (
+	"expvar"
 	"sync"
 
 	"github.com/dropbox/godropbox/container/set"
@@ -16,6 +17,13 @@ const (
 	WarmUpServer    = MemcachedState(4)
 )
 
+var (
+	// Counters for number of connections that succeeded / errored / were skipped, by address.
+	connOkByAddr      = expvar.NewMap("ShardManagerConnOkByAddrCounter")
+	connErrByAddr     = expvar.NewMap("ShardManagerConnErrByAddrCounter")
+	connSkippedByAddr = expvar.NewMap("ShardManagerConnSkippedByAddrCounter")
+)
+
 type ShardState struct {
 	Address string
 	State   MemcachedState
@@ -107,16 +115,16 @@ func (m *BaseShardManager) GetShard(
 		return
 	}
 
-	if m.shardStates[shardId].State != ActiveServer {
+	state := m.shardStates[shardId]
+	if state.State != ActiveServer {
 		m.logInfo("Memcache shard ", shardId, " is not in active state.")
+		connSkippedByAddr.Add(state.Address, 1)
 		return
 	}
 
-	conn, err = m.pool.Get("tcp", m.shardStates[shardId].Address)
-	if err != nil {
-		m.logError(err)
-		conn = nil
-	}
+	entry := &ShardMapping{}
+	m.fillEntryWithConnection(state.Address, entry)
+	conn, err = entry.Connection, entry.ConnErr
 
 	return
 }
@@ -140,13 +148,9 @@ func (m *BaseShardManager) GetShardsForKeys(
 			if shardId != -1 {
 				state := m.shardStates[shardId]
 				if state.State == ActiveServer {
-					conn, err := m.pool.Get("tcp", state.Address)
-					if err != nil {
-						m.logError(err)
-						entry.ConnErr = err
-					} else {
-						entry.Connection = conn
-					}
+					m.fillEntryWithConnection(state.Address, entry)
+				} else {
+					connSkippedByAddr.Add(state.Address, 1)
 				}
 			}
 			entry.Keys = make([]string, 0, 1)
@@ -177,13 +181,9 @@ func (m *BaseShardManager) GetShardsForItems(
 			if shardId != -1 {
 				state := m.shardStates[shardId]
 				if state.State == ActiveServer {
-					conn, err := m.pool.Get("tcp", state.Address)
-					if err != nil {
-						m.logError(err)
-						entry.ConnErr = err
-					} else {
-						entry.Connection = conn
-					}
+					m.fillEntryWithConnection(state.Address, entry)
+				} else {
+					connSkippedByAddr.Add(state.Address, 1)
 				}
 			}
 			entry.Items = make([]*Item, 0, 1)
@@ -217,13 +217,7 @@ func (m *BaseShardManager) GetShardsForSentinels(
 					state.State == WriteOnlyServer ||
 					state.State == WarmUpServer {
 
-					conn, err := m.pool.Get("tcp", state.Address)
-					if err != nil {
-						m.logError(err)
-						entry.ConnErr = err
-					} else {
-						entry.Connection = conn
-					}
+					m.fillEntryWithConnection(state.Address, entry)
 
 					// During WARM_UP state, we do try to write sentinels to
 					// memcache but any failures are ignored. We run memcache
@@ -232,6 +226,8 @@ func (m *BaseShardManager) GetShardsForSentinels(
 					if state.State == WarmUpServer {
 						entry.WarmingUp = true
 					}
+				} else {
+					connSkippedByAddr.Add(state.Address, 1)
 				}
 			}
 			entry.Items = make([]*Item, 0, 1)
@@ -261,3 +257,15 @@ func (m *BaseShardManager) GetAllShards() map[int]net2.ManagedConn {
 
 	return results
 }
+
+func (m *BaseShardManager) fillEntryWithConnection(address string, entry *ShardMapping) {
+	conn, err := m.pool.Get("tcp", address)
+	if err != nil {
+		m.logError(err)
+		connErrByAddr.Add(address, 1)
+		entry.ConnErr = err
+	} else {
+		connOkByAddr.Add(address, 1)
+		entry.Connection = conn
+	}
+}