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
|
package election
import (
"github.com/chrislusf/seaweedfs/weed/pb"
"math"
"sync"
"time"
)
type ClusterNode struct {
Address pb.ServerAddress
Version string
counter int
createdTs time.Time
}
type Leaders struct {
leaders [3]pb.ServerAddress
}
type Cluster struct {
nodes map[pb.ServerAddress]*ClusterNode
nodesLock sync.RWMutex
leaders *Leaders
}
func NewCluster() *Cluster {
return &Cluster{
nodes: make(map[pb.ServerAddress]*ClusterNode),
leaders: &Leaders{},
}
}
func (cluster *Cluster) AddClusterNode(nodeType string, address pb.ServerAddress, version string) {
switch nodeType {
case "filer":
cluster.nodesLock.Lock()
defer cluster.nodesLock.Unlock()
if existingNode, found := cluster.nodes[address]; found {
existingNode.counter++
return
}
cluster.nodes[address] = &ClusterNode{
Address: address,
Version: version,
counter: 1,
createdTs: time.Now(),
}
cluster.ensureLeader(true, address)
case "master":
}
}
func (cluster *Cluster) RemoveClusterNode(nodeType string, address pb.ServerAddress) {
switch nodeType {
case "filer":
cluster.nodesLock.Lock()
defer cluster.nodesLock.Unlock()
if existingNode, found := cluster.nodes[address]; !found {
return
} else {
existingNode.counter--
if existingNode.counter <= 0 {
delete(cluster.nodes, address)
cluster.ensureLeader(false, address)
}
}
case "master":
}
}
func (cluster *Cluster) ListClusterNode(nodeType string) (nodes []*ClusterNode) {
switch nodeType {
case "filer":
cluster.nodesLock.RLock()
defer cluster.nodesLock.RUnlock()
for _, node := range cluster.nodes {
nodes = append(nodes, node)
}
case "master":
}
return
}
func (cluster *Cluster) ensureLeader(isAdd bool, address pb.ServerAddress) {
if isAdd {
if cluster.leaders.addLeaderIfVacant(address) {
// has added the address as one leader
}
} else {
if cluster.leaders.removeLeaderIfExists(address) {
// pick the freshest one, since it is less likely to go away
var shortestDuration int64 = math.MaxInt64
now := time.Now()
var candidateAddress pb.ServerAddress
for _, node := range cluster.nodes {
if cluster.leaders.isOneLeader(node.Address) {
continue
}
duration := now.Sub(node.createdTs).Nanoseconds()
if duration < shortestDuration {
shortestDuration = duration
candidateAddress = node.Address
}
}
if candidateAddress != "" {
cluster.leaders.addLeaderIfVacant(candidateAddress)
}
// removed the leader, and maybe added a new leader
}
}
}
func (leaders *Leaders) addLeaderIfVacant(address pb.ServerAddress) (hasChanged bool) {
if leaders.isOneLeader(address) {
return
}
for i := 0; i < len(leaders.leaders); i++ {
if leaders.leaders[i] == "" {
leaders.leaders[i] = address
hasChanged = true
return
}
}
return
}
func (leaders *Leaders) removeLeaderIfExists(address pb.ServerAddress) (hasChanged bool) {
if !leaders.isOneLeader(address) {
return
}
for i := 0; i < len(leaders.leaders); i++ {
if leaders.leaders[i] == address {
leaders.leaders[i] = ""
hasChanged = true
return
}
}
return
}
func (leaders *Leaders) isOneLeader(address pb.ServerAddress) bool {
for i := 0; i < len(leaders.leaders); i++ {
if leaders.leaders[i] == address {
return true
}
}
return false
}
func (leaders *Leaders) GetLeaders() (addresses []pb.ServerAddress) {
for i := 0; i < len(leaders.leaders); i++ {
if leaders.leaders[i] != "" {
addresses = append(addresses, leaders.leaders[i])
}
}
return
}
|
