Repository: matrixorigin/matrixcube
Pull request #663 (2022-01-13 03:12:46)

Minor fix for testWorkerPoolConcurrentJobs
**What type of PR is this?**

- [ ] API-change
- [ ] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [x] 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/raftstore/worker_pool.go b/raftstore/worker_pool.go
index cc720b9..27eebf1 100644
--- a/raftstore/worker_pool.go
+++ b/raftstore/worker_pool.go
@@ -132,10 +132,10 @@ type workerPool struct {
 	logger  *zap.Logger
 	loader  replicaLoader
 	workers []*replicaWorker
-	// workerID -> replica
+	// workerID -> replicaEventHandler
 	busy map[uint64]replicaEventHandler
-	// shardID -> replica
-	pending map[uint64]replicaEventHandler
+	// shardID -> replicaEventHandler
+	pending sync.Map
 	// shardID -> struct{}{}
 	processing map[uint64]struct{}
 	// shardID -> struct{}{}
@@ -153,7 +153,6 @@ func newWorkerPool(logger *zap.Logger, ldb logdb.LogDB, loader replicaLoader, wo
 		logger:        log.Adjust(logger).Named("worker-pool"),
 		loader:        loader,
 		busy:          make(map[uint64]replicaEventHandler),
-		pending:       make(map[uint64]replicaEventHandler),
 		processing:    make(map[uint64]struct{}),
 		readyC:        make(chan struct{}, 1),
 		workerStopper: syncutil.NewStopper(),
@@ -253,11 +252,20 @@ func (p *workerPool) workerPoolMain() {
 }
 
 func (p *workerPool) addPending(h replicaEventHandler) {
-	p.pending[h.getShardID()] = h
+	p.pending.Store(h.getShardID(), h)
 }
 
 func (p *workerPool) removePending(shardID uint64) {
-	delete(p.pending, shardID)
+	p.pending.Delete(shardID)
+}
+
+func (p *workerPool) getPendingCount() int {
+	count := 0
+	p.pending.Range(func(k, v interface{}) bool {
+		count++
+		return true
+	})
+	return count
 }
 
 func (p *workerPool) completed(workerID uint64) {
@@ -324,19 +332,22 @@ func (p *workerPool) canSchedule(h replicaEventHandler) bool {
 }
 
 func (p *workerPool) scheduleWorker() bool {
-	if len(p.pending) == 0 {
-		return false
-	}
+	scheduled := false
 	if w := p.getWorker(); w != nil {
-		for shardID, h := range p.pending {
+		p.pending.Range(func(k, v interface{}) bool {
+			shardID := k.(uint64)
+			h := v.(replicaEventHandler)
 			if p.canSchedule(h) {
 				p.scheduleJob(h, w)
 				p.removePending(shardID)
-				return true
+				scheduled = true
+				return false
 			}
-		}
+			return true
+		})
 	}
-	return false
+
+	return scheduled
 }
 
 func (p *workerPool) scheduleJob(h replicaEventHandler, w *replicaWorker) {


Test patch:
diff --git a/raftstore/worker_pool_test.go b/raftstore/worker_pool_test.go
index e7822ba..4c63959 100644
--- a/raftstore/worker_pool_test.go
+++ b/raftstore/worker_pool_test.go
@@ -95,7 +95,7 @@ func TestWorkerPoolCanScheduleSimpleJob(t *testing.T) {
 	p.start()
 	defer func() {
 		p.close()
-		assert.Equal(t, 0, len(p.pending))
+		assert.Equal(t, 0, p.getPendingCount())
 		assert.Equal(t, 0, len(p.busy))
 	}()
 	p.notify(h.getShardID())
@@ -156,7 +156,7 @@ func TestWorkerPoolScheduleNothingWhenNoIdleWorker(t *testing.T) {
 	p := newWorkerPool(nil, ldb, nil, 32)
 	p.start()
 	defer p.close()
-	p.pending[20] = nil
+	p.pending.Store(20, nil)
 	for _, w := range p.workers {
 		p.busy[w.workerID] = nil
 	}
@@ -174,7 +174,7 @@ func TestWorkerPoolWillNotConcurrentlyProcessTheSameShard(t *testing.T) {
 	p := newWorkerPool(nil, ldb, nil, 32)
 	p.start()
 	defer p.close()
-	p.pending[10] = nil
+	p.pending.Store(10, nil)
 	p.processing[10] = struct{}{}
 	assert.False(t, p.canSchedule(&testReplicaEventHandler{shardID: 10}))
 }