refactor: segment server

.gitignore

@@ -1,10 +1,3 @@
-
-# configuration file
-./config/*.json
-config/*.yaml
-config/cluster/*.yaml
-./config/*.toml
-
 # macOS
 .DS_Store
 
@@ -15,16 +8,24 @@ config/cluster/*.yaml
 #!.vscode/launch.json
 #!.vscode/extensions.json
 #!.vscode/*.code-snippets
+__debug_bin
 
 # JetBrains
 .idea/
 
+# configuration file
+/config/*.json
+/config/*.toml
+/config/*.yaml
+/config/cluster/*.yaml
+/config/**/*.local.*
+
 # data dir of etcd
 etcd
 bin
-ø
+
 *.xml
 
 *coverage*
 
-.run
+.run

go.sum

@@ -77,7 +77,6 @@ github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA
 github.com/certifi/gocertifi v0.0.0-20191021191039-0944d244cd40/go.mod h1:sGbDF6GwGcLpkNXPUTkMRoywsNa/ol15pxFe6ERfguA=
 github.com/certifi/gocertifi v0.0.0-20200922220541-2c3bb06c6054 h1:uH66TXeswKn5PW5zdZ39xEwfS9an067BirqA+P4QaLI=
 github.com/certifi/gocertifi v0.0.0-20200922220541-2c3bb06c6054/go.mod h1:sGbDF6GwGcLpkNXPUTkMRoywsNa/ol15pxFe6ERfguA=
-github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
 github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
 github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+qY=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
@@ -106,7 +105,6 @@ github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkE
 github.com/coreos/etcd v3.3.13+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/go-semver v0.3.0 h1:wkHLiw0WNATZnSG7epLsujiMCgPAc9xhjJ4tgnAxmfM=
 github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
-github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e h1:Wf6HqHfScWJN9/ZjdUKyjop4mf3Qdd+1TvvltAvM3m8=
 github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4=
 github.com/coreos/go-systemd/v22 v22.3.2 h1:D9/bQk5vlXQFZ6Kwuu6zaiXJ9oTPe68++AzAJc1DzSI=
 github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
@@ -178,7 +176,6 @@ github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69
 github.com/golang-jwt/jwt v3.2.2+incompatible h1:IfV12K8xAKAnZqdXVzCZ+TOjboZ2keLg81eXfW3O+oY=
 github.com/golang-jwt/jwt v3.2.2+incompatible/go.mod h1:8pz2t5EyA70fFQQSrl6XZXzqecmYZeUEB8OUGHkxJ+I=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
-github.com/golang/glog v1.0.0/go.mod h1:EWib/APOK0SL3dFbYqvxE3UYd8E6s1ouQ7iEp/0LWV4=
 github.com/golang/groupcache v0.0.0-20190129154638-5b532d6fd5ef/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
 github.com/golang/groupcache v0.0.0-20191227052852-215e87163ea7/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
@@ -230,7 +227,6 @@ github.com/google/go-cmp v0.5.3/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.7 h1:81/ik6ipDQS2aGcBfIN5dHDB36BwrStyeAQquSYCV4o=
-github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8/DtOE=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/gofuzz v1.1.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
@@ -641,7 +637,6 @@ golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLdyRGr576XBO4/greRjx4P4O3yc=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
-golang.org/x/net v0.0.0-20211015210444-4f30a5c0130f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20211029224645-99673261e6eb/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20211209124913-491a49abca63/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd h1:O7DYs+zxREGLKzKoMQrtrEacpb0ZVXA5rIwylE2Xchk=
@@ -729,18 +724,15 @@ golang.org/x/sys v0.0.0-20210403161142-5e06dd20ab57/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20210603125802-9665404d3644/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210831042530-f4d43177bf5e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210927094055-39ccf1dd6fa6/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20211103235746-7861aae1554b/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220209214540-3681064d5158 h1:rm+CHSpPEEW2IsXUib1ThaHIjuBVZjxNgSKmBLFfD4c=
 golang.org/x/sys v0.0.0-20220209214540-3681064d5158/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
-golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -754,7 +746,6 @@ golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
-golang.org/x/time v0.0.0-20201208040808-7e3f01d25324/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20210220033141-f8bda1e9f3ba/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac h1:7zkz7BUtwNFFqcowJ+RIgu2MaV/MapERkDIy+mwPyjs=
 golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
@@ -893,7 +884,6 @@ google.golang.org/genproto v0.0.0-20210310155132-4ce2db91004e/go.mod h1:FWY/as6D
 google.golang.org/genproto v0.0.0-20210319143718-93e7006c17a6/go.mod h1:FWY/as6DDZQgahTzZj3fqbO1CbirC29ZNUFHwi0/+no=
 google.golang.org/genproto v0.0.0-20210402141018-6c239bbf2bb1/go.mod h1:9lPAdzaEmUacj36I+k7YKbEc5CXzPIeORRgDAUOu28A=
 google.golang.org/genproto v0.0.0-20210602131652-f16073e35f0c/go.mod h1:UODoCrxHCcBojKKwX1terBiRUaqAsFqJiF615XL43r0=
-google.golang.org/genproto v0.0.0-20220218161850-94dd64e39d7c/go.mod h1:kGP+zUP2Ddo0ayMi4YuN7C3WZyJvGLZRh8Z5wnAqvEI=
 google.golang.org/genproto v0.0.0-20220310185008-1973136f34c6 h1:FglFEfyj61zP3c6LgjmVHxYxZWXYul9oiS1EZqD5gLc=
 google.golang.org/genproto v0.0.0-20220310185008-1973136f34c6/go.mod h1:kGP+zUP2Ddo0ayMi4YuN7C3WZyJvGLZRh8Z5wnAqvEI=
 google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=

internal/raft/log/compaction.go

@@ -83,17 +83,17 @@ func (l *Log) Compact(i uint64) error {
 	return nil
 }
 
-func (w *WAL) suppressCompact(cb exeCallback) error {
+func (w *WAL) suppressCompact(cb executeCallback) error {
 	result := make(chan error, 1)
-	w.exec <- exeTask{
+	w.executec <- executeTask{
 		cb:     cb,
 		result: result,
 	}
 	return <-result
 }
 
 func (w *WAL) tryCompact(offset, last int64, nodeID vanus.ID, index, term uint64) {
-	w.exec <- exeTask{
+	w.executec <- executeTask{
 		cb: func() (compactTask, error) {
 			return compactTask{
 				offset: offset,
@@ -147,7 +147,7 @@ func (m *compactMeta) Range(cb meta.RangeCallback) error {
 var emptyMark = struct{}{}
 
 func (w WAL) run() {
-	for task := range w.exec {
+	for task := range w.executec {
 		ct, err := task.cb()
 		if task.result != nil {
 			if err != nil {
@@ -163,8 +163,8 @@ func (w WAL) run() {
 }
 
 func (w *WAL) runCompact() {
-	peroid := 30 * time.Second
-	ticker := time.NewTicker(peroid)
+	period := 30 * time.Second
+	ticker := time.NewTicker(period)
 	defer ticker.Stop()
 
 	var compacted int64

internal/raft/log/recovery.go

@@ -31,7 +31,9 @@ import (
 	"github.com/linkall-labs/vanus/observability/log"
 )
 
-func RecoverLogsAndWAL(walDir string, metaStore *meta.SyncStore, offsetStore *meta.AsyncStore) (map[vanus.ID]*Log, *WAL, error) {
+func RecoverLogsAndWAL(
+	walDir string, metaStore *meta.SyncStore, offsetStore *meta.AsyncStore,
+) (map[vanus.ID]*Log, *WAL, error) {
 	var compacted int64
 	if v, exist := metaStore.Load(walCompactKey); exist {
 		var ok bool

internal/raft/log/wal.go

@@ -24,10 +24,13 @@ import (
 	walog "github.com/linkall-labs/vanus/internal/store/wal"
 )
 
-var (
-	walCompactKey = []byte("wal/compact")
+const (
+	defaultExecuteTaskBufferSize = 256
+	defaultCompactTaskBufferSize = 256
 )
 
+var walCompactKey = []byte("wal/compact")
+
 type compactInfo struct {
 	index, term uint64
 }
@@ -38,10 +41,10 @@ type compactTask struct {
 	info         compactInfo
 }
 
-type exeCallback func() (compactTask, error)
+type executeCallback func() (compactTask, error)
 
-type exeTask struct {
-	cb     exeCallback
+type executeTask struct {
+	cb     executeCallback
 	result chan error
 }
 
@@ -51,7 +54,7 @@ type WAL struct {
 	metaStore *meta.SyncStore
 
 	barrier  *skiplist.SkipList
-	exec     chan exeTask
+	executec chan executeTask
 	compactc chan compactTask
 }
 
@@ -60,8 +63,8 @@ func newWAL(wal *walog.WAL, metaStore *meta.SyncStore) *WAL {
 		WAL:       wal,
 		metaStore: metaStore,
 		barrier:   skiplist.New(skiplist.Int64),
-		exec:      make(chan exeTask, 256),
-		compactc:  make(chan compactTask, 256),
+		executec:  make(chan executeTask, defaultExecuteTaskBufferSize),
+		compactc:  make(chan compactTask, defaultCompactTaskBufferSize),
 	}
 
 	go w.run()

internal/raft/transport/host.go

@@ -56,6 +56,7 @@ func NewHost(resolver Resolver, callback string) Host {
 func (h *host) Send(ctx context.Context, msg *raftpb.Message, to uint64, endpoint string) {
 	mux := h.resolveMultiplexer(ctx, to, endpoint)
 	if mux == nil {
+		// TODO(james.yin): report MsgUnreachable.
 		return
 	}
 	mux.Send(msg)
@@ -64,6 +65,7 @@ func (h *host) Send(ctx context.Context, msg *raftpb.Message, to uint64, endpoin
 func (h *host) Sendv(ctx context.Context, msgs []*raftpb.Message, to uint64, endpoint string) {
 	mux := h.resolveMultiplexer(ctx, to, endpoint)
 	if mux == nil {
+		// TODO(james.yin): report MsgUnreachable.
 		return
 	}
 	mux.Sendv(msgs)
@@ -81,21 +83,24 @@ func (h *host) resolveMultiplexer(ctx context.Context, to uint64, endpoint strin
 	}
 
 	if mux, ok := h.peers.Load(endpoint); ok {
-		return mux.(*peer)
+		p, _ := mux.(*peer)
+		return p
 	}
 	// TODO(james.yin): clean unused peer
 	p := newPeer(context.TODO(), endpoint, h.callback)
 	if mux, loaded := h.peers.LoadOrStore(endpoint, p); loaded {
 		defer p.Close()
-		return mux.(*peer)
+		p2, _ := mux.(*peer)
+		return p2
 	}
 	return p
 }
 
-// Receive implements Demultiplexer
+// Receive implements Demultiplexer.
 func (h *host) Receive(ctx context.Context, msg *raftpb.Message, endpoint string) error {
 	if receiver, ok := h.receivers.Load(msg.To); ok {
-		receiver.(Receiver).Receive(ctx, msg, msg.From, endpoint)
+		r, _ := receiver.(Receiver)
+		r.Receive(ctx, msg, msg.From, endpoint)
 	}
 	return nil
 }

internal/store/segment/block/index.go → internal/raft/transport/host_test.go

@@ -12,13 +12,4 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package block
-
-const (
-	v1IndexLength = 8 + 4
-)
-
-type index struct {
-	offset int64
-	length int32
-}
+package transport

internal/raft/transport/loopback_test.go

@@ -0,0 +1,15 @@
+// Copyright 2022 Linkall Inc.
+//
+// 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 transport

internal/raft/transport/multiplexing.go

@@ -15,16 +15,16 @@
 package transport
 
 import (
-	// standard libraries
+	// standard libraries.
 	"context"
 
-	// third-party libraries
+	// first-party libraries.
 	"github.com/linkall-labs/raft/raftpb"
 )
 
 type Multiplexer interface {
 	Send(msg *raftpb.Message)
-	Sendv(msg []*raftpb.Message)
+	Sendv(msgs []*raftpb.Message)
 }
 
 type Demultiplexer interface {

internal/raft/transport/multiplexing_test.go

@@ -0,0 +1,15 @@
+// Copyright 2022 Linkall Inc.
+//
+// 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 transport

internal/raft/transport/peer_test.go

@@ -0,0 +1,15 @@
+// Copyright 2022 Linkall Inc.
+//
+// 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 transport

internal/raft/transport/resolver.go

@@ -17,6 +17,8 @@ package transport
 import (
 	// standard libraries
 	"sync"
+
+	"github.com/linkall-labs/vanus/observability/log"
 )
 
 type Resolver interface {
@@ -41,6 +43,11 @@ func (r *SimpleResolver) Resolve(node uint64) string {
 }
 
 func (r *SimpleResolver) Register(node uint64, endpoint string) {
+	log.Info(nil, "Register raft node route.", map[string]interface{}{
+		"nodeID":   node,
+		"endpoint": endpoint,
+	})
+
 	r.Lock()
 	defer r.Unlock()
 	r.nodes[node] = endpoint