Merge branch 'kv-engine' into 'master'

Kv engine and handle the message which timeout too much to avoid blocking too long

See merge request paas/nsqserver!43

apps/nsq_data_tool/tool.go

@@ -79,7 +79,7 @@ func checkTopicStats() {
 		log.Printf("error: %v", err.Error())
 		return
 	}
-	producers, _, err := ci.GetNSQDStats(srcnodes, "", "", true)
+	producers, _, err := ci.GetNSQDStatsWithClients(srcnodes, "", "", true)
 	if err != nil {
 		log.Printf("error: %v", err.Error())
 		return

apps/nsqd/nsqd.go

@@ -182,6 +182,11 @@ func nsqdFlagSet(opts *nsqd.Options) *flag.FlagSet {
 	flagSet.Int("queue-write-buffer-size", int(opts.QueueWriteBufferSize), "the write buffer size for topic disk queue file")
 	flagSet.Int("pub-queue-size", int(opts.PubQueueSize), "the pub queue size for topic")
 	flagSet.Int("sleepms-between-log-sync-pull", int(opts.SleepMsBetweenLogSyncPull), "the sleep ms between each log sync pull")
+
+	flagSet.Bool("kv-enabled", opts.KVEnabled, "enable the kv topic")
+	flagSet.Int("kv-block-cache", int(opts.KVBlockCache), "kv engine block cache")
+	flagSet.Int("kv-write-buffer-size", int(opts.KVWriteBufferSize), "kv engine write buffer size")
+	flagSet.Int("kv-max-write-buffer-number", int(opts.KVMaxWriteBufferNumber), "kv max write buffer number")
 	return flagSet
 }
 

bench/multi_bench/multi_bench.go

@@ -17,7 +17,7 @@ import (
 	"time"
 
 	"github.com/absolute8511/glog"
-	"github.com/spaolacci/murmur3"
+	"github.com/twmb/murmur3"
 	"github.com/youzan/go-nsq"
 	"github.com/youzan/nsq/internal/app"
 	"github.com/youzan/nsq/internal/clusterinfo"

consistence/commitlog.go

@@ -79,6 +79,13 @@ type CommitLogData struct {
 	MsgNum int32
 }
 
+func (cl *CommitLogData) dqSeekCnt() int64 {
+	if cl.MsgCnt > 0 {
+		return cl.MsgCnt - 1
+	}
+	return 0
+}
+
 func GetLogDataSize() int {
 	return logDataSize
 }
@@ -758,6 +765,35 @@ func (self *TopicCommitLogMgr) GetCurrentEnd() (int64, int64) {
 	return self.currentStart, int64(self.currentCount) * int64(GetLogDataSize())
 }
 
+func (self *TopicCommitLogMgr) GetMaxAvailableCleanOffset(fileIndex int64, fileOffset int64) (int64, error) {
+	self.Lock()
+	defer self.Unlock()
+	if fileIndex == self.currentStart {
+		if fileOffset/int64(GetLogDataSize())+int64(MIN_KEEP_LOG_ITEM) >= int64(self.currentCount) {
+			if int64(self.currentCount) <= int64(MIN_KEEP_LOG_ITEM) {
+				return fileOffset, ErrCommitLogCleanKeepMin
+			} else {
+				return (int64(self.currentCount-1) - int64(MIN_KEEP_LOG_ITEM)) * int64(GetLogDataSize()), nil
+			}
+		}
+	} else if fileIndex == self.currentStart-1 {
+		fName := getSegmentFilename(self.path, fileIndex)
+		stat, err := os.Stat(fName)
+		if err != nil {
+			return fileOffset, err
+		}
+		leftSize := stat.Size() - fileOffset
+		if leftSize < 0 {
+			return fileOffset, ErrCommitLogOffsetInvalid
+		}
+		if leftSize/int64(GetLogDataSize())+int64(self.currentCount) <= int64(MIN_KEEP_LOG_ITEM) {
+			noffset := stat.Size() - (int64(MIN_KEEP_LOG_ITEM)-int64(self.currentCount)+1)*int64(GetLogDataSize())
+			return noffset, nil
+		}
+	}
+	return fileOffset, nil
+}
+
 func (self *TopicCommitLogMgr) prepareCleanOldData(fileIndex int64, fileOffset int64) (LogStartInfo, int64, error) {
 	self.Lock()
 	defer self.Unlock()

consistence/data_placement_mgr.go

@@ -11,7 +11,7 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/spaolacci/murmur3"
+	"github.com/twmb/murmur3"
 
 	"github.com/youzan/nsq/nsqd"
 )

consistence/etcd_client.go

@@ -19,8 +19,8 @@ type EtcdClient struct {
 var etcdTransport client.CancelableTransport = &http.Transport{
 	Proxy: http.ProxyFromEnvironment,
 	Dial: (&net.Dialer{
-		Timeout:   30 * time.Second,
-		KeepAlive: 15 * time.Second,
+		Timeout:   10 * time.Second,
+		KeepAlive: 10 * time.Second,
 	}).Dial,
 	TLSHandshakeTimeout: 10 * time.Second,
 	WriteBufferSize:     1024,

consistence/leadership.go

@@ -58,7 +58,7 @@ type NsqLookupdNodeInfo struct {
 	Epoch    EpochType
 }
 
-func (self *NsqLookupdNodeInfo) GetID() string {
+func (self NsqLookupdNodeInfo) GetID() string {
 	return self.ID
 }