Repository: matrixorigin/matrixcube
Pull request #608 (2021-12-24 08:52:07)

Made LogDB less verbose in logging
**What type of PR is this?**

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

**Which issue(s) this PR fixes:**

Fixes #

**What this PR does / why we need it:**

Not Available

**Special notes for your reviewer:**

Not Available

**Additional documentation (e.g. design docs, usage docs, etc.):**

Not Available



Patch:
diff --git a/components/prophet/election/election_leadship.go b/components/prophet/election/election_leadship.go
index ef8f4e2..fac0733 100644
--- a/components/prophet/election/election_leadship.go
+++ b/components/prophet/election/election_leadship.go
@@ -23,6 +23,7 @@ import (
 	"github.com/matrixorigin/matrixcube/components/log"
 	"github.com/matrixorigin/matrixcube/components/prophet/option"
 	"github.com/matrixorigin/matrixcube/components/prophet/util"
+	"github.com/matrixorigin/matrixcube/util/stop"
 	"go.etcd.io/etcd/api/v3/mvccpb"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/client/v3/concurrency"
@@ -52,8 +53,8 @@ type Leadership struct {
 	ctx                          context.Context
 	allowCampaign                bool
 	becomeLeader, becomeFollower func(string) bool
-
-	logger *zap.Logger
+	stopper                      *stop.Stopper
+	logger                       *zap.Logger
 }
 
 func newLeadership(elector *elector,
@@ -74,6 +75,7 @@ func newLeadership(elector *elector,
 		tag:            tag,
 		logger: log.Adjust(logger).With(zap.String("tag", tag),
 			zap.String("purpose", purpose)),
+		stopper: stop.NewStopper("leadership"),
 	}
 }
 
@@ -113,7 +115,7 @@ func (ls *Leadership) Stop() error {
 	if lease != nil {
 		return lease.Close(ls.elector.client.Ctx())
 	}
-
+	ls.stopper.Stop()
 	return nil
 }
 
@@ -160,6 +162,7 @@ func (ls *Leadership) ElectionLoop(ctx context.Context) {
 		case <-ctx.Done():
 			ls.logger.Info("loop exit due to context done",
 				mainLoopFiled)
+			ls.stopper.Stop()
 			return
 		default:
 			currentLeader, rev, err := ls.CurrentLeader()
@@ -261,7 +264,7 @@ func (ls *Leadership) campaign() error {
 	// Start keepalive
 	ctx, cancel := context.WithCancel(ls.ctx)
 	defer cancel()
-	go l.keepAlive(ctx)
+	ls.stopper.RunTask(ctx, l.keepAlive)
 
 	var lock *concurrency.Mutex
 	if ls.elector.options.lockIfBecomeLeader {
diff --git a/components/prophet/member/member.go b/components/prophet/member/member.go
index 98e08e1..f521a30 100644
--- a/components/prophet/member/member.go
+++ b/components/prophet/member/member.go
@@ -17,14 +17,10 @@ package member
 import (
 	"context"
 	"sync/atomic"
-	"time"
 
-	"github.com/fagongzi/goetty"
 	"github.com/matrixorigin/matrixcube/components/log"
-	"github.com/matrixorigin/matrixcube/components/prophet/codec"
 	"github.com/matrixorigin/matrixcube/components/prophet/election"
 	"github.com/matrixorigin/matrixcube/components/prophet/pb/metapb"
-	"github.com/matrixorigin/matrixcube/util/buf"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/server/v3/embed"
 	"go.uber.org/zap"
@@ -32,16 +28,15 @@ import (
 
 // Member is used for the election related logic.
 type Member struct {
-	candidate   bool
-	etcd        *embed.Etcd
-	client      *clientv3.Client
-	elector     election.Elector
-	leadership  *election.Leadership
-	leader      atomic.Value   // stored as *metapb.Member
-	member      *metapb.Member // current prophet's info.
-	memberValue string
-	id          uint64 //etcd server id
-
+	candidate                            bool
+	etcd                                 *embed.Etcd
+	client                               *clientv3.Client
+	elector                              election.Elector
+	leadership                           *election.Leadership
+	leader                               atomic.Value   // stored as *metapb.Member
+	member                               *metapb.Member // current prophet's info.
+	memberValue                          string
+	id                                   uint64 //etcd server id
 	becomeLeaderFunc, becomeFollowerFunc func() error
 	logger                               *zap.Logger
 }
@@ -179,34 +174,3 @@ func (m *Member) Client() *clientv3.Client {
 func (m *Member) GetEtcdLeader() uint64 {
 	return m.etcd.Server.Lead()
 }
-
-func (m *Member) getLeaderClient(addr string) goetty.IOSession {
-	for {
-		leader := m.GetLeader()
-		if leader != nil {
-			conn, err := m.createLeaderClient(leader.Addr)
-			if err == nil {
-				m.logger.Info("create leader connection", zap.String("leader", leader.Addr))
-				return conn
-			}
-
-			m.logger.Error("fail to create leader connection",
-				zap.String("leader", leader.Addr),
-				zap.Error(err))
-		}
-
-		time.Sleep(time.Second)
-	}
-}
-
-func (m *Member) createLeaderClient(leader string) (goetty.IOSession, error) {
-	encoder, decoder := codec.NewClientCodec(10 * buf.MB)
-	conn := goetty.NewIOSession(goetty.WithCodec(encoder, decoder),
-		goetty.WithEnableAsyncWrite(16))
-	_, err := conn.Connect(leader, time.Second*3)
-	if err != nil {
-		return nil, err
-	}
-
-	return conn, nil
-}
diff --git a/components/prophet/prophet.go b/components/prophet/prophet.go
index a473280..2d048c3 100644
--- a/components/prophet/prophet.go
+++ b/components/prophet/prophet.go
@@ -35,7 +35,6 @@ import (
 	"github.com/matrixorigin/matrixcube/components/prophet/util"
 	"github.com/matrixorigin/matrixcube/components/prophet/util/typeutil"
 	"github.com/matrixorigin/matrixcube/config"
-	"github.com/matrixorigin/matrixcube/util/task"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/server/v3/embed"
 	"go.uber.org/zap"
@@ -76,7 +75,6 @@ type defaultProphet struct {
 	cancel         context.CancelFunc
 	cfg            *config.Config
 	persistOptions *pconfig.PersistOptions
-	runner         *task.Runner
 	stopOnce       sync.Once
 
 	// about leader election
@@ -153,7 +151,6 @@ func NewProphet(cfg *config.Config) Prophet {
 	p.elector = elector
 	p.etcd = etcd
 	p.member = member.NewMember(etcdClient, etcd, elector, cfg.Prophet.StorageNode, p.enableLeader, p.disableLeader, logger)
-	p.runner = task.NewRunner()
 	p.completeC = make(chan struct{})
 	p.jobMu.jobs = make(map[metapb.JobType]metapb.Job)
 	return p
@@ -193,8 +190,6 @@ func (p *defaultProphet) Stop() {
 		p.logger.Info("client stopped")
 		p.trans.Stop()
 		p.logger.Info("transport stopped")
-		p.runner.Stop()
-		p.logger.Info("runner stopped")
 		p.member.Stop()
 		p.logger.Info("member stopped")
 		p.cancel()
diff --git a/go.mod b/go.mod
index 5465b58..cf880dd 100644
--- a/go.mod
+++ b/go.mod
@@ -10,7 +10,7 @@ require (
 	github.com/cockroachdb/pebble v0.0.0-20210503173641-1387689d3d7c
 	github.com/coreos/go-semver v0.3.0
 	github.com/docker/go-units v0.4.0
-	github.com/fagongzi/goetty v1.11.3
+	github.com/fagongzi/goetty v1.13.0
 	github.com/fagongzi/util v0.0.0-20210923134909-bccc37b5040d
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang/mock v1.3.1
diff --git a/go.sum b/go.sum
index ad93df5..4a48ccb 100644
--- a/go.sum
+++ b/go.sum
@@ -118,8 +118,8 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.m
 github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/etcd-io/bbolt v1.3.3/go.mod h1:ZF2nL25h33cCyBtcyWeZ2/I3HQOfTP+0PIEvHjkjCrw=
-github.com/fagongzi/goetty v1.11.3 h1:N4UcOF/wORC8nU4OcSnW6kekO+CTyl3XXKIETSizCvY=
-github.com/fagongzi/goetty v1.11.3/go.mod h1:J5KOKdUzqQWinZWqULUPmr1l1bv27DxKKSlNc6jwSVg=
+github.com/fagongzi/goetty v1.13.0 h1:k1RT+32/O97zu370BhHk8sL/kSAqF2ebSY1Nj+x/dBg=
+github.com/fagongzi/goetty v1.13.0/go.mod h1:M1HA6OyQA3/4bV70W588Uek6sF7NxgxT/3gGhlf0yj0=
 github.com/fagongzi/util v0.0.0-20201116094402-221cc40c4593/go.mod h1:jYDIbpaqHXCCQ7QIDXRVfsQYAGKSNNb6N8BPTgdpcdE=
 github.com/fagongzi/util v0.0.0-20210923134909-bccc37b5040d h1:1pILVCatHj3eVo9i52dZyY4BwjTmSIeN+/hoJh8rD0Y=
 github.com/fagongzi/util v0.0.0-20210923134909-bccc37b5040d/go.mod h1:5cqSns2zMRcJeVGvAqeTrbXFqh5AqBFr5uVKP9T2kiE=
@@ -454,8 +454,6 @@ github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyC
 github.com/valyala/fasthttp v1.6.0/go.mod h1:FstJa9V+Pj9vQ7OJie2qMHdwemEDaDiSdBnvPM1Su9w=
 github.com/valyala/fasttemplate v1.0.1/go.mod h1:UQGH1tvbgY+Nz5t2n7tXsz52dQxojPUpymEIMZ47gx8=
 github.com/valyala/tcplisten v0.0.0-20161114210144-ceec8f93295a/go.mod h1:v3UYOV9WzVtRmSR+PDvWpU/qWl4Wa5LApYYX4ZtKbio=
-github.com/valyala/tcplisten v1.0.0 h1:rBHj/Xf+E1tRGZyWIWwJDiRY0zc1Js+CV5DqwacVSA8=
-github.com/valyala/tcplisten v1.0.0/go.mod h1:T0xQ8SeCZGxckz9qRXTfG43PvQ/mcWh7FwZEA7Ioqkc=
 github.com/xeipuuv/gojsonpointer v0.0.0-20180127040702-4e3ac2762d5f/go.mod h1:N2zxlSyiKSe5eX1tZViRH5QA0qijqEDrYZiPEAiq3wU=
 github.com/xeipuuv/gojsonreference v0.0.0-20180127040603-bd5ef7bd5415/go.mod h1:GwrjFmJcFw6At/Gs6z4yjiIwzuJ1/+UwLxMQDVQXShQ=
 github.com/xeipuuv/gojsonschema v1.2.0/go.mod h1:anYRn/JVcOK2ZgGU+IjEV4nwlhoK5sQluxsYJ78Id3Y=
@@ -639,8 +637,9 @@ golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210809222454-d867a43fc93e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210816074244-15123e1e1f71/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20211013075003-97ac67df715c h1:taxlMj0D/1sOAuv/CbSD+MMDof2vbyPTqz5FNYKpXt8=
 golang.org/x/sys v0.0.0-20211013075003-97ac67df715c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e h1:fLOSk5Q00efkSvAm+4xcoXD+RRmLmmulPn5I3Y9F2EM=
+golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 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=
diff --git a/logdb/logdb.go b/logdb/logdb.go
index 4775208..d48bc48 100644
--- a/logdb/logdb.go
+++ b/logdb/logdb.go
@@ -188,7 +188,7 @@ func (l *KVLogDB) GetAllSnapshots(shardID uint64) ([]raftpb.Snapshot, error) {
 
 func (l *KVLogDB) SaveRaftState(shardID uint64,
 	replicaID uint64, rd raft.Ready, ctx *WorkerContext) error {
-	if isEmptyRaftReady(rd) {
+	if IsEmptyRaftReady(rd) {
 		return nil
 	}
 
@@ -387,7 +387,7 @@ func (l *KVLogDB) getMaxIndex(shardID uint64, replicaID uint64) (uint64, error)
 	return binary.BigEndian.Uint64(v), nil
 }
 
-func isEmptyRaftReady(rd raft.Ready) bool {
+func IsEmptyRaftReady(rd raft.Ready) bool {
 	return raft.IsEmptyHardState(rd.HardState) &&
 		raft.IsEmptySnap(rd.Snapshot) &&
 		len(rd.Entries) == 0
diff --git a/raftstore/replica.go b/raftstore/replica.go
index d98c13b..5054fad 100644
--- a/raftstore/replica.go
+++ b/raftstore/replica.go
@@ -185,7 +185,7 @@ func newReplica(store *store, shard Shard, r Replica, reason string) (*replica,
 }
 
 // start() should return error
-func (pr *replica) start() {
+func (pr *replica) start(campaign bool) {
 	pr.logger.Info("begin to start replica")
 	pr.readStopper = stop.NewStopper(fmt.Sprintf("read-stopper[%d/%d/%d]",
 		pr.shardID, pr.replicaID, pr.replica.ContainerID),
@@ -225,7 +225,11 @@ func (pr *replica) start() {
 
 	pr.setStarted()
 	// If this shard has only one replica and I am the one, campaign directly.
-	if len(shard.Replicas) == 1 && shard.Replicas[0].ContainerID == pr.storeID {
+	if campaign {
+		pr.logger.Info("try to campaign",
+			log.ReasonField("restart"))
+		pr.addAction(action{actionType: campaignAction})
+	} else if len(shard.Replicas) == 1 && shard.Replicas[0].ContainerID == pr.storeID {
 		pr.logger.Info("try to campaign",
 			log.ReasonField("only self"))
 		pr.addAction(action{actionType: campaignAction})
diff --git a/raftstore/replica_apply_result.go b/raftstore/replica_apply_result.go
index 7f7ed0c..7e0e870 100644
--- a/raftstore/replica_apply_result.go
+++ b/raftstore/replica_apply_result.go
@@ -234,7 +234,7 @@ func (pr *replica) applySplit(result splitResult) {
 	reason := fmt.Sprintf("create by shard %d splitted", pr.shardID)
 	newReplicaCreator(pr.store).
 		withReason(reason).
-		withStartReplica(func(r *replica) {
+		withStartReplica(false, func(r *replica) {
 			r.stats.approximateKeys = estimatedKeys
 			r.stats.approximateSize = estimatedSize
 		}, func(r *replica) {
diff --git a/raftstore/replica_create.go b/raftstore/replica_create.go
index ee66c6c..ff038c3 100644
--- a/raftstore/replica_create.go
+++ b/raftstore/replica_create.go
@@ -40,6 +40,7 @@ type replicaCreator struct {
 	saveLog                           bool
 	reason                            string
 	startReplica                      bool
+	campaign                          bool
 	afterStartedFunc, beforeStartFunc func(*replica)
 	replicaRecordGetter               func(Shard) Replica
 	wc                                *logdb.WorkerContext
@@ -66,10 +67,12 @@ func (rc *replicaCreator) withReason(reason string) *replicaCreator {
 	return rc
 }
 
-func (rc *replicaCreator) withStartReplica(beforeStartFunc, afterStartedFunc func(*replica)) *replicaCreator {
+func (rc *replicaCreator) withStartReplica(campaign bool,
+	beforeStartFunc, afterStartedFunc func(*replica)) *replicaCreator {
 	rc.startReplica = true
 	rc.afterStartedFunc = afterStartedFunc
 	rc.beforeStartFunc = beforeStartFunc
+	rc.campaign = campaign
 	return rc
 }
 
@@ -207,7 +210,7 @@ func (rc *replicaCreator) maybeStartReplicas(shards []Shard, replicas []*replica
 		if rc.beforeStartFunc != nil {
 			rc.beforeStartFunc(pr)
 		}
-		pr.start()
+		pr.start(rc.campaign)
 		shard := shards[idx]
 		for _, p := range shard.Replicas {
 			rc.store.replicaRecords.Store(p.ID, p)
diff --git a/raftstore/replica_event_raft_ready.go b/raftstore/replica_event_raft_ready.go
index 2df61dc..f2be32f 100644
--- a/raftstore/replica_event_raft_ready.go
+++ b/raftstore/replica_event_raft_ready.go
@@ -137,6 +137,10 @@ func (pr *replica) appendEntries(rd raft.Ready) error {
 }
 
 func (pr *replica) saveRaftState(rd raft.Ready, wc *logdb.WorkerContext) error {
+	if logdb.IsEmptyRaftReady(rd) {
+		return nil
+	}
+
 	var startTime int64
 	if ce := pr.logger.Check(zap.DebugLevel,
 		"begin to save raft state"); ce != nil {
diff --git a/raftstore/store.go b/raftstore/store.go
index ffe4691..ccba658 100644
--- a/raftstore/store.go
+++ b/raftstore/store.go
@@ -496,7 +496,7 @@ func (s *store) startShards() {
 
 	newReplicaCreator(s).
 		withReason("restart").
-		withStartReplica(nil, func(r *replica) {
+		withStartReplica(true, nil, func(r *replica) {
 			if metadata, ok := localDestroyings[r.shardID]; ok {
 				r.startDestroyReplicaTask(metadata.LogIndex, metadata.Metadata.RemoveData, "restart")
 			}
diff --git a/raftstore/store_handler.go b/raftstore/store_handler.go
index 3e815d5..6b541d0 100644
--- a/raftstore/store_handler.go
+++ b/raftstore/store_handler.go
@@ -194,7 +194,7 @@ func (s *store) tryToCreateReplicate(msg meta.RaftMessage) bool {
 			msg.From.ID,
 			msg.From.ContainerID,
 			msg.From.Role.String())).
-		withStartReplica(nil, nil).
+		withStartReplica(false, nil, nil).
 		withReplicaRecordGetter(func(s Shard) Replica { return target }).
 		create([]Shard{
 			{
diff --git a/raftstore/store_route_handler.go b/raftstore/store_route_handler.go
index 89807a9..aab6c61 100644
--- a/raftstore/store_route_handler.go
+++ b/raftstore/store_route_handler.go
@@ -31,7 +31,7 @@ func (s *store) doDynamicallyCreate(shard Shard) bool {
 
 	newReplicaCreator(s).
 		withReason("event").
-		withStartReplica(nil, nil).
+		withStartReplica(false, nil, nil).
 		withSaveMetadata(true).
 		create([]Shard{shard})
 	return true
diff --git a/util/task/task.go b/util/task/task.go
deleted file mode 100644
index f1e28e6..0000000
--- a/util/task/task.go
+++ /dev/null
@@ -1,482 +0,0 @@
-// Copyright 2021 MatrixOrigin.
-//
-// 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,
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package task
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"sync"
-	"time"
-)
-
-var (
-	errUnavailable = errors.New("runner is unavailable")
-)
-
-var (
-	defaultWaitStoppedTimeout       = time.Minute
-	defaultQueueName                = "__default__"
-	batch                     int64 = 64
-
-	jobPool sync.Pool
-)
-
-type state int
-
-const (
-	running  = state(0)
-	stopping = state(1)
-	stopped  = state(2)
-)
-
-var (
-	// ErrJobCancelled error job cancelled
-	ErrJobCancelled = errors.New("Job cancelled")
-)
-
-// JobState is the job state
-type JobState int
-
-const (
-	// Pending job is wait to running
-	Pending = JobState(0)
-	// Running job is running
-	Running = JobState(1)
-	// Cancelling job is cancelling
-	Cancelling = JobState(2)
-	// Cancelled job is cancelled
-	Cancelled = JobState(3)
-	// Finished job is complete
-	Finished = JobState(4)
-	// Failed job is failed when execute
-	Failed = JobState(5)
-)
-
-func acquireJob() *Job {
-	v := jobPool.Get()
-	if v == nil {
-		return &Job{}
-	}
-	return v.(*Job)
-}
-
-func releaseJob(job *Job) {
-	job.reset()
-	jobPool.Put(job)
-}
-
-// Job is do for something with state
-type Job struct {
-	sync.RWMutex
-
-	desc        string
-	fun         func() error
-	state       JobState
-	result      interface{}
-	needRelease bool
-}
-
-func newJob(desc string, fun func() error) *Job {
-	job := acquireJob()
-	job.fun = fun
-	job.state = Pending
-	job.desc = desc
-	job.needRelease = true
-
-	return job
-}
-
-func (job *Job) reset() {
-	job.Lock()
-	job.fun = nil
-	job.state = Pending
-	job.desc = ""
-	job.result = nil
-	job.needRelease = true
-	job.Unlock()
-}
-
-// IsComplete return true means the job is complete.
-func (job *Job) IsComplete() bool {
-	return !job.IsNotComplete()
-}
-
-// IsNotComplete return true means the job is not complete.
-func (job *Job) IsNotComplete() bool {
-	job.RLock()
-	yes := job.state == Pending || job.state == Running || job.state == Cancelling
-	job.RUnlock()
-
-	return yes
-}
-
-// SetResult set result
-func (job *Job) SetResult(result interface{}) {
-	job.Lock()
-	job.result = result
-	job.Unlock()
-}
-
-// GetResult returns job result
-func (job *Job) GetResult() interface{} {
-	job.RLock()
-	r := job.result
-	job.RUnlock()
-	return r
-}
-
-// Cancel cancel the job
-func (job *Job) Cancel() {
-	job.Lock()
-	if job.state == Pending {
-		job.state = Cancelling
-	}
-	job.Unlock()
-}
-
-// IsRunning returns true if job state is Running
-func (job *Job) IsRunning() bool {
-	return job.isSpecState(Running)
-}
-
-// IsPending returns true if job state is Pending
-func (job *Job) IsPending() bool {
-	return job.isSpecState(Pending)
-}
-
-// IsFinished returns true if job state is Finished
-func (job *Job) IsFinished() bool {
-	return job.isSpecState(Finished)
-}
-
-// IsCancelling returns true if job state is Cancelling
-func (job *Job) IsCancelling() bool {
-	return job.isSpecState(Cancelling)
-}
-
-// IsCancelled returns true if job state is Cancelled
-func (job *Job) IsCancelled() bool {
-	return job.isSpecState(Cancelled)
-}
-
-// IsFailed returns true if job state is Failed
-func (job *Job) IsFailed() bool {
-	return job.isSpecState(Failed)
-}
-
-func (job *Job) isSpecState(spec JobState) bool {
-	job.RLock()
-	yes := job.state == spec
-	job.RUnlock()
-
-	return yes
-}
-
-func (job *Job) setState(state JobState) {
-	job.state = state
-}
-
-// Runner TODO
-type Runner struct {
-	sync.RWMutex
-
-	stop       sync.WaitGroup
-	stopC      chan struct{}
-	lastID     uint64
-	cancels    map[uint64]context.CancelFunc
-	state      state
-	namedQueue map[string]*Queue
-	tasks      map[string]bool
-}
-
-// NewRunner returns a task runner
-func NewRunner() *Runner {
-	t := &Runner{
-		stopC:      make(chan struct{}),
-		state:      running,
-		namedQueue: make(map[string]*Queue),
-		cancels:    make(map[uint64]context.CancelFunc),
-		tasks:      make(map[string]bool),
-	}
-
-	t.AddNamedWorker(defaultQueueName, func() {})
-	return t
-}
-
-// AddNamedWorker add a named worker, the named worker has uniq queue, so jobs are linear execution
-func (s *Runner) AddNamedWorker(name string, stopped func()) (uint64, error) {
-	s.Lock()
-	defer s.Unlock()
-
-	if s.state != running {
-		return 0, errUnavailable
-	}
-
-	if _, ok := s.namedQueue[name]; ok {
-		return 0, fmt.Errorf("%s worker already added", name)
-	}
-
-	id, ctx := s.allocCtxLocked()
-	q := NewWithContext(128, ctx)
-	s.namedQueue[name] = q
-	s.startWorkerLocked(ctx, name, q, stopped)
-	return id, nil
-}
-
-// IsNamedWorkerBusy returns true if named queue is not empty
-func (s *Runner) IsNamedWorkerBusy(worker string) bool {
-	s.RLock()
-	defer s.RUnlock()
-
-	return s.getNamedQueueLocked(worker).Len() > 0
-}
-
-func (s *Runner) startWorkerLocked(ctx context.Context, name string, q *Queue, stopped func()) {
-	s.doRunCancelableTaskLocked(ctx, name, func(ctx context.Context) {
-		jobs := make([]interface{}, batch)
-		if stopped != nil {
-			defer stopped()
-		}
-
-		for {
-			n, err := q.Get(batch, jobs)
-			if err != nil {
-				return
-			}
-
-			for i := int64(0); i < n; i++ {
-				job := jobs[i].(*Job)
-
-				job.Lock()
-
-				switch job.state {
-				case Pending:
-					job.setState(Running)
-					job.Unlock()
-					err := job.fun()
-					job.Lock()
-					if err != nil {
-						if err == ErrJobCancelled {
-							job.setState(Cancelled)
-						} else {
-							job.setState(Failed)
-						}
-					} else {
-						if job.state == Cancelling {
-							job.setState(Cancelled)
-						} else {
-							job.setState(Finished)
-						}
-					}
-				case Cancelling:
-					job.setState(Cancelled)
-				}
-
-				job.Unlock()
-
-				if job.needRelease {
-					releaseJob(job)
-				}
-			}
-		}
-	})
-}
-
-// RunJob run a job
-func (s *Runner) RunJob(desc string, task func() error) error {
-	return s.RunJobWithNamedWorker(desc, defaultQueueName, task)
-}
-
-// RunJobWithNamedWorker run a job in a named worker
-func (s *Runner) RunJobWithNamedWorker(desc, worker string, task func() error) error {
-	return s.RunJobWithNamedWorkerWithCB(desc, worker, task, nil)
-}
-
-// RunJobWithNamedWorkerWithCB run a job in a named worker
-func (s *Runner) RunJobWithNamedWorkerWithCB(desc, worker string, task func() error, cb func(*Job)) error {
-	s.Lock()
-
-	if s.state != running {
-		s.Unlock()
-		return errUnavailable
-	}
-
-	job := newJob(desc, task)
-	q := s.getNamedQueueLocked(worker)
-	if q == nil {
-		s.Unlock()
-		return fmt.Errorf("named worker %s is not exists", worker)
-	}
-
-	if cb != nil {
-		job.needRelease = false
-		cb(job)
-	}
-
-	q.Put(job)
-
-	s.Unlock()
-	return nil
-}
-
-// RunCancelableTask run a task that can be cancelled
-// Example:
-// err := s.RunCancelableTask(func(ctx context.Context) {
-// 	select {
-// 	case <-ctx.Done():
-// 	// cancelled
-// 	case <-time.After(time.Second):
-// 		// do something
-// 	}
-// })
-// if err != nil {
-// 	// hanle error
-// 	return
-// }
-func (s *Runner) RunCancelableTask(name string, task func(context.Context)) (uint64, error) {
-	s.Lock()
-	defer s.Unlock()
-
-	if s.state != running {
-		return 0, errUnavailable
-	}
-
-	id, ctx := s.allocCtxLocked()
-	s.doRunCancelableTaskLocked(ctx, name, task)
-	return id, nil
-}
-
-// RunTask runs a task in new goroutine
-func (s *Runner) RunTask(task func()) error {
-	s.Lock()
-	defer s.Unlock()
-
-	if s.state != running {
-		return errUnavailable
-	}
-
-	s.stop.Add(1)
-
-	go func() {
-		defer s.stop.Done()
-		task()
-	}()
-
-	return nil
-}
-
-// StopCancelableTask stop cancelable spec task
-func (s *Runner) StopCancelableTask(id uint64) error {
-	s.Lock()
-	defer s.Unlock()
-
-	if s.state == stopping ||
-		s.state == stopped {
-		return errors.New("stopper is already stoppped")
-	}
-
-	cancel, ok := s.cancels[id]
-	if !ok {
-		return fmt.Errorf("target task<%d> not found", id)
-	}
-
-	delete(s.cancels, id)
-	cancel()
-	return nil
-}
-
-// Stop stop all task
-// RunTask will failure with an error
-// Wait complete for the tasks that already in execute
-// Cancel the tasks that is not start
-func (s *Runner) StopWithTimeout(timeout time.Duration) ([]string, error) {
-	return s.doStop(timeout)
-}
-
-// Stop stop all task
-// RunTask will failure with an error
-// Wait complete for the tasks that already in execute
-// Cancel the tasks that is not start
-func (s *Runner) Stop() ([]string, error) {
-	return s.doStop(defaultWaitStoppedTimeout)
-}
-
-func (s *Runner) doStop(timeout time.Duration) ([]string, error) {
-	s.Lock()
-	if s.state == stopping ||
-		s.state == stopped {
-		s.Unlock()
-		return nil, errors.New("stopper is already stoppped")
-	}
-	s.state = stopping
-
-	for _, cancel := range s.cancels {
-		cancel()
-	}
-	s.Unlock()
-
-	go func() {
-		s.stop.Wait()
-		s.stopC <- struct{}{}
-	}()
-
-	select {
-	case <-time.After(timeout):
-		s.Lock()
-		defer s.Unlock()
-		var timeoutWorkers []string
-		for name, run := range s.tasks {
-			if run {
-				timeoutWorkers = append(timeoutWorkers, name)
-			}
-		}
-		return timeoutWorkers, errors.New("waitting for task complete timeout")
-	case <-s.stopC:
-	}
-
-	s.Lock()
-	defer s.Unlock()
-	s.state = stopped
-	return nil, nil
-}
-
-func (s *Runner) doRunCancelableTaskLocked(ctx context.Context, name string, task func(context.Context)) {
-	s.tasks[name] = true
-	go func() {
-		if err := recover(); err != nil {
-			panic(err)
-		}
-		defer func() {
-			s.stop.Done()
-			s.Lock()
-			s.tasks[name] = false
-			s.Unlock()
-		}()
-		task(ctx)
-	}()
-}
-
-func (s *Runner) getNamedQueueLocked(name string) *Queue {
-	return s.namedQueue[name]
-}
-
-func (s *Runner) allocCtxLocked() (uint64, context.Context) {
-	ctx, cancel := context.WithCancel(context.Background())
-	s.lastID++
-	id := s.lastID
-	s.cancels[id] = cancel
-	s.stop.Add(1)
-	return id, ctx
-}


Test patch:
diff --git a/raftstore/replica_create_test.go b/raftstore/replica_create_test.go
index 3fd1e05..01ae7f7 100644
--- a/raftstore/replica_create_test.go
+++ b/raftstore/replica_create_test.go
@@ -25,7 +25,7 @@ func TestShardCreateWithStart(t *testing.T) {
 	db := NewTestDataBuilder()
 	newReplicaCreator(s).
 		withReason("TestShardCreateWithSaveMetadata").
-		withStartReplica(nil, func(r *replica) {
+		withStartReplica(false, nil, func(r *replica) {
 			pr = r
 		}).
 		create([]Shard{
diff --git a/raftstore/testutil.go b/raftstore/testutil.go
index 7f710f4..05c21a0 100644
--- a/raftstore/testutil.go
+++ b/raftstore/testutil.go
@@ -40,7 +40,7 @@ import (
 	"github.com/matrixorigin/matrixcube/storage/kv/mem"
 	"github.com/matrixorigin/matrixcube/storage/kv/pebble"
 	"github.com/matrixorigin/matrixcube/util"
-	"github.com/matrixorigin/matrixcube/util/task"
+	"github.com/matrixorigin/matrixcube/util/stop"
 	"github.com/matrixorigin/matrixcube/util/testutil"
 	"github.com/matrixorigin/matrixcube/util/uuid"
 	"github.com/matrixorigin/matrixcube/vfs"
@@ -636,7 +636,7 @@ func newTestKVClient(t *testing.T, store Store) TestKVClient {
 	kv := &testKVClient{
 		errCtx:   make(map[string]chan error),
 		doneCtx:  make(map[string]chan string),
-		runner:   task.NewRunner(),
+		stopper:  stop.NewStopper("test-kv-client"),
 		requests: make(map[string]rpc.Request),
 	}
 	kv.proxy = store.GetShardsProxy()
@@ -648,7 +648,7 @@ func newTestKVClient(t *testing.T, store Store) TestKVClient {
 type testKVClient struct {
 	sync.RWMutex
 
-	runner   *task.Runner
+	stopper  *stop.Stopper
 	proxy    ShardsProxy
 	doneCtx  map[string]chan string
 	errCtx   map[string]chan error
@@ -783,7 +783,7 @@ func (kv *testKVClient) Get(key string, timeout time.Duration) (string, error) {
 }
 
 func (kv *testKVClient) Close() {
-	kv.runner.Stop()
+	kv.stopper.Stop()
 }
 
 func (kv *testKVClient) addContext(id string, c chan string, ec chan error) {
diff --git a/util/task/task_test.go b/util/task/task_test.go
deleted file mode 100644
index dbfe36b..0000000
--- a/util/task/task_test.go
+++ /dev/null
@@ -1,223 +0,0 @@
-// Copyright 2021 MatrixOrigin.
-//
-// 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,
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package task
-
-import (
-	"context"
-	"fmt"
-	"sort"
-	"sync"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/assert"
-)
-
-func TestTask(t *testing.T) {
-	runner := NewRunner()
-	c := make(chan struct{})
-	defer close(c)
-
-	err := runner.RunTask(func() {
-		c <- struct{}{}
-	})
-
-	if err != nil {
-		t.Error("run task failed, return a error", err)
-		return
-	}
-
-	select {
-	case <-c:
-	case <-time.After(time.Millisecond * 50):
-		t.Error("run task failed, task not run after 50ms")
-	}
-
-	runner.AddNamedWorker("name-0", func() {})
-	yes := false
-	var ok int32
-	complete := make(chan struct{}, 1)
-	runner.RunJobWithNamedWorker("", "name-0", func() error {
-		atomic.StoreInt32(&ok, 1)
-		return nil
-	})
-
-	runner.RunJobWithNamedWorker("", "name-0", func() error {
-		if atomic.LoadInt32(&ok) > 0 {
-			yes = true
-		}
-		complete <- struct{}{}
-		return nil
-	})
-
-	<-complete
-	if !yes {
-		t.Error("run named task failed, task not liner excution")
-	}
-
-	start := make(chan struct{}, 1)
-	defer close(start)
-
-	result := false
-
-	_, err = runner.RunCancelableTask("1", func(c context.Context) {
-		select {
-		case <-c.Done():
-			result = true
-		case <-start:
-
-		}
-	})
-
-	if err != nil {
-		t.Error("run cancelable task failed, return a error", err)
-		return
-	}
-
-	defaultWaitStoppedTimeout = time.Second
-	_, err = runner.Stop()
-	if err != nil {
-		t.Error("stop runner failed, return a error", err)
-		return
-	}
-
-	start <- struct{}{}
-	if !result {
-		t.Error("cancelable task excuted after stop")
-		return
-	}
-}
-
-func TestNamedTask(t *testing.T) {
-	runner := NewRunner()
-	runner.AddNamedWorker("apply", func() {})
-
-	cnt := 0
-	complete := make(chan struct{}, 1)
-	runner.RunJobWithNamedWorker("", "apply", func() error {
-		cnt++
-		complete <- struct{}{}
-		return nil
-	})
-
-	<-complete
-	if cnt != 1 {
-		t.Errorf("run named job failed. expect=<%d>, actual=<%d>", 1, cnt)
-	}
-
-	runner.RunJobWithNamedWorker("", "apply", func() error {
-		cnt++
-		complete <- struct{}{}
-		return nil
-	})
-
-	<-complete
-	if cnt != 2 {
-		t.Errorf("run named job failed. expect=<%d>, actual=<%d>", 2, cnt)
-	}
-}
-
-func TestStopWithID(t *testing.T) {
-	runner := NewRunner()
-	c := make(chan struct{})
-	defer close(c)
-
-	result := false
-	start := make(chan struct{}, 1)
-	defer close(start)
-
-	id, err := runner.RunCancelableTask("1", func(c context.Context) {
-		select {
-		case <-c.Done():
-			result = true
-		case <-start:
-
-		}
-	})
-	if err != nil {
-		t.Error("run cancelable task failed, return a error", err)
-		return
-	}
-
-	err = runner.StopCancelableTask(id)
-	if err != nil {
-		t.Error("stop cancelable task failed, return a error", err)
-		return
-	}
-
-	for index := 0; index < 10; index++ {
-		runner.RunCancelableTask(fmt.Sprintf("w-%d", index), func(c context.Context) {
-			select {
-			case <-c.Done():
-				result = true
-			case <-time.After(time.Hour):
-
-			}
-		})
-	}
-
-	defaultWaitStoppedTimeout = time.Second
-	_, err = runner.Stop()
-	if err != nil {
-		t.Error("stop runner failed, return a error", err)
-		return
-	}
-
-	start <- struct{}{}
-	if !result {
-		t.Error("cancelable task excuted after stop")
-		return
-	}
-}
-
-func TestStopWithTimeoutWorkers(t *testing.T) {
-	runner := NewRunner()
-	_, err := runner.AddNamedWorker("w-0", nil)
-	assert.NoError(t, err)
-	_, err = runner.AddNamedWorker("w-1", nil)
-	assert.NoError(t, err)
-	_, err = runner.AddNamedWorker("w-2", nil)
-	assert.NoError(t, err)
-
-	var wg sync.WaitGroup
-	wg.Add(4)
-	runner.RunJobWithNamedWorker("", "w-0", func() error {
-		wg.Done()
-		time.Sleep(time.Second * 10)
-		return nil
-	})
-	runner.RunJobWithNamedWorker("", "w-1", func() error {
-		wg.Done()
-		time.Sleep(time.Second * 10)
-		return nil
-	})
-	runner.RunJobWithNamedWorker("", "w-2", func() error {
-		wg.Done()
-		return nil
-	})
-	runner.RunCancelableTask("w-3", func(ctx context.Context) {
-		wg.Done()
-		time.Sleep(time.Second * 10)
-	})
-
-	wg.Wait()
-	timeoutWorkers, err := runner.doStop(time.Second)
-	assert.Error(t, err)
-	assert.Equal(t, 3, len(timeoutWorkers))
-	sort.Strings(timeoutWorkers)
-	assert.Equal(t, "w-0", timeoutWorkers[0])
-	assert.Equal(t, "w-1", timeoutWorkers[1])
-	assert.Equal(t, "w-3", timeoutWorkers[2])
-}