Merge pull request #539 from youtube/suguwork

Misc fixes

go/ratelimiter/ratelimiter.go

@@ -0,0 +1,54 @@
+// Copyright 2012, Google Inc. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Package ratelimiter implements rate limiting functionality.
+package ratelimiter
+
+import (
+	"sync"
+	"time"
+)
+
+// RateLimiter was inspired by https://github.com/golang/go/wiki/RateLimiting.
+// However, the go example is not good for setting high qps limits because
+// it will cause the ticker to fire too often. Also, the ticker will continue
+// to fire when the system is idle. This new Ratelimiter achieves the same thing,
+// but by using just counters with no tickers or channels.
+type RateLimiter struct {
+	maxCount int
+	interval time.Duration
+
+	mu       sync.Mutex
+	curCount int
+	lastTime time.Time
+}
+
+// NewRateLimiter creates a new RateLimiter. maxCount is the max burst allowed
+// while interval specifies the duration for a burst. The effective rate limit is
+// equal to maxCount/interval. For example, if you want to a max QPS of 5000,
+// and want to limit bursts to no more than 500, you'd specify a maxCount of 500
+// and an interval of 100*time.Millilsecond.
+func NewRateLimiter(maxCount int, interval time.Duration) *RateLimiter {
+	return &RateLimiter{
+		maxCount: maxCount,
+		interval: interval,
+	}
+}
+
+// Allow returns true if a request is within the rate limit norms.
+// Otherwise, it returns false.
+func (rl *RateLimiter) Allow() bool {
+	rl.mu.Lock()
+	defer rl.mu.Unlock()
+	if time.Now().Sub(rl.lastTime) < rl.interval {
+		if rl.curCount > 0 {
+			rl.curCount--
+			return true
+		}
+		return false
+	}
+	rl.curCount = rl.maxCount - 1
+	rl.lastTime = time.Now()
+	return true
+}

go/ratelimiter/ratelimiter_test.go

@@ -0,0 +1,60 @@
+// Copyright 2012, Google Inc. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package ratelimiter
+
+import (
+	"testing"
+	"time"
+)
+
+func TestLimiter1(t *testing.T) {
+	rl := NewRateLimiter(1, 10*time.Millisecond)
+	var result bool
+	result = rl.Allow()
+	if !result {
+		t.Error("Allow: false, want true")
+	}
+	result = rl.Allow()
+	if result {
+		t.Error("Allow: true, want false")
+	}
+
+	time.Sleep(11 * time.Millisecond)
+	result = rl.Allow()
+	if !result {
+		t.Error("Allow: false, want true")
+	}
+	result = rl.Allow()
+	if result {
+		t.Error("Allow: true, want false")
+	}
+}
+
+func TestLimiter2(t *testing.T) {
+	rl := NewRateLimiter(2, 10*time.Millisecond)
+	var result bool
+	for i := 0; i < 2; i++ {
+		result = rl.Allow()
+		if !result {
+			t.Errorf("Allow(%d): false, want true", i)
+		}
+	}
+	result = rl.Allow()
+	if result {
+		t.Error("Allow: true, want false")
+	}
+
+	time.Sleep(11 * time.Millisecond)
+	for i := 0; i < 2; i++ {
+		result = rl.Allow()
+		if !result {
+			t.Errorf("Allow(%d): false, want true", i)
+		}
+	}
+	result = rl.Allow()
+	if result {
+		t.Error("Allow: true, want false")
+	}
+}

go/vt/client/client.go

@@ -171,7 +171,7 @@ func makeBindVars(args []driver.Value) map[string]interface{} {
 	}
 	bv := make(map[string]interface{}, len(args))
 	for i, v := range args {
-		bv[fmt.Sprintf("v%d", i)] = v
+		bv[fmt.Sprintf("v%d", i+1)] = v
 	}
 	return bv
 }

go/vt/client/fakeserver_test.go

@@ -153,7 +153,7 @@ var execMap = map[string]struct {
 		execQuery: &proto.Query{
 			Sql: "request1",
 			BindVariables: map[string]interface{}{
-				"v0": int64(0),
+				"v1": int64(0),
 			},
 			TabletType: topo.TYPE_RDONLY,
 			Session:    nil,

go/vt/tabletserver/queryctl.go

@@ -50,7 +50,6 @@ func init() {
 	flag.StringVar(&qsConfig.RowCache.Binary, "rowcache-bin", DefaultQsConfig.RowCache.Binary, "rowcache binary file")
 	flag.IntVar(&qsConfig.RowCache.Memory, "rowcache-memory", DefaultQsConfig.RowCache.Memory, "rowcache max memory usage in MB")
 	flag.StringVar(&qsConfig.RowCache.Socket, "rowcache-socket", DefaultQsConfig.RowCache.Socket, "socket filename hint: a unique filename will be generated based on this input")
-	flag.IntVar(&qsConfig.RowCache.TcpPort, "rowcache-port", DefaultQsConfig.RowCache.TcpPort, "DEPRECATED")
 	flag.IntVar(&qsConfig.RowCache.Connections, "rowcache-connections", DefaultQsConfig.RowCache.Connections, "rowcache max simultaneous connections")
 	flag.IntVar(&qsConfig.RowCache.Threads, "rowcache-threads", DefaultQsConfig.RowCache.Threads, "rowcache number of threads")
 	flag.BoolVar(&qsConfig.RowCache.LockPaged, "rowcache-lock-paged", DefaultQsConfig.RowCache.LockPaged, "whether rowcache locks down paged memory")
@@ -61,7 +60,6 @@ type RowCacheConfig struct {
 	Binary      string
 	Memory      int
 	Socket      string
-	TcpPort     int
 	Connections int
 	Threads     int
 	LockPaged   bool
@@ -132,7 +130,7 @@ var DefaultQsConfig = Config{
 	TxPoolTimeout:      1,
 	IdleTimeout:        30 * 60,
 	StreamBufferSize:   32 * 1024,
-	RowCache:           RowCacheConfig{Memory: -1, TcpPort: -1, Connections: -1, Threads: -1},
+	RowCache:           RowCacheConfig{Memory: -1, Connections: -1, Threads: -1},
 	SpotCheckRatio:     0,
 	StrictMode:         true,
 	StrictTableAcl:     false,