CovenantSQL · auxten · Feb 20, 2019 · Feb 19, 2019 · Feb 19, 2019 · Feb 19, 2019
diff --git a/cmd/cqld/adapter.go b/cmd/cqld/adapter.go
diff --git a/cmd/cqld/bench_test.go b/cmd/cqld/bench_test.go
@@ -235,7 +235,7 @@ func TestStartBP_CallRPC(t *testing.T) {
 	}
 }
 
-func BenchmarkKayakKVServer_GetAllNodeInfo(b *testing.B) {
+func BenchmarkKVServer_GetAllNodeInfo(b *testing.B) {
 	log.SetLevel(log.DebugLevel)
 	start3BPs()
 

diff --git a/cmd/cqld/bootstrap.go b/cmd/cqld/bootstrap.go
@@ -20,18 +20,13 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
-	"path/filepath"
 	"syscall"
 	"time"
 
 	"github.com/CovenantSQL/CovenantSQL/api"
-
 	bp "github.com/CovenantSQL/CovenantSQL/blockproducer"
 	"github.com/CovenantSQL/CovenantSQL/conf"
 	"github.com/CovenantSQL/CovenantSQL/crypto/kms"
-	"github.com/CovenantSQL/CovenantSQL/kayak"
-	kt "github.com/CovenantSQL/CovenantSQL/kayak/types"
-	kl "github.com/CovenantSQL/CovenantSQL/kayak/wal"
 	"github.com/CovenantSQL/CovenantSQL/proto"
 	"github.com/CovenantSQL/CovenantSQL/route"
 	"github.com/CovenantSQL/CovenantSQL/rpc"
@@ -42,18 +37,11 @@ import (
 )
 
 const (
-	kayakServiceName     = "Kayak"
-	kayakApplyMethodName = "Apply"
-	kayakFetchMethodName = "Fetch"
-	kayakWalFileName     = "kayak.ldb"
-	kayakPrepareTimeout  = 5 * time.Second
-	kayakCommitTimeout   = time.Minute
-	kayakLogWaitTimeout  = 10 * time.Second
+	dhtGossipServiceName = "DHTG"
+	dhtGossipTimeout     = time.Second * 20
 )
 
 func runNode(nodeID proto.NodeID, listenAddr string) (err error) {
-	rootPath := conf.GConf.WorkingRoot
-
 	genesis, err := loadGenesis()
 	if err != nil {
 		return
@@ -124,30 +112,29 @@ func runNode(nodeID proto.NodeID, listenAddr string) (err error) {
 			return err
 		}
 
-		// init kayak
-		log.Info("init kayak runtime")
-		var kayakRuntime *kayak.Runtime
-		if kayakRuntime, err = initKayakTwoPC(rootPath, thisNode, peers, st, server); err != nil {
-			log.WithError(err).Error("init kayak runtime failed")
-			return err
-		}
-
-		// init kayak and consistent
-		log.Info("init kayak and consistent runtime")
-		kvServer := &KayakKVServer{
-			Runtime:   kayakRuntime,
-			KVStorage: st,
-		}
+		// init dht node server
+		log.Info("init consistent runtime")
+		kvServer := NewKVServer(thisNode.ID, peers, st, dhtGossipTimeout)
 		dht, err := route.NewDHTService(conf.GConf.DHTFileName, kvServer, true)
 		if err != nil {
 			log.WithError(err).Error("init consistent hash failed")
 			return err
 		}
+		defer kvServer.Stop()
 
-		// set consistent handler to kayak storage
-		kvServer.KVStorage.consistent = dht.Consistent
+		// set consistent handler to local storage
+		kvServer.storage.consistent = dht.Consistent
 
-		// register service rpc
+		// register gossip service rpc
+		gossipService := NewGossipService(kvServer)
+		log.Info("register dht gossip service rpc")
+		err = server.RegisterService(route.DHTGossipRPCName, gossipService)
+		if err != nil {
+			log.WithError(err).Error("register dht gossip service failed")
+			return err
+		}
+
+		// register dht service rpc
 		log.Info("register dht service rpc")
 		err = server.RegisterService(route.DHTRPCName, dht)
 		if err != nil {
@@ -211,49 +198,8 @@ func createServer(privateKeyPath, pubKeyStorePath string, masterKey []byte, list
 	return
 }
 
-func initKayakTwoPC(rootDir string, node *proto.Node, peers *proto.Peers, h kt.Handler, server *rpc.Server) (runtime *kayak.Runtime, err error) {
-	// create kayak config
-	log.Info("create kayak config")
-
-	walPath := filepath.Join(rootDir, kayakWalFileName)
-
-	var logWal kt.Wal
-	if logWal, err = kl.NewLevelDBWal(walPath); err != nil {
-		err = errors.Wrap(err, "init kayak log pool failed")
-		return
-	}
-
-	config := &kt.RuntimeConfig{
-		Handler:          h,
-		PrepareThreshold: 1.0,
-		CommitThreshold:  1.0,
-		PrepareTimeout:   kayakPrepareTimeout,
-		CommitTimeout:    kayakCommitTimeout,
-		LogWaitTimeout:   kayakLogWaitTimeout,
-		Peers:            peers,
-		Wal:              logWal,
-		NodeID:           node.ID,
-		ServiceName:      kayakServiceName,
-		ApplyMethodName:  kayakApplyMethodName,
-		FetchMethodName:  kayakFetchMethodName,
-	}
-
-	// create kayak runtime
-	log.Info("init kayak runtime")
-	if runtime, err = kayak.NewRuntime(config); err != nil {
-		err = errors.Wrap(err, "init kayak runtime failed")
-		return
-	}
-
-	// register rpc service
-	if _, err = NewKayakService(server, kayakServiceName, runtime); err != nil {
-		err = errors.Wrap(err, "init kayak rpc service failed")
-		return
-	}
-
-	// init runtime
-	log.Info("start kayak runtime")
-	runtime.Start()
+func initDHTGossip() (err error) {
+	log.Info("init gossip service")
 
 	return
 }

diff --git a/cmd/cqld/gossip.go b/cmd/cqld/gossip.go
@@ -0,0 +1,43 @@
+/*
+ * Copyright 2019 The CovenantSQL Authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package main
+
+import "github.com/CovenantSQL/CovenantSQL/proto"
+
+// GossipRequest defines the gossip request payload.
+type GossipRequest struct {
+	proto.Envelope
+	Node *proto.Node
+	TTL  uint32
+}
+
+// GossipService defines the gossip service instance.
+type GossipService struct {
+	s *KVServer
+}
+
+// NewGossipService returns new gossip service.
+func NewGossipService(s *KVServer) *GossipService {
+	return &GossipService{
+		s: s,
+	}
+}
+
+// SetNode update current node info and broadcast node update request.
+func (s *GossipService) SetNode(req *GossipRequest, resp *interface{}) (err error) {
+	return s.s.SetNodeEx(req.Node, req.TTL, req.GetNodeID().ToNodeID())
+}
diff --git a/cmd/cqld/kayak.go b/cmd/cqld/kayak.go
diff --git a/cmd/cqld/main.go b/cmd/cqld/main.go
@@ -143,7 +143,7 @@ func main() {
 	defer utils.StopProfile()
 
 	if err := runNode(conf.GConf.ThisNodeID, conf.GConf.ListenAddr); err != nil {
-		log.WithError(err).Fatal("run kayak failed")
+		log.WithError(err).Fatal("run block producer node failed")
 	}
 
 	log.Info("server stopped")

diff --git a/route/acl.go b/route/acl.go
@@ -67,8 +67,8 @@ const (
 	DHTFindNeighbor
 	// DHTFindNode gets node info
 	DHTFindNode
-	// KayakCall is used by BP for data consistency
-	KayakCall
+	// DHTGSetNode is used by BP for dht data gossip
+	DHTGSetNode
 	// MetricUploadMetrics uploads node metrics
 	MetricUploadMetrics
 	// DBSQuery is used by client to read/write database
@@ -121,6 +121,8 @@ const (
 
 	// DHTRPCName defines the block producer dh-rpc service name
 	DHTRPCName = "DHT"
+	// DHTGossipRPCName defines the block producer dh-rpc gossip service name
+	DHTGossipRPCName = "DHTG"
 	// BlockProducerRPCName defines main chain rpc name
 	BlockProducerRPCName = "MCC"
 	// SQLChainRPCName defines the sql chain rpc name
@@ -138,10 +140,10 @@ func (s RemoteFunc) String() string {
 		return "DHT.FindNeighbor"
 	case DHTFindNode:
 		return "DHT.FindNode"
+	case DHTGSetNode:
+		return "DHTG.SetNode"
 	case MetricUploadMetrics:
 		return "Metric.UploadMetrics"
-	case KayakCall:
-		return "Kayak.Call"
 	case DBSQuery:
 		return "DBS.Query"
 	case DBSAck:
@@ -213,8 +215,8 @@ func IsPermitted(callerEnvelope *proto.Envelope, funcName RemoteFunc) (ok bool)
 		// DHT related
 		case DHTPing, DHTFindNode, DHTFindNeighbor, MetricUploadMetrics:
 			return true
-			// Kayak related
-		case KayakCall:
+			// DHTGSetNode is for block producer to update node info
+		case DHTGSetNode:
 			return false
 			// DBSDeploy
 		case DBSDeploy:

diff --git a/route/acl_test.go b/route/acl_test.go
@@ -51,8 +51,8 @@ func TestIsPermitted(t *testing.T) {
 		nodeID := proto.NodeID("0000")
 		testEnv := &proto.Envelope{NodeID: nodeID.ToRawNodeID()}
 		testAnonymous := &proto.Envelope{NodeID: kms.AnonymousRawNodeID}
-		So(IsPermitted(&proto.Envelope{NodeID: &conf.GConf.BP.RawNodeID}, KayakCall), ShouldBeTrue)
-		So(IsPermitted(testEnv, KayakCall), ShouldBeFalse)
+		So(IsPermitted(&proto.Envelope{NodeID: &conf.GConf.BP.RawNodeID}, DHTGSetNode), ShouldBeTrue)
+		So(IsPermitted(testEnv, DHTGSetNode), ShouldBeFalse)
 		So(IsPermitted(testEnv, DHTFindNode), ShouldBeTrue)
 		So(IsPermitted(testEnv, RemoteFunc(9999)), ShouldBeFalse)
 		So(IsPermitted(testAnonymous, DHTFindNode), ShouldBeFalse)