Skip to content

Commit 122e9c3

Browse files
author
Qi Xiao
committed
Add GetCurrentBP method based on DHT.FindNeighbor
1 parent 4548fb7 commit 122e9c3

5 files changed

Lines changed: 115 additions & 5 deletions

File tree

‎consistent/consistent.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,7 @@ func InitConsistent(storePath string, persistImpl Persistence, initBP bool) (c *
8989
Addr: "",
9090
PublicKey: kms.BP.PublicKey,
9191
Nonce: kms.BP.Nonce,
92+
Role: proto.Leader,
9293
}
9394
}
9495

‎proto/nodeinfo.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -103,6 +103,11 @@ func (id *NodeID) ToRawNodeID() *RawNodeID {
103103
return &RawNodeID{*idHash}
104104
}
105105

106+
// IsEmpty test if a nodeID is empty.
107+
func (id *NodeID) IsEmpty() bool {
108+
return id == nil || "" == string(*id)
109+
}
110+
106111
// InitNodeCryptoInfo generate Node asymmetric key pair and generate Node.NonceInfo
107112
// Node.ID = Node.NonceInfo.Hash
108113
func (node *Node) InitNodeCryptoInfo(timeThreshold time.Duration) (err error) {

‎proto/nodeinfo_test.go‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -112,3 +112,15 @@ func TestNodeID_ToRawNodeID(t *testing.T) {
112112
So(node.ToRawNodeID().String(), ShouldEqual, id)
113113
})
114114
}
115+
116+
func TestNodeID_IsEmpty(t *testing.T) {
117+
Convey("NodeID is empty", t, func() {
118+
var nodeID NodeID
119+
So(nodeID.IsEmpty(), ShouldBeTrue)
120+
var nodeIDPtr *NodeID
121+
So(nodeIDPtr.IsEmpty(), ShouldBeTrue)
122+
id := "00000000011a34cb8142780f692a4097d883aa2ac8a534a070a134f11bcca573"
123+
node := NodeID(id)
124+
So(node.IsEmpty(), ShouldBeFalse)
125+
})
126+
}

‎rpc/rpcutil.go‎

Lines changed: 77 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,9 @@ package rpc
1919
import (
2020
"context"
2121
"net/rpc"
22+
"sync"
23+
"math/rand"
24+
"errors"
2225

2326
"github.com/hashicorp/yamux"
2427
"gitlab.com/thunderdb/ThunderDB/crypto/kms"
@@ -27,6 +30,16 @@ import (
2730
"gitlab.com/thunderdb/ThunderDB/utils/log"
2831
)
2932

33+
var (
34+
// ErrNoChiefBlockProducerAvailable defines failure on find chief block producer.
35+
ErrNoChiefBlockProducerAvailable = errors.New("no chief block producer found")
36+
37+
// currentBP represents current chief block producer node.
38+
currentBP proto.NodeID
39+
// currentBPLock represents the chief block producer access lock.
40+
currentBPLock sync.Mutex
41+
)
42+
3043
// Caller is a wrapper for session pooling and RPC calling.
3144
type Caller struct {
3245
pool *SessionPool
@@ -84,7 +97,7 @@ func (c *Caller) CallNodeWithContext(
8497
return
8598
}
8699

87-
// GetNodeAddr tries best to get node addr
100+
// GetNodeAddr tries best to get node addr.
88101
func GetNodeAddr(id *proto.RawNodeID) (addr string, err error) {
89102
addr, err = route.GetNodeAddrCache(id)
90103
if err != nil {
@@ -120,7 +133,7 @@ func GetNodeAddr(id *proto.RawNodeID) (addr string, err error) {
120133
return
121134
}
122135

123-
// GetNodeInfo tries best to get node info
136+
// GetNodeInfo tries best to get node info.
124137
func GetNodeInfo(id *proto.RawNodeID) (nodeInfo *proto.Node, err error) {
125138
nodeInfo, err = kms.GetNodeInfo(proto.NodeID(id.String()))
126139
if err != nil {
@@ -163,7 +176,7 @@ func GetNodeInfo(id *proto.RawNodeID) (nodeInfo *proto.Node, err error) {
163176
return
164177
}
165178

166-
// PingBP Send DHT.Ping Request with Anonymous ETLS session
179+
// PingBP Send DHT.Ping Request with Anonymous ETLS session.
167180
func PingBP(node *proto.Node, BPNodeID proto.NodeID) (err error) {
168181
client := NewCaller()
169182

@@ -181,3 +194,64 @@ func PingBP(node *proto.Node, BPNodeID proto.NodeID) (err error) {
181194

182195
return
183196
}
197+
198+
// GetCurrentBP returns nearest hash distance block producer as current node chief block producer.
199+
func GetCurrentBP() (bpNodeID proto.NodeID, err error) {
200+
currentBPLock.Lock()
201+
defer currentBPLock.Unlock()
202+
203+
if !currentBP.IsEmpty() {
204+
bpNodeID = currentBP
205+
return
206+
}
207+
208+
var localNodeID proto.NodeID
209+
if localNodeID, err = kms.GetLocalNodeID(); err != nil {
210+
return
211+
}
212+
213+
// get random block producer first
214+
bpList := route.GetBPs()
215+
randomBP := bpList[rand.Intn(len(bpList))]
216+
217+
// call random block producer for nearest block producer node
218+
req := &proto.FindNeighborReq{
219+
NodeID: localNodeID,
220+
Roles: []proto.ServerRole{
221+
proto.Leader,
222+
proto.Follower,
223+
},
224+
Count: 1,
225+
}
226+
res := new(proto.FindNeighborResp)
227+
if err = NewCaller().CallNode(randomBP, "DHT.FindNeighbor", req, res); err != nil {
228+
return
229+
}
230+
231+
if len(res.Nodes) <= 0 {
232+
// node not found
233+
err = ErrNoChiefBlockProducerAvailable
234+
return
235+
}
236+
237+
if res.Nodes[0].Role != proto.Leader && res.Nodes[0].Role != proto.Follower {
238+
// not block producer
239+
err = ErrNoChiefBlockProducerAvailable
240+
return
241+
}
242+
243+
currentBP = res.Nodes[0].ID
244+
bpNodeID = currentBP
245+
246+
return
247+
}
248+
249+
// SetCurrentBP sets current node chief block producer.
250+
func SetCurrentBP(bpNodeID proto.NodeID) {
251+
currentBPLock.Lock()
252+
defer currentBPLock.Unlock()
253+
254+
currentBP = bpNodeID
255+
256+
return
257+
}

‎rpc/rpcutil_test.go‎

Lines changed: 20 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -44,8 +44,8 @@ func TestCaller_CallNode(t *testing.T) {
4444
defer os.Remove(publicKeyStore)
4545

4646
_, testFile, _, _ := runtime.Caller(0)
47-
confFile := filepath.Join(filepath.Dir(testFile), "../test/node_0/config.yaml")
48-
privateKeyPath := filepath.Join(filepath.Dir(testFile), "../test/node_0/private.key")
47+
confFile := filepath.Join(filepath.Dir(testFile), "../test/node_standalone/config.yaml")
48+
privateKeyPath := filepath.Join(filepath.Dir(testFile), "../test/node_standalone/private.key")
4949

5050
conf.GConf, _ = conf.LoadConfig(confFile)
5151
log.Debugf("GConf: %#v", conf.GConf)
@@ -134,5 +134,23 @@ func TestCaller_CallNode(t *testing.T) {
134134
}
135135
log.Debugf("respA2: %v", respA)
136136

137+
// test get current bp, should only be myself
138+
chiefBPNodeID, err := GetCurrentBP()
139+
if err != nil {
140+
log.Fatal(err)
141+
}
142+
log.Debugf("current chief bp is: %v", chiefBPNodeID)
143+
144+
// set another random node as block producer
145+
randomNode := proto.NodeID("00000000011a34cb8142780f692a4097d883aa2ac8a534a070a134f11bcca573")
146+
SetCurrentBP(randomNode)
147+
chiefBPNodeID, err = GetCurrentBP()
148+
if err != nil {
149+
log.Fatal(err)
150+
}
151+
if chiefBPNodeID != randomNode {
152+
log.Fatalf("SetCurrentBP does not works, set: %v, current: %v", randomNode, chiefBPNodeID)
153+
}
154+
137155
server.Stop()
138156
}

0 commit comments

Comments
 (0)