Repository: kubernetes-sigs/kube-scheduler-simulator
Pull request #213 (2022-08-14 22:19:05)

fix(util): handle semaphore internally
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide/first-contribution.md#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Please label this pull request according to what type of issue you are addressing, especially if this is a release targeted pull request. For reference on required PR/issue labels, read here:
https://git.k8s.io/community/contributors/devel/sig-release/release.md#issuepr-kind-label
3. Ensure you have added or ran the appropriate tests for your PR: https://git.k8s.io/community/contributors/devel/sig-testing/testing.md
4. If you want *faster* PR reviews, read how: https://git.k8s.io/community/contributors/guide/pull-requests.md#best-practices-for-faster-reviews
5. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

#### What type of PR is this?

/kind feature

<!--
Add one of the following kinds:
/kind bug
/kind cleanup
/kind documentation
/kind feature

Optionally add one or more of the following kinds if applicable:
/kind api-change
/kind deprecation
/kind failing-test
/kind flake
/kind regression
-->

#### What this PR does / why we need it:

For cleanup. We don't need to `eg.Sem.Acquire` anymore. It's handled internally.

#### Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #

#### Special notes for your reviewer:


/label tide/merge-method-squash
/assign @196Ikuchil 

k8s-ci-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/kube-scheduler-simulator/pull/213#" title="Author self-approved">sanposhiho</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fkube-scheduler-simulator).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/kubernetes-sigs/kube-scheduler-simulator/blob/master/OWNERS)~~ [sanposhiho]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
k8s-ci-robot: @sanposhiho: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-kube-scheduler-simulator-backend-lint | f69955c85595274e22a97fc4f09593b28b9b4f9e | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kube-scheduler-simulator/213/pull-kube-scheduler-simulator-backend-lint/1558941308939145216) | true | `/test pull-kube-scheduler-simulator-backend-lint`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kube-scheduler-simulator&pr=213). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asanposhiho). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/kube-scheduler-simulator/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
196Ikuchil: 👍

/lgtm

Patch:
diff --git a/simulator/export/export.go b/simulator/export/export.go
index 8b4701ff..68b6292c 100644
--- a/simulator/export/export.go
+++ b/simulator/export/export.go
@@ -149,29 +149,29 @@ func (s *Service) get(ctx context.Context, opts options) (*ResourcesForExport, e
 	errgrp := util.NewErrGroupWithSemaphore(ctx)
 	resources := ResourcesForExport{}
 
-	if err := s.listPods(ctx, &resources, &errgrp, opts); err != nil {
+	if err := s.listPods(ctx, &resources, errgrp, opts); err != nil {
 		return nil, xerrors.Errorf("call listPods: %w", err)
 	}
-	if err := s.listNodes(ctx, &resources, &errgrp, opts); err != nil {
+	if err := s.listNodes(ctx, &resources, errgrp, opts); err != nil {
 		return nil, xerrors.Errorf("call listNodes: %w", err)
 	}
-	if err := s.listPvs(ctx, &resources, &errgrp, opts); err != nil {
+	if err := s.listPvs(ctx, &resources, errgrp, opts); err != nil {
 		return nil, xerrors.Errorf("call listPvs: %w", err)
 	}
-	if err := s.listPvcs(ctx, &resources, &errgrp, opts); err != nil {
+	if err := s.listPvcs(ctx, &resources, errgrp, opts); err != nil {
 		return nil, xerrors.Errorf("call listPvcs: %w", err)
 	}
-	if err := s.listStorageClasses(ctx, &resources, &errgrp, opts); err != nil {
+	if err := s.listStorageClasses(ctx, &resources, errgrp, opts); err != nil {
 		return nil, xerrors.Errorf("call listStorageClasses: %w", err)
 	}
-	if err := s.listPcs(ctx, &resources, &errgrp, opts); err != nil {
+	if err := s.listPcs(ctx, &resources, errgrp, opts); err != nil {
 		return nil, xerrors.Errorf("call listPcs: %w", err)
 	}
-	if err := s.getSchedulerConfig(ctx, &resources, &errgrp); err != nil {
+	if err := s.getSchedulerConfig(&resources, errgrp); err != nil {
 		return nil, xerrors.Errorf("call getSchedulerConfig: %w", err)
 	}
 
-	if err := errgrp.Grp.Wait(); err != nil {
+	if err := errgrp.Wait(); err != nil {
 		return nil, xerrors.Errorf("get resources all: %w", err)
 	}
 	return &resources, nil
@@ -193,32 +193,32 @@ func (s *Service) Export(ctx context.Context, opts ...Option) (*ResourcesForExpo
 func (s *Service) apply(ctx context.Context, resources *ResourcesForImport, opts options) error {
 	errgrp := util.NewErrGroupWithSemaphore(ctx)
 
-	if err := s.applyPcs(ctx, resources, &errgrp, opts); err != nil {
+	if err := s.applyPcs(ctx, resources, errgrp, opts); err != nil {
 		return xerrors.Errorf("call applyPcs: %w", err)
 	}
-	if err := s.applyStorageClasses(ctx, resources, &errgrp, opts); err != nil {
+	if err := s.applyStorageClasses(ctx, resources, errgrp, opts); err != nil {
 		return xerrors.Errorf("call applyStorageClasses: %w", err)
 	}
-	if err := s.applyPvcs(ctx, resources, &errgrp, opts); err != nil {
+	if err := s.applyPvcs(ctx, resources, errgrp, opts); err != nil {
 		return xerrors.Errorf("call applyPvcs: %w", err)
 	}
-	if err := s.applyNodes(ctx, resources, &errgrp, opts); err != nil {
+	if err := s.applyNodes(ctx, resources, errgrp, opts); err != nil {
 		return xerrors.Errorf("call applyNodes: %w", err)
 	}
-	if err := s.applyPods(ctx, resources, &errgrp, opts); err != nil {
+	if err := s.applyPods(ctx, resources, errgrp, opts); err != nil {
 		return xerrors.Errorf("call applyPods: %w", err)
 	}
 
-	if err := errgrp.Grp.Wait(); err != nil {
+	if err := errgrp.Wait(); err != nil {
 		return xerrors.Errorf("apply resources: %w", err)
 	}
 
 	// `applyPvs` should be called after `applyPvcs` finished,
 	// because `applyPvs` look up PersistentVolumeClaim for `Spec.ClaimRef.UID` field.
-	if err := s.applyPvs(ctx, resources, &errgrp, opts); err != nil {
+	if err := s.applyPvs(ctx, resources, errgrp, opts); err != nil {
 		return xerrors.Errorf("call applyPvs: %w", err)
 	}
-	if err := errgrp.Grp.Wait(); err != nil {
+	if err := errgrp.Wait(); err != nil {
 		return xerrors.Errorf("apply PVs: %w", err)
 	}
 	return nil
@@ -244,12 +244,8 @@ func (s *Service) Import(ctx context.Context, resources *ResourcesForImport, opt
 	return nil
 }
 
-func (s *Service) listPods(ctx context.Context, r *ResourcesForExport, eg *util.ErrGroupWithSemaphore, opts options) error {
-	if err := eg.Sem.Acquire(ctx, 1); err != nil {
-		return xerrors.Errorf("acquire semaphore: %w", err)
-	}
-	eg.Grp.Go(func() error {
-		defer eg.Sem.Release(1)
+func (s *Service) listPods(ctx context.Context, r *ResourcesForExport, eg *util.SemaphoredErrGroup, opts options) error {
+	if err := eg.Go(func() error {
 		pods, err := s.podService.List(ctx, metav1.NamespaceAll)
 		if err != nil {
 			if !opts.ignoreErr {
@@ -260,16 +256,14 @@ func (s *Service) listPods(ctx context.Context, r *ResourcesForExport, eg *util.
 		}
 		r.Pods = pods.Items
 		return nil
-	})
+	}); err != nil {
+		return xerrors.Errorf("start error group: %w", err)
+	}
 	return nil
 }
 
-func (s *Service) listNodes(ctx context.Context, r *ResourcesForExport, eg *util.ErrGroupWithSemaphore, opts options) error {
-	if err := eg.Sem.Acquire(ctx, 1); err != nil {
-		return xerrors.Errorf("acquire semaphore: %w", err)
-	}
-	eg.Grp.Go(func() error {
-		defer eg.Sem.Release(1)
+func (s *Service) listNodes(ctx context.Context, r *ResourcesForExport, eg *util.SemaphoredErrGroup, opts options) error {
+	if err := eg.Go(func() error {
 		nodes, err := s.nodeService.List(ctx)
 		if err != nil {
 			if !opts.ignoreErr {
@@ -280,16 +274,14 @@ func (s *Service) listNodes(ctx context.Context, r *ResourcesForExport, eg *util
 		}
 		r.Nodes = nodes.Items
 		return nil
-	})
+	}); err != nil {
+		return xerrors.Errorf("start error group: %w", err)
+	}
 	return nil
 }
 
-func (s *Service) listPvs(ctx context.Context, r *ResourcesForExport, eg *util.ErrGroupWithSemaphore, opts options) error {
-	if err := eg.Sem.Acquire(ctx, 1); err != nil {
-		return xerrors.Errorf("acquire semaphore: %w", err)
-	}
-	eg.Grp.Go(func() error {
-		defer eg.Sem.Release(1)
+func (s *Service) listPvs(ctx context.Context, r *ResourcesForExport, eg *util.SemaphoredErrGroup, opts options) error {
+	if err := eg.Go(func() error {
 		pvs, err := s.pvService.List(ctx)
 		if err != nil {
 			if !opts.ignoreErr {
@@ -300,16 +292,14 @@ func (s *Service) listPvs(ctx context.Context, r *ResourcesForExport, eg *util.E
 		}
 		r.Pvs = pvs.Items
 		return nil
-	})
+	}); err != nil {
+		return xerrors.Errorf("start error group: %w", err)
+	}
 	return nil
 }
 
-func (s *Service) listPvcs(ctx context.Context, r *ResourcesForExport, eg *util.ErrGroupWithSemaphore, opts options) error {
-	if err := eg.Sem.Acquire(ctx, 1); err != nil {
-		return xerrors.Errorf("acquire semaphore: %w", err)
-	}
-	eg.Grp.Go(func() error {
-		defer eg.Sem.Release(1)
+func (s *Service) listPvcs(ctx context.Context, r *ResourcesForExport, eg *util.SemaphoredErrGroup, opts options) error {
+	if err := eg.Go(func() error {
 		pvcs, err := s.pvcService.List(ctx, metav1.NamespaceAll)
 		if err != nil {
 			if !opts.ignoreErr {
@@ -320,16 +310,14 @@ func (s *Service) listPvcs(ctx context.Context, r *ResourcesForExport, eg *util.
 		}
 		r.Pvcs = pvcs.Items
 		return nil
-	})
+	}); err != nil {
+		return xerrors.Errorf("start error group: %w", err)
+	}
 	return nil
 }
 
-func (s *Service) listStorageClasses(ctx context.Context, r *ResourcesForExport, eg *util.ErrGroupWithSemaphore, opts options) error {
-	if err := eg.Sem.Acquire(ctx, 1); err != nil {
-		return xerrors.Errorf("acquire semaphore: %w", err)
-	}
-	eg.Grp.Go(func() error {
-		defer eg.Sem.Release(1)
+func (s *Service) listStorageClasses(ctx context.Context, r *ResourcesForExport, eg *util.SemaphoredErrGroup, opts options) error {
+	if err := eg.Go(func() error {
 		scs, err := s.storageClassService.List(ctx)
 		if err != nil {
 			if !opts.ignoreErr {
@@ -340,16 +328,14 @@ func (s *Service) listStorageClasses(ctx context.Context, r *ResourcesForExport,
 		}
 		r.StorageClasses = scs.Items
 		return nil
-	})
+	}); err != nil {
+		return xerrors.Errorf("start error group: %w", err)
+	}
 	return nil
 }
 
-func (s *Service) listPcs(ctx context.Context, r *ResourcesForExport, eg *util.ErrGroupWithSemaphore, opts options) error {
-	if err := eg.Sem.Acquire(ctx, 1); err != nil {
-		return xerrors.Errorf("acquire semaphore: %w", err)
-	}
-	eg.Grp.Go(func() error {
-		defer eg.Sem.Release(1)
+func (s *Service) listPcs(ctx context.Context, r *ResourcesForExport, eg *util.SemaphoredErrGroup, opts options) error {
+	if err := eg.Go(func() error {
 		pcs, err := s.priorityclassService.List(ctx)
 		if err != nil {
 			if !opts.ignoreErr {
@@ -366,34 +352,30 @@ func (s *Service) listPcs(ctx context.Context, r *ResourcesForExport, eg *util.E
 		}
 		r.PriorityClasses = result
 		return nil
-	})
+	}); err != nil {
+		return xerrors.Errorf("start error group: %w", err)
+	}
 	return nil
 }
 
-func (s *Service) getSchedulerConfig(ctx context.Context, r *ResourcesForExport, eg *util.ErrGroupWithSemaphore) error {
-	if err := eg.Sem.Acquire(ctx, 1); err != nil {
-		return xerrors.Errorf("acquire semaphore: %w", err)
-	}
-	eg.Grp.Go(func() error {
-		defer eg.Sem.Release(1)
+func (s *Service) getSchedulerConfig(r *ResourcesForExport, eg *util.SemaphoredErrGroup) error {
+	if err := eg.Go(func() error {
 		ss := s.schedulerService.GetSchedulerConfig()
 		r.SchedulerConfig = ss
 		return nil
-	})
+	}); err != nil {
+		return xerrors.Errorf("start error group: %w", err)
+	}
 	return nil
 }
 
-func (s *Service) applyPcs(ctx context.Context, r *ResourcesForImport, eg *util.ErrGroupWithSemaphore, opts options) error {
+func (s *Service) applyPcs(ctx context.Context, r *ResourcesForImport, eg *util.SemaphoredErrGroup, opts options) error {
 	for i := range r.PriorityClasses {
 		pc := r.PriorityClasses[i]
 		if isSystemPriorityClass(*pc.Name) {
 			continue
 		}
-		if err := eg.Sem.Acquire(ctx, 1); err != nil {
-			return xerrors.Errorf("acquire semaphore: %w", err)
-		}
-		eg.Grp.Go(func() error {
-			defer eg.Sem.Release(1)
+		if err := eg.Go(func() error {
 			pc.ObjectMetaApplyConfiguration.UID = nil
 			_, err := s.priorityclassService.Apply(ctx, &pc)
 			if err != nil {
@@ -403,19 +385,17 @@ func (s *Service) applyPcs(ctx context.Context, r *ResourcesForImport, eg *util.
 				klog.Errorf("failed to apply priorityClasses: %v", err)
 			}
 			return nil
-		})
+		}); err != nil {
+			return xerrors.Errorf("start error group: %w", err)
+		}
 	}
 	return nil
 }
 
-func (s *Service) applyStorageClasses(ctx context.Context, r *ResourcesForImport, eg *util.ErrGroupWithSemaphore, opts options) error {
+func (s *Service) applyStorageClasses(ctx context.Context, r *ResourcesForImport, eg *util.SemaphoredErrGroup, opts options) error {
 	for i := range r.StorageClasses {
 		sc := r.StorageClasses[i]
-		if err := eg.Sem.Acquire(ctx, 1); err != nil {
-			return xerrors.Errorf("acquire semaphore: %w", err)
-		}
-		eg.Grp.Go(func() error {
-			defer eg.Sem.Release(1)
+		if err := eg.Go(func() error {
 			sc.ObjectMetaApplyConfiguration.UID = nil
 			_, err := s.storageClassService.Apply(ctx, &sc)
 			if err != nil {
@@ -425,19 +405,17 @@ func (s *Service) applyStorageClasses(ctx context.Context, r *ResourcesForImport
 				klog.Errorf("failed to apply StorageClass: %v", err)
 			}
 			return nil
-		})
+		}); err != nil {
+			return xerrors.Errorf("start error group: %w", err)
+		}
 	}
 	return nil
 }
 
-func (s *Service) applyPvcs(ctx context.Context, r *ResourcesForImport, eg *util.ErrGroupWithSemaphore, opts options) error {
+func (s *Service) applyPvcs(ctx context.Context, r *ResourcesForImport, eg *util.SemaphoredErrGroup, opts options) error {
 	for i := range r.Pvcs {
 		pvc := r.Pvcs[i]
-		if err := eg.Sem.Acquire(ctx, 1); err != nil {
-			return xerrors.Errorf("acquire semaphore: %w", err)
-		}
-		eg.Grp.Go(func() error {
-			defer eg.Sem.Release(1)
+		if err := eg.Go(func() error {
 			pvc.ObjectMetaApplyConfiguration.UID = nil
 			_, err := s.pvcService.Apply(ctx, *pvc.Namespace, &pvc)
 			if err != nil {
@@ -447,19 +425,17 @@ func (s *Service) applyPvcs(ctx context.Context, r *ResourcesForImport, eg *util
 				klog.Errorf("failed to apply PersistentVolumeClaims: %v", err)
 			}
 			return nil
-		})
+		}); err != nil {
+			return xerrors.Errorf("start error group: %w", err)
+		}
 	}
 	return nil
 }
 
-func (s *Service) applyPvs(ctx context.Context, r *ResourcesForImport, eg *util.ErrGroupWithSemaphore, opts options) error {
+func (s *Service) applyPvs(ctx context.Context, r *ResourcesForImport, eg *util.SemaphoredErrGroup, opts options) error {
 	for i := range r.Pvs {
 		pv := r.Pvs[i]
-		if err := eg.Sem.Acquire(ctx, 1); err != nil {
-			return xerrors.Errorf("acquire semaphore: %w", err)
-		}
-		eg.Grp.Go(func() error {
-			defer eg.Sem.Release(1)
+		if err := eg.Go(func() error {
 			pv.ObjectMetaApplyConfiguration.UID = nil
 			if pv.Status != nil && pv.Status.Phase != nil {
 				if *pv.Status.Phase == "Bound" {
@@ -481,19 +457,17 @@ func (s *Service) applyPvs(ctx context.Context, r *ResourcesForImport, eg *util.
 				klog.Errorf("failed to apply PersistentVolume: %v", err)
 			}
 			return nil
-		})
+		}); err != nil {
+			return xerrors.Errorf("start error group: %w", err)
+		}
 	}
 	return nil
 }
 
-func (s *Service) applyNodes(ctx context.Context, r *ResourcesForImport, eg *util.ErrGroupWithSemaphore, opts options) error {
+func (s *Service) applyNodes(ctx context.Context, r *ResourcesForImport, eg *util.SemaphoredErrGroup, opts options) error {
 	for i := range r.Nodes {
 		node := r.Nodes[i]
-		if err := eg.Sem.Acquire(ctx, 1); err != nil {
-			return xerrors.Errorf("acquire semaphore: %w", err)
-		}
-		eg.Grp.Go(func() error {
-			defer eg.Sem.Release(1)
+		if err := eg.Go(func() error {
 			node.ObjectMetaApplyConfiguration.UID = nil
 			_, err := s.nodeService.Apply(ctx, &node)
 			if err != nil {
@@ -503,19 +477,17 @@ func (s *Service) applyNodes(ctx context.Context, r *ResourcesForImport, eg *uti
 				klog.Errorf("failed to apply Node: %v", err)
 			}
 			return nil
-		})
+		}); err != nil {
+			return xerrors.Errorf("start error group: %w", err)
+		}
 	}
 	return nil
 }
 
-func (s *Service) applyPods(ctx context.Context, r *ResourcesForImport, eg *util.ErrGroupWithSemaphore, opts options) error {
+func (s *Service) applyPods(ctx context.Context, r *ResourcesForImport, eg *util.SemaphoredErrGroup, opts options) error {
 	for i := range r.Pods {
 		pod := r.Pods[i]
-		if err := eg.Sem.Acquire(ctx, 1); err != nil {
-			return xerrors.Errorf("acquire semaphore: %w", err)
-		}
-		eg.Grp.Go(func() error {
-			defer eg.Sem.Release(1)
+		if err := eg.Go(func() error {
 			pod.ObjectMetaApplyConfiguration.UID = nil
 			_, err := s.podService.Apply(ctx, *pod.Namespace, &pod)
 			if err != nil {
@@ -525,7 +497,9 @@ func (s *Service) applyPods(ctx context.Context, r *ResourcesForImport, eg *util
 				klog.Errorf("failed to apply Pod: %v", err)
 			}
 			return nil
-		})
+		}); err != nil {
+			return xerrors.Errorf("start error group: %w", err)
+		}
 	}
 	return nil
 }
diff --git a/simulator/util/errGroupWithSemaphore.go b/simulator/util/errGroupWithSemaphore.go
deleted file mode 100644
index 7d75327d..00000000
--- a/simulator/util/errGroupWithSemaphore.go
+++ /dev/null
@@ -1,23 +0,0 @@
-package util
-
-import (
-	"context"
-	"runtime"
-
-	"golang.org/x/sync/errgroup"
-	"golang.org/x/sync/semaphore"
-)
-
-type ErrGroupWithSemaphore struct {
-	Grp *errgroup.Group
-	Sem *semaphore.Weighted
-}
-
-func NewErrGroupWithSemaphore(ctx context.Context) ErrGroupWithSemaphore {
-	g, _ := errgroup.WithContext(ctx)
-	sem := semaphore.NewWeighted(int64(runtime.GOMAXPROCS(0)))
-	return ErrGroupWithSemaphore{
-		Grp: g,
-		Sem: sem,
-	}
-}
diff --git a/simulator/util/semaphored_errgroup.go b/simulator/util/semaphored_errgroup.go
new file mode 100644
index 00000000..b2f63bf8
--- /dev/null
+++ b/simulator/util/semaphored_errgroup.go
@@ -0,0 +1,41 @@
+package util
+
+import (
+	"context"
+	"runtime"
+
+	"golang.org/x/sync/errgroup"
+	"golang.org/x/sync/semaphore"
+	"golang.org/x/xerrors"
+)
+
+type SemaphoredErrGroup struct {
+	g *errgroup.Group
+	s *semaphore.Weighted
+}
+
+func NewErrGroupWithSemaphore(ctx context.Context) *SemaphoredErrGroup {
+	g, _ := errgroup.WithContext(ctx)
+	sem := semaphore.NewWeighted(int64(runtime.GOMAXPROCS(0)))
+	return &SemaphoredErrGroup{
+		g: g,
+		s: sem,
+	}
+}
+
+func (e *SemaphoredErrGroup) Go(fn func() error) error {
+	ctx := context.Background()
+	if err := e.s.Acquire(ctx, 1); err != nil {
+		return xerrors.Errorf("acquire semaphore: %w", err)
+	}
+
+	e.g.Go(func() error {
+		defer e.s.Release(1)
+		return fn()
+	})
+	return nil
+}
+
+func (e *SemaphoredErrGroup) Wait() error {
+	return e.g.Wait()
+}


Test patch:
diff --git a/simulator/export/export_test.go b/simulator/export/export_test.go
index 3a4fe707..4c9b9ae1 100644
--- a/simulator/export/export_test.go
+++ b/simulator/export/export_test.go
@@ -1785,8 +1785,8 @@ func TestFunction_listPcs(t *testing.T) {
 				SchedulerConfig: &v1beta2config.KubeSchedulerConfiguration{},
 			}
 
-			err := s.listPcs(ctx, resources, &errgrp, options{})
-			if err := errgrp.Grp.Wait(); err != nil {
+			err := s.listPcs(ctx, resources, errgrp, options{})
+			if err := errgrp.Wait(); err != nil {
 				t.Fatalf("listPcs: %v", err)
 			}
 			diffResponse := cmp.Diff(resources, tt.wantResourcesForExport())
@@ -1865,8 +1865,8 @@ func TestFunction_applyPcs(t *testing.T) {
 			s := NewExportService(fakeclientset, mockPodService, mockNodeService, mockPVService, mockPVCService, mockStorageClassService, mockPriorityClassService, mockSchedulerService)
 			tt.prepareEachServiceMockFn(mockPriorityClassService)
 
-			err := s.applyPcs(ctx, tt.applyConfiguration(), &errgrp, options{})
-			if err := errgrp.Grp.Wait(); err != nil {
+			err := s.applyPcs(ctx, tt.applyConfiguration(), errgrp, options{})
+			if err := errgrp.Wait(); err != nil {
 				t.Fatalf("applyPcs: %v", err)
 			}