修改为龙舟
This commit is contained in:
115
fsm/main.go
115
fsm/main.go
@@ -3,11 +3,9 @@ package fsm
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/gob"
|
||||
"flag"
|
||||
"fmt"
|
||||
"fusenapi/utils/autoconfig"
|
||||
"log"
|
||||
"net"
|
||||
"os"
|
||||
"os/signal"
|
||||
"path/filepath"
|
||||
@@ -44,17 +42,14 @@ var addresses []string = []string{
|
||||
"localhost:5502",
|
||||
}
|
||||
|
||||
func StartNode(replicaID uint64, exampleShardID uint64, addr string, gdb *gorm.DB) *dragonboat.NodeHost {
|
||||
var shardID uint64 = 128
|
||||
|
||||
func StartNode(ServerID uint64, serverconfigs []*autoconfig.ConfigServer, gdb *gorm.DB) *SharedState {
|
||||
|
||||
// addr := "localhost"
|
||||
|
||||
// addr = fmt.Sprintf("%s:%d", addr, port)
|
||||
|
||||
flag.Parse()
|
||||
if len(addr) == 0 && replicaID != 1 && replicaID != 2 && replicaID != 3 {
|
||||
fmt.Fprintf(os.Stderr, "node id must be 1, 2 or 3 when address is not specified\n")
|
||||
os.Exit(1)
|
||||
}
|
||||
// https://github.com/golang/go/issues/17393
|
||||
if runtime.GOOS == "darwin" {
|
||||
signal.Ignore(syscall.Signal(0xd))
|
||||
@@ -67,17 +62,18 @@ func StartNode(replicaID uint64, exampleShardID uint64, addr string, gdb *gorm.D
|
||||
// leave the initialMembers to be empty. we still populate the initialMembers
|
||||
// here for simplicity.
|
||||
|
||||
for idx, v := range addresses {
|
||||
for _, v := range serverconfigs {
|
||||
// key is the ReplicaID, ReplicaID is not allowed to be 0
|
||||
// value is the raft address
|
||||
initialMembers[uint64(idx+1)] = v
|
||||
|
||||
initialMembers[v.ReplicaId] = fmt.Sprintf("%s:%d", v.Host, v.Port-2000)
|
||||
}
|
||||
|
||||
// for simplicity, in this example program, addresses of all those 3 initial
|
||||
// raft members are hard coded. when address is not specified on the command
|
||||
// line, we assume the node being launched is an initial raft member.
|
||||
|
||||
var nodeAddr = initialMembers[uint64(replicaID)]
|
||||
var nodeAddr = initialMembers[ServerID]
|
||||
|
||||
fmt.Fprintf(os.Stdout, "node address: %s\n", nodeAddr)
|
||||
// change the log verbosity
|
||||
@@ -89,8 +85,8 @@ func StartNode(replicaID uint64, exampleShardID uint64, addr string, gdb *gorm.D
|
||||
// See GoDoc for all available options
|
||||
rc := config.Config{
|
||||
// ShardID and ReplicaID of the raft node
|
||||
ReplicaID: uint64(replicaID),
|
||||
ShardID: exampleShardID,
|
||||
ReplicaID: uint64(ServerID),
|
||||
ShardID: shardID,
|
||||
|
||||
ElectionRTT: 10,
|
||||
|
||||
@@ -102,9 +98,8 @@ func StartNode(replicaID uint64, exampleShardID uint64, addr string, gdb *gorm.D
|
||||
CompactionOverhead: 5,
|
||||
}
|
||||
datadir := filepath.Join(
|
||||
"example-data",
|
||||
"queue-data",
|
||||
fmt.Sprintf("node%d", replicaID))
|
||||
"shared-state",
|
||||
fmt.Sprintf("node%d", ServerID))
|
||||
|
||||
nhc := config.NodeHostConfig{
|
||||
|
||||
@@ -124,95 +119,19 @@ func StartNode(replicaID uint64, exampleShardID uint64, addr string, gdb *gorm.D
|
||||
panic(err)
|
||||
}
|
||||
|
||||
if err := nh.StartReplica(initialMembers, false, New, rc); err != nil {
|
||||
if err := nh.StartReplica(initialMembers, false, NewFsStateMachine, rc); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "failed to add cluster, %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
return nh
|
||||
|
||||
}
|
||||
|
||||
// StartNode 启动节点
|
||||
func StartNode1(ServerID string, RaftBind string, serverconfigs []*autoconfig.ConfigServer, gdb *gorm.DB) *FsStateMachine {
|
||||
|
||||
fsm := &FsStateMachine{
|
||||
store: make(map[int64]*UserState),
|
||||
|
||||
gdb: gdb,
|
||||
ss := &SharedState{
|
||||
shardID: shardID,
|
||||
replicaID: ServerID,
|
||||
nh: nh,
|
||||
}
|
||||
|
||||
var retainSnapshotCount = 2
|
||||
// var ServerID string = "fs1"
|
||||
// var RaftBind string = "localhost:5500"
|
||||
var RaftDir string = fmt.Sprintf("/tmp/raftdir/%s", ServerID)
|
||||
return ss
|
||||
|
||||
// Setup Raft configuration.
|
||||
config := raft.DefaultConfig()
|
||||
config.LocalID = raft.ServerID(ServerID)
|
||||
|
||||
// Setup Raft communication.
|
||||
addr, err := net.ResolveTCPAddr("tcp", RaftBind)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
transport, err := raft.NewTCPTransport(RaftBind, addr, 3, 30*time.Second, os.Stderr)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
// Create the snapshot store. This allows the Raft to truncate the log.
|
||||
snapshots, err := raft.NewFileSnapshotStore(RaftDir, retainSnapshotCount, os.Stderr)
|
||||
if err != nil {
|
||||
panic(fmt.Errorf("file snapshot store: %s", err))
|
||||
}
|
||||
|
||||
// Create the log store and stable store.
|
||||
logStore := raft.NewInmemStore()
|
||||
stableStore := raft.NewInmemStore()
|
||||
|
||||
// Create the Raft system.
|
||||
fsm.ra, err = raft.NewRaft(config, fsm, logStore, stableStore, snapshots, transport)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
var dup map[string]bool = make(map[string]bool)
|
||||
var rserver []raft.Server = []raft.Server{
|
||||
{
|
||||
Suffrage: raft.Voter,
|
||||
ID: config.LocalID,
|
||||
Address: transport.LocalAddr(),
|
||||
},
|
||||
}
|
||||
dup[string(config.LocalID)] = true
|
||||
dup[string("backend")] = true
|
||||
dup[string("product-model")] = true
|
||||
dup[string("product-template")] = true
|
||||
|
||||
for _, cfg := range serverconfigs {
|
||||
if _, ok := dup[cfg.Name]; !ok {
|
||||
dup[cfg.Name] = true
|
||||
rserver = append(rserver, raft.Server{
|
||||
Suffrage: raft.Voter,
|
||||
ID: raft.ServerID(cfg.Name),
|
||||
Address: raft.ServerAddress(fmt.Sprintf("%s:%d", cfg.Host, cfg.Port-2000)),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
configuration := raft.Configuration{
|
||||
Servers: rserver,
|
||||
}
|
||||
|
||||
fu := fsm.ra.BootstrapCluster(configuration)
|
||||
if err := fu.Error(); err != nil {
|
||||
log.Println(err)
|
||||
}
|
||||
|
||||
waitForCluster(fsm.ra)
|
||||
|
||||
return fsm
|
||||
}
|
||||
|
||||
// func JoinCluster(ServerID string, LeaderAddress string, RaftBind string, gdb *gorm.DB) *StateCluster {
|
||||
|
||||
Reference in New Issue
Block a user