1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
|
package dex
import (
"fmt"
"sync"
"github.com/dexon-foundation/dexon/p2p/discover"
)
type nodeInfo struct {
info *notaryNodeInfo
added bool
}
func (n *nodeInfo) NewNode() *discover.Node {
return discover.NewNode(n.info.ID, n.info.IP, n.info.UDP, n.info.TCP)
}
type notarySet struct {
round uint64
m map[string]*nodeInfo
lock sync.RWMutex
}
func newNotarySet(round uint64, s map[string]struct{}) *notarySet {
m := make(map[string]*nodeInfo)
for nodeID := range s {
m[nodeID] = &nodeInfo{}
}
return ¬arySet{
round: round,
m: m,
}
}
// Call this function when the notaryNodeInfoMsg is received.
func (n *notarySet) AddInfo(info *notaryNodeInfo) error {
n.lock.Lock()
defer n.lock.Unlock()
// check round
if info.Round != n.round {
return fmt.Errorf("invalid round")
}
nInfo, ok := n.m[info.ID.String()]
if !ok {
return fmt.Errorf("not in notary set")
}
// if the info exists check timstamp
if nInfo.info != nil {
if nInfo.info.Timestamp > info.Timestamp {
return fmt.Errorf("old msg")
}
}
nInfo.info = info
return nil
}
// MarkAdded mark the notary node as added
// to prevent duplcate addition in the future.
func (n *notarySet) MarkAdded(nodeID string) {
if info, ok := n.m[nodeID]; ok {
info.added = true
}
}
// Return all nodes
func (n *notarySet) Nodes() []*discover.Node {
n.lock.RLock()
defer n.lock.RUnlock()
list := make([]*discover.Node, 0, len(n.m))
for _, info := range n.m {
list = append(list, info.NewNode())
}
return list
}
// Return nodes that need to be added to p2p server as notary node.
func (n *notarySet) NodesToAdd() []*discover.Node {
n.lock.RLock()
defer n.lock.RUnlock()
var list []*discover.Node
for _, info := range n.m {
// craete a new discover.Node
if !info.added {
list = append(list, info.NewNode())
}
}
return list
}
type notarySetManager struct {
m map[uint64]*notarySet
lock sync.RWMutex
queued map[uint64]map[string]*notaryNodeInfo
round uint64 // biggest round of managed notary sets
newNotaryNodeInfoCh chan *notaryNodeInfo
}
func newNotarySetManager(
newNotaryNodeInfoCh chan *notaryNodeInfo) *notarySetManager {
return ¬arySetManager{
m: make(map[uint64]*notarySet),
queued: make(map[uint64]map[string]*notaryNodeInfo),
newNotaryNodeInfoCh: newNotaryNodeInfoCh,
}
}
// Register injects a new notary set into the manager and
// processes the queued info.
func (n *notarySetManager) Register(r uint64, s *notarySet) {
n.lock.Lock()
defer n.lock.Unlock()
if r > n.round {
n.round = r
}
n.m[r] = s
n.processQueuedInfo()
}
// Unregister removes the notary set of the given round.
func (n *notarySetManager) Unregister(r uint64) {
n.lock.Lock()
defer n.lock.Unlock()
delete(n.m, r)
}
// Round returns the notary set of the given round.
func (n *notarySetManager) Round(r uint64) (*notarySet, bool) {
n.lock.RLock()
defer n.lock.RUnlock()
s, ok := n.m[r]
return s, ok
}
// Before returns all the notary sets that before the given round.
func (n *notarySetManager) Before(r uint64) []*notarySet {
n.lock.RLock()
defer n.lock.RUnlock()
var list []*notarySet
for round, s := range n.m {
if round < r {
list = append(list, s)
}
}
return list
}
// TryAddInfo associates the given info to the notary set if the notary set is
// managed by the manager.
// If the notary node info is belong to future notary set, queue the info.
func (n *notarySetManager) TryAddInfo(info *notaryNodeInfo) {
n.lock.Lock()
defer n.lock.Unlock()
n.tryAddInfo(info)
}
// This function is extract for calling without lock.
// Make sure the caller already accquired the lock.
func (n *notarySetManager) tryAddInfo(info *notaryNodeInfo) {
if info.Round > n.round {
if q, ok := n.queued[info.Round]; ok {
q[info.Hash().String()] = info
return
}
n.queued[info.Round] = map[string]*notaryNodeInfo{
info.Hash().String(): info,
}
return
}
s, ok := n.Round(info.Round)
if !ok {
return
}
s.AddInfo(info)
// TODO(sonic): handle timeout
n.newNotaryNodeInfoCh <- info
}
func (n *notarySetManager) processQueuedInfo() {
n.lock.Lock()
defer n.lock.Unlock()
if q, ok := n.queued[n.round]; ok {
for _, info := range q {
n.tryAddInfo(info)
}
}
// Clear queue infos before current round.
for round := range n.queued {
if round <= n.round {
delete(n.queued, round)
}
}
}
|
