Repository: matrixorigin/matrixcube (go)

Issue:
Failed worker pool test in CI
```
=== RUN   TestWorkerPoolCanScheduleSimpleJob
{"level":"info","ts":"2021-12-15 02:10:01.312","logger":"worker-pool","caller":"raftstore/worker_pool.go:221","msg":"worker pool is ready to stop"}
    worker_pool_test.go:99: 
        	Error Trace:	worker_pool_test.go:99
        	            				worker_pool_test.go:113
        	Error:      	Not equal: 
        	            	expected: 0
        	            	actual  : 1
        	Test:       	TestWorkerPoolCanScheduleSimpleJob
--- FAIL: TestWorkerPoolCanScheduleSimpleJob (0.00s)
```

Pull request description:
Handle unprocessed worker request on exit
This change makes tests easier to test the exit state of the pool.

**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 #539 

**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/raftstore/replica_apply_result.go b/raftstore/replica_apply_result.go
index b59a397..8752540 100644
--- a/raftstore/replica_apply_result.go
+++ b/raftstore/replica_apply_result.go
@@ -29,23 +29,24 @@ import (
 )
 
 type applyResult struct {
-	shardID     uint64
-	index       uint64
-	adminResult *adminResult
-	metrics     applyMetrics
+	shardID       uint64
+	index         uint64
+	adminResult   *adminResult
+	ignoreMetrics bool
+	metrics       applyMetrics
 }
 
 func (res *applyResult) hasSplitResult() bool {
-	return nil != res.adminResult && res.adminResult.splitResult != nil
+	return nil != res.adminResult &&
+		res.adminResult.adminType == rpc.AdminCmdType_BatchSplit
 }
 
-// TODO: should probably use values not pointers
 type adminResult struct {
 	adminType            rpc.AdminCmdType
-	configChangeResult   *configChangeResult
-	splitResult          *splitResult
-	compactionResult     *compactionResult
-	updateMetadataResult *updateMetadataResult
+	configChangeResult   configChangeResult
+	splitResult          splitResult
+	compactionResult     compactionResult
+	updateMetadataResult updateMetadataResult
 }
 
 type updateMetadataResult struct {
@@ -74,7 +75,9 @@ func (pr *replica) notifyPendingProposal(id []byte,
 
 func (pr *replica) handleApplyResult(result applyResult) {
 	pr.updateAppliedIndex(result)
-	pr.updateMetricsHints(result)
+	if !result.ignoreMetrics {
+		pr.updateMetricsHints(result)
+	}
 	if result.adminResult != nil {
 		pr.handleAdminResult(result)
 	}
@@ -95,9 +98,7 @@ func (pr *replica) updateMetricsHints(result applyResult) {
 		pr.stats.approximateSize = result.metrics.approximateDiffHint
 	} else {
 		pr.stats.deleteKeysHint += result.metrics.deleteKeysHint
-		if result.metrics.approximateDiffHint > 0 {
-			pr.stats.approximateSize = result.metrics.approximateDiffHint
-		}
+		pr.stats.approximateSize = result.metrics.approximateDiffHint
 	}
 }
 
@@ -114,13 +115,13 @@ func (pr *replica) handleAdminResult(result applyResult) {
 	}
 }
 
-func (pr *replica) applyUpdateMetadataResult(cp *updateMetadataResult) {
+func (pr *replica) applyUpdateMetadataResult(cp updateMetadataResult) {
 	for _, cc := range cp.changes {
 		pr.rn.ApplyConfChange(cc)
 	}
 }
 
-func (pr *replica) applyCompactionResult(r *compactionResult) {
+func (pr *replica) applyCompactionResult(r compactionResult) {
 	pr.logger.Info("log compaction called",
 		zap.Uint64("index", r.index))
 	// generate a dummy snapshot so we can run the log compaction.
@@ -175,7 +176,7 @@ func (pr *replica) applyCompactionResult(r *compactionResult) {
 	}
 }
 
-func (pr *replica) applyConfChange(cp *configChangeResult) {
+func (pr *replica) applyConfChange(cp configChangeResult) {
 	if cp.index == 0 {
 		// TODO: when the entry was treated as a NoOP, configChangeResult should be
 		// nil and applyConfChange() should be called in the first place.
@@ -248,7 +249,7 @@ func (pr *replica) applyConfChange(cp *configChangeResult) {
 		log.ShardField("shard", pr.getShard()))
 }
 
-func (pr *replica) applySplit(result *splitResult) {
+func (pr *replica) applySplit(result splitResult) {
 	pr.logger.Info("shard split applied, current shard will destory",
 		zap.Int("new-shards-count", len(result.newShards)))
 
diff --git a/raftstore/replica_create.go b/raftstore/replica_create.go
index d11fb79..34bd86b 100644
--- a/raftstore/replica_create.go
+++ b/raftstore/replica_create.go
@@ -43,6 +43,7 @@ type replicaCreator struct {
 	replicaRecordGetter               func(Shard) Replica
 	wc                                *logdb.WorkerContext
 	logger                            *zap.Logger
+	shardsMetadata                    []meta.ShardMetadata
 }
 
 func newReplicaCreator(store *store) *replicaCreator {
@@ -81,7 +82,7 @@ func (rc *replicaCreator) adjust() {
 	if rc.reason == "" {
 		rc.reason = "unknown"
 	}
-	if rc.saveMetadata && rc.wc == nil {
+	if rc.wc == nil {
 		rc.wc = rc.store.logdb.NewWorkerContext()
 	}
 	rc.logger = rc.store.logger.With(log.ReasonField(rc.reason))
@@ -94,6 +95,10 @@ func (rc *replicaCreator) create(shards []Shard) {
 	rc.maybeStartReplicas(shards, replicas)
 }
 
+func (rc *replicaCreator) getShardsMetadata() []meta.ShardMetadata {
+	return rc.shardsMetadata
+}
+
 func (rc *replicaCreator) getTarget(originShards []Shard) ([]Shard, []*replica) {
 	if !rc.startReplica {
 		return originShards, nil
@@ -123,10 +128,6 @@ func (rc *replicaCreator) getTarget(originShards []Shard) ([]Shard, []*replica)
 }
 
 func (rc *replicaCreator) maybeSaveMetadata(shards []Shard) {
-	if !rc.saveMetadata {
-		return
-	}
-
 	doWithShardsByGroupID(rc.store.DataStorageByGroup, func(ds storage.DataStorage, v []Shard) {
 		var sm []meta.ShardMetadata
 		var ids []uint64
@@ -152,6 +153,18 @@ func (rc *replicaCreator) maybeSaveMetadata(shards []Shard) {
 			}
 		}
 
+		if len(sm) == 0 {
+			return
+		}
+
+		if len(sm) > 0 {
+			rc.shardsMetadata = append(rc.shardsMetadata, sm...)
+		}
+
+		if !rc.saveMetadata {
+			return
+		}
+
 		if err := ds.SaveShardMetadata(sm); err != nil {
 			rc.logger.Fatal("failed to save shards metadata",
 				zap.Error(err))
diff --git a/raftstore/replica_state_machine.go b/raftstore/replica_state_machine.go
index 77cd079..183fd24 100644
--- a/raftstore/replica_state_machine.go
+++ b/raftstore/replica_state_machine.go
@@ -180,22 +180,26 @@ func (d *stateMachine) applyCommittedEntries(entries []raftpb.Entry) {
 			// noop entry with empty payload proposed by the leader at the beginning
 			// of its term
 			d.updateAppliedIndexTerm(entry.Index, entry.Term)
-			d.resultHandler.handleApplyResult(applyResult{index: entry.Index})
+			d.resultHandler.handleApplyResult(applyResult{
+				index:         entry.Index,
+				ignoreMetrics: true,
+			})
 			continue
 		}
 
-		d.applyRequestBatch(d.applyCtx)
+		ignoreMetrics := d.applyRequestBatch(d.applyCtx)
 		result := applyResult{
-			shardID:     d.shardID,
-			adminResult: d.applyCtx.adminResult,
-			index:       entry.Index,
-			metrics:     d.applyCtx.metrics,
+			shardID:       d.shardID,
+			adminResult:   d.applyCtx.adminResult,
+			index:         entry.Index,
+			ignoreMetrics: ignoreMetrics,
+			metrics:       d.applyCtx.metrics,
 		}
 		if isConfigChangeEntry(entry) {
 			if result.adminResult == nil {
 				result.adminResult = &adminResult{
 					adminType:          rpc.AdminCmdType_ConfigChange,
-					configChangeResult: &configChangeResult{},
+					configChangeResult: configChangeResult{},
 				}
 			} else {
 				result.adminResult.configChangeResult.confChange = d.applyCtx.v2cc
@@ -232,7 +236,9 @@ func (d *stateMachine) notifyShardRemoved(ctx *applyContext) {
 		resp, isConfigChangeRequestBatch(ctx.req))
 }
 
-func (d *stateMachine) applyRequestBatch(ctx *applyContext) {
+// applyRequestBatch returns a boolean value indicating whether to skip
+// updating the metrics.
+func (d *stateMachine) applyRequestBatch(ctx *applyContext) bool {
 	// FIXME: update impacted tests
 	// if sc, ok := d.store.cfg.Test.Shards[d.shardID]; ok && sc.SkipApply {
 	//	return
@@ -240,9 +246,9 @@ func (d *stateMachine) applyRequestBatch(ctx *applyContext) {
 	if d.isRemoved() {
 		d.logger.Fatal("applying entries on removed replica")
 	}
-
 	var err error
 	var resp rpc.ResponseBatch
+	ignoreMetrics := true
 	if !d.checkEpoch(ctx.req) {
 		if ce := d.logger.Check(zap.DebugLevel, "apply committed log skipped"); ce != nil {
 			ce.Write(zap.Uint64("index", ctx.index),
@@ -270,6 +276,7 @@ func (d *stateMachine) applyRequestBatch(ctx *applyContext) {
 			if ce := d.logger.Check(zap.DebugLevel, "apply write requests"); ce != nil {
 				ce.Write(zap.Uint64("index", ctx.index))
 			}
+			ignoreMetrics = false
 			resp = d.execWriteRequest(ctx)
 		}
 
@@ -283,6 +290,7 @@ func (d *stateMachine) applyRequestBatch(ctx *applyContext) {
 	// executeContext
 	d.resultHandler.notifyPendingProposal(ctx.req.Header.ID,
 		resp, isConfigChangeRequestBatch(ctx.req))
+	return ignoreMetrics
 }
 
 func (d *stateMachine) close() {
diff --git a/raftstore/replica_state_machine_exec.go b/raftstore/replica_state_machine_exec.go
index 3ca7907..819699e 100644
--- a/raftstore/replica_state_machine_exec.go
+++ b/raftstore/replica_state_machine_exec.go
@@ -71,7 +71,7 @@ func (d *stateMachine) doExecCompactLog(ctx *applyContext) (rpc.ResponseBatch, e
 	resp := newAdminResponseBatch(rpc.AdminCmdType_CompactLog, &rpc.CompactLogResponse{})
 	ctx.adminResult = &adminResult{
 		adminType: rpc.AdminCmdType_CompactLog,
-		compactionResult: &compactionResult{
+		compactionResult: compactionResult{
 			index: compactIndex,
 		},
 	}
@@ -188,7 +188,7 @@ func (d *stateMachine) doExecConfigChange(ctx *applyContext) (rpc.ResponseBatch,
 	})
 	ctx.adminResult = &adminResult{
 		adminType: rpc.AdminCmdType_ConfigChange,
-		configChangeResult: &configChangeResult{
+		configChangeResult: configChangeResult{
 			index:   ctx.index,
 			changes: []rpc.ConfigChangeRequest{req},
 			shard:   res,
@@ -217,7 +217,6 @@ func (d *stateMachine) doExecSplit(ctx *applyContext) (rpc.ResponseBatch, error)
 
 	newShardsCount := len(splitReqs.Requests)
 	var newShards []Shard
-	var metadata []meta.ShardMetadata
 	current.Epoch.Version += uint64(newShardsCount)
 	expectStart := current.Start
 	last := len(splitReqs.Requests) - 1
@@ -252,10 +251,11 @@ func (d *stateMachine) doExecSplit(ctx *applyContext) (rpc.ResponseBatch, error)
 		ctx.metrics.admin.splitSucceed++
 	}
 
-	d.replicaCreatorFactory().
-		withReason("splited").
+	// We only create shard init raft log in logdb, create new shards metadata in memory,
+	// and update atomically with the old metadata later.
+	replicaFactory := d.replicaCreatorFactory()
+	replicaFactory.withReason("splited").
 		withLogdbContext(d.wc).
-		withSaveMetadata(false). // no sync
 		create(newShards)
 
 	// We can't destroy Old Shard directly, but mark it as being destroyed. Because at this time, we are not
@@ -272,7 +272,7 @@ func (d *stateMachine) doExecSplit(ctx *applyContext) (rpc.ResponseBatch, error)
 			RemoveData: false,
 		},
 	}
-	err := d.dataStorage.Split(old, metadata, splitReqs.Context)
+	err := d.dataStorage.Split(old, replicaFactory.getShardsMetadata(), splitReqs.Context)
 	if err != nil {
 		if err == storage.ErrAborted {
 			return rpc.ResponseBatch{}, nil
@@ -288,7 +288,7 @@ func (d *stateMachine) doExecSplit(ctx *applyContext) (rpc.ResponseBatch, error)
 	})
 	ctx.adminResult = &adminResult{
 		adminType: rpc.AdminCmdType_BatchSplit,
-		splitResult: &splitResult{
+		splitResult: splitResult{
 			newShards: newShards,
 		},
 	}
@@ -416,7 +416,7 @@ func (d *stateMachine) doUpdateMetadata(ctx *applyContext) (rpc.ResponseBatch, e
 	resp := newAdminResponseBatch(rpc.AdminCmdType_UpdateMetadata, &rpc.UpdateMetadataResponse{})
 	ctx.adminResult = &adminResult{
 		adminType: rpc.AdminCmdType_UpdateMetadata,
-		updateMetadataResult: &updateMetadataResult{
+		updateMetadataResult: updateMetadataResult{
 			changes: cc,
 		},
 	}
diff --git a/raftstore/worker_pool.go b/raftstore/worker_pool.go
index b37ee69..cc720b9 100644
--- a/raftstore/worker_pool.go
+++ b/raftstore/worker_pool.go
@@ -78,6 +78,11 @@ func (w *replicaWorker) workerMain() {
 	for {
 		select {
 		case <-w.stopper.ShouldStop():
+			select {
+			case <-w.requestC:
+				w.completed()
+			default:
+			}
 			w.wc.Close()
 			w.logger.Debug("worker pool worker stopped",
 				zap.Uint64("worker-id", w.workerID))


Test patch:
diff --git a/raftstore/replica_apply_result_test.go b/raftstore/replica_apply_result_test.go
index 3990ebc..719a165 100644
--- a/raftstore/replica_apply_result_test.go
+++ b/raftstore/replica_apply_result_test.go
@@ -17,6 +17,7 @@ import (
 	"testing"
 
 	"github.com/matrixorigin/matrixcube/pb/meta"
+	"github.com/matrixorigin/matrixcube/pb/rpc"
 	"github.com/matrixorigin/matrixcube/storage"
 	skv "github.com/matrixorigin/matrixcube/storage/kv"
 	"github.com/matrixorigin/matrixcube/util/leaktest"
@@ -35,7 +36,7 @@ func TestApplySplit(t *testing.T) {
 	kv := pr.sm.dataStorage.(storage.KVStorageWrapper).GetKVStorage()
 	assert.NoError(t, kv.Set(skv.EncodeDataKey([]byte{1}, nil), []byte("v"), false))
 
-	result := &splitResult{
+	result := splitResult{
 		newShards: []Shard{
 			{ID: 2, Start: []byte{1}, End: []byte{5}, Replicas: []Replica{{ID: 200}}},
 			{ID: 3, Start: []byte{5}, End: []byte{10}, Replicas: []Replica{{ID: 300}}},
@@ -93,27 +94,72 @@ func TestUpdateMetricsHints(t *testing.T) {
 	s := NewSingleTestClusterStore(t).GetStore(0).(*store)
 	pr := newTestReplica(Shard{ID: 1}, Replica{ID: 1}, s)
 
-	pr.updateMetricsHints(applyResult{metrics: applyMetrics{approximateDiffHint: 1, deleteKeysHint: 1, writtenBytes: 1, writtenKeys: 1}})
+	pr.updateMetricsHints(applyResult{
+		metrics: applyMetrics{
+			approximateDiffHint: 1,
+			deleteKeysHint:      1,
+			writtenBytes:        1,
+			writtenKeys:         1,
+		},
+	})
 	assert.Equal(t, uint64(1), pr.stats.approximateSize)
 	assert.Equal(t, uint64(1), pr.stats.deleteKeysHint)
 	assert.Equal(t, uint64(1), pr.stats.writtenBytes)
 	assert.Equal(t, uint64(1), pr.stats.writtenKeys)
 
-	pr.updateMetricsHints(applyResult{metrics: applyMetrics{approximateDiffHint: 3, deleteKeysHint: 1, writtenBytes: 1, writtenKeys: 1}})
+	pr.updateMetricsHints(applyResult{
+		metrics: applyMetrics{
+			approximateDiffHint: 3,
+			deleteKeysHint:      1,
+			writtenBytes:        1,
+			writtenKeys:         1,
+		},
+	})
 	assert.Equal(t, uint64(3), pr.stats.approximateSize)
 	assert.Equal(t, uint64(2), pr.stats.deleteKeysHint)
 	assert.Equal(t, uint64(2), pr.stats.writtenBytes)
 	assert.Equal(t, uint64(2), pr.stats.writtenKeys)
 
 	pr.updateMetricsHints(applyResult{})
-	assert.Equal(t, uint64(3), pr.stats.approximateSize)
+	assert.Equal(t, uint64(0), pr.stats.approximateSize)
 	assert.Equal(t, uint64(2), pr.stats.deleteKeysHint)
 	assert.Equal(t, uint64(2), pr.stats.writtenBytes)
 	assert.Equal(t, uint64(2), pr.stats.writtenKeys)
 
-	pr.updateMetricsHints(applyResult{adminResult: &adminResult{splitResult: &splitResult{}}})
+	pr.updateMetricsHints(applyResult{
+		adminResult: &adminResult{
+			adminType:   rpc.AdminCmdType_BatchSplit,
+			splitResult: splitResult{},
+		},
+	})
 	assert.Equal(t, uint64(0), pr.stats.approximateSize)
 	assert.Equal(t, uint64(0), pr.stats.deleteKeysHint)
 	assert.Equal(t, uint64(2), pr.stats.writtenBytes)
 	assert.Equal(t, uint64(2), pr.stats.writtenKeys)
 }
+
+func TestUpdateMetricsHintsCanBeIgnored(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	s := NewSingleTestClusterStore(t).GetStore(0).(*store)
+	pr := newTestReplica(Shard{ID: 1}, Replica{ID: 1}, s)
+	pr.updateMetricsHints(applyResult{
+		metrics: applyMetrics{
+			approximateDiffHint: 1,
+			deleteKeysHint:      1,
+			writtenBytes:        1,
+			writtenKeys:         1,
+		},
+	})
+	assert.Equal(t, uint64(1), pr.stats.approximateSize)
+	assert.Equal(t, uint64(1), pr.stats.deleteKeysHint)
+	assert.Equal(t, uint64(1), pr.stats.writtenBytes)
+	assert.Equal(t, uint64(1), pr.stats.writtenKeys)
+
+	pr.handleApplyResult(applyResult{
+		ignoreMetrics: true,
+	})
+	assert.Equal(t, uint64(1), pr.stats.approximateSize)
+	pr.handleApplyResult(applyResult{})
+	assert.Equal(t, uint64(0), pr.stats.approximateSize)
+}
diff --git a/raftstore/replica_create_test.go b/raftstore/replica_create_test.go
index d3d6334..6927afa 100644
--- a/raftstore/replica_create_test.go
+++ b/raftstore/replica_create_test.go
@@ -11,8 +11,9 @@ import (
 )
 
 func TestShardCreateWithSaveMetadata(t *testing.T) {
-	testShardCreateWithSaveMetadataWithSync(t, false)
-	testShardCreateWithSaveMetadataWithSync(t, true)
+	testShardCreateWithSaveMetadataWithSync(t, true, false)
+	testShardCreateWithSaveMetadataWithSync(t, true, true)
+	testShardCreateWithSaveMetadataWithSync(t, false, false)
 }
 
 func TestShardCreateWithStart(t *testing.T) {
@@ -35,29 +36,44 @@ func TestShardCreateWithStart(t *testing.T) {
 	assert.Equal(t, db.CreateShard(1, "1/0"), pr.getShard())
 }
 
-func testShardCreateWithSaveMetadataWithSync(t *testing.T, sync bool) {
+func testShardCreateWithSaveMetadataWithSync(t *testing.T, save bool, sync bool) {
 	defer leaktest.AfterTest(t)()
 	s, close := newTestStore(t)
 	defer close()
 
 	s.meta.SetID(100)
 	db := NewTestDataBuilder()
-	newReplicaCreator(s).
-		withReason("TestShardCreateWithSaveMetadata").
-		withSaveMetadata(sync).
+	f := newReplicaCreator(s)
+	if save {
+		f.withSaveMetadata(sync)
+	}
+	f.withReason("TestShardCreateWithSaveMetadata").
 		create([]Shard{
 			db.CreateShard(1, "1/10/v/t,2/100/v/t"),
 		})
-	stats, err := s.DataStorageByGroup(0).GetInitialStates()
-	assert.NoError(t, err)
-	assert.Equal(t, 1, len(stats))
-	assert.Equal(t, uint64(1), stats[0].LogIndex)
-	assert.Equal(t, uint64(1), stats[0].ShardID)
-	assert.Equal(t, meta.ReplicaState_Normal, stats[0].Metadata.State)
-	if sync {
-		assert.True(t, s.DataStorageByGroup(0).(storage.StatsKeeper).Stats().SyncCount > 0)
-	} else {
-		assert.True(t, s.DataStorageByGroup(0).(storage.StatsKeeper).Stats().SyncCount == 0)
+
+	assert.Equal(t, 1, len(f.getShardsMetadata()))
+	assert.Equal(t, meta.ShardMetadata{
+		ShardID:  1,
+		LogIndex: 1,
+		Metadata: meta.ShardLocalState{
+			State: meta.ReplicaState_Normal,
+			Shard: db.CreateShard(1, "1/10/v/t,2/100/v/t"),
+		},
+	}, f.getShardsMetadata()[0])
+	if save {
+		stats, err := s.DataStorageByGroup(0).GetInitialStates()
+		assert.NoError(t, err)
+		assert.Equal(t, 1, len(stats))
+		assert.Equal(t, uint64(1), stats[0].LogIndex)
+		assert.Equal(t, uint64(1), stats[0].ShardID)
+		assert.Equal(t, meta.ReplicaState_Normal, stats[0].Metadata.State)
+
+		if sync {
+			assert.True(t, s.DataStorageByGroup(0).(storage.StatsKeeper).Stats().SyncCount > 0)
+		} else {
+			assert.True(t, s.DataStorageByGroup(0).(storage.StatsKeeper).Stats().SyncCount == 0)
+		}
 	}
 
 	stat, err := s.logdb.ReadRaftState(1, 2, 0)