resourcemanager: avoid to get nil after task finished when to stop ta…

…sk (#40699)

close #40700

resourcemanager/pooltask/task_manager.go

@@ -139,8 +139,12 @@ func (t *TaskManager[T, U, C, CT, TF]) StopTask(taskID uint64) {
 	shardID := getShardID(taskID)
 	t.task[shardID].rw.Lock()
 	defer t.task[shardID].rw.Unlock()
-	l := t.task[shardID].stats[taskID].stats
-	for e := l.Front(); e != nil; e = e.Next() {
-		e.Value.(tContainer[T, U, C, CT, TF]).task.SetStatus(StopTask)
+	// When call the StopTask, the task may have been deleted from the manager.
+	s, ok := t.task[shardID].stats[taskID]
+	if ok {
+		l := s.stats
+		for e := l.Front(); e != nil; e = e.Next() {
+			e.Value.(tContainer[T, U, C, CT, TF]).task.SetStatus(StopTask)
+		}
 	}
 }

util/gpool/spmc/spmcpool_test.go

@@ -114,6 +114,8 @@ func TestStopPool(t *testing.T) {
 	control.Stop()
 	close(exit)
 	control.Wait()
+	// it should pass. Stop can be used after the pool is closed. we should prevent it from panic.
+	control.Stop()
 	wg.Wait()
 	// close pool
 	pool.ReleaseAndWait()