Repository: kubernetes-sigs/kwok
Pull request #449 (2023-04-03 08:12:26)

[kwok] Clean up controller
<!--  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?

<!--
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
-->

/kind cleanup

#### What this PR does / why we need it:

#### 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:

#### Does this PR introduce a user-facing change?
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".

For more information on release notes see: https://git.k8s.io/community/contributors/guide/release-notes.md
-->
```release-note
NONE
```

#### Additional documentation e.g., KEPs (Kubernetes Enhancement Proposals), usage docs, etc.:

<!--
This section can be blank if this pull request does not require a release note.

When adding links which point to resources within git repositories, like
KEPs or supporting documentation, please reference a specific commit and avoid
linking directly to the master branch. This ensures that links reference a
specific point in time, rather than a document that may change over time.

See here for guidance on getting permanent links to files: https://help.github.com/en/articles/getting-permanent-links-to-files

Please use the following format for linking documentation:
- [KEP]: <link>
- [Usage]: <link>
- [Other doc]: <link>
-->
```docs

```


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

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/kwok/pull/449#" title="Author self-approved">wzshiming</a>*

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

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/kwok/blob/main/OWNERS)~~ [wzshiming]

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":[]} -->
netlify[bot]: ### <span aria-hidden="true">👷</span> Deploy Preview for *k8s-kwok* processing.


|  Name | Link |
|:-:|------------------------|
|<span aria-hidden="true">🔨</span> Latest commit | ec198b29afd05b40a301b6dc5bdbb710cc0bf881 |
|<span aria-hidden="true">🔍</span> Latest deploy log | https://app.netlify.com/sites/k8s-kwok/deploys/642a8a6c826863000817d8ac |
k8s-ci-robot: PR needs rebase.

<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.
</details>
k8s-ci-robot: @wzshiming: 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-kwok-e2e-test-main | 6cfbf487a7cf514df9f64cff5a3abf9b02bb3587 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-e2e-test-main/1645324884269600768) | true | `/test pull-kwok-e2e-test-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kwok&pr=449). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Awzshiming). 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/kwok/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 -->
k8s-ci-robot: @wzshiming: 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-kwok-unit-test-main | f6b2fd715c1ad6c294ee0863f5ebcdc7fa0e9818 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-unit-test-main/1645736573083324416) | true | `/test pull-kwok-unit-test-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kwok&pr=449). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Awzshiming). 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/kwok/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 -->
k8s-ci-robot: @wzshiming: 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-kwok-unit-test-main | ab57418f8b9061d57dbdffe342b43a61dbfe2e71 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-unit-test-main/1646345845219528704) | true | `/test pull-kwok-unit-test-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kwok&pr=449). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Awzshiming). 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/kwok/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 -->
k8s-ci-robot: @wzshiming: 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-kwok-verify-main | 876858a39d4ea20166d975c575a97b3a683db082 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-verify-main/1649331891347656704) | true | `/test pull-kwok-verify-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kwok&pr=449). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Awzshiming). 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/kwok/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 -->
k8s-ci-robot: @wzshiming: The following tests **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-kwok-verify-main | 876858a39d4ea20166d975c575a97b3a683db082 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-verify-main/1649331891347656704) | true | `/test pull-kwok-verify-main`
pull-kwok-unit-test-main | 876858a39d4ea20166d975c575a97b3a683db082 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-unit-test-main/1649331891439931392) | true | `/test pull-kwok-unit-test-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kwok&pr=449). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Awzshiming). 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/kwok/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 -->
k8s-ci-robot: @wzshiming: The following tests **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-kwok-unit-test-main | 876858a39d4ea20166d975c575a97b3a683db082 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-unit-test-main/1649331891439931392) | true | `/test pull-kwok-unit-test-main`
pull-kwok-verify-main | 83fc688ca872e63dca31e3eafea235fb0aef7d21 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-verify-main/1649332707332722688) | true | `/test pull-kwok-verify-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kwok&pr=449). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Awzshiming). 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/kwok/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 -->
k8s-ci-robot: @wzshiming: The following tests **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-kwok-verify-main | 83fc688ca872e63dca31e3eafea235fb0aef7d21 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-verify-main/1649332707332722688) | true | `/test pull-kwok-verify-main`
pull-kwok-unit-test-main | 83fc688ca872e63dca31e3eafea235fb0aef7d21 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-unit-test-main/1649332707416608768) | true | `/test pull-kwok-unit-test-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kwok&pr=449). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Awzshiming). 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/kwok/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 -->
k8s-ci-robot: @wzshiming: The following tests **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-kwok-verify-main | 83fc688ca872e63dca31e3eafea235fb0aef7d21 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-verify-main/1649332707332722688) | true | `/test pull-kwok-verify-main`
pull-kwok-unit-test-main | 83fc688ca872e63dca31e3eafea235fb0aef7d21 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-unit-test-main/1649332707416608768) | true | `/test pull-kwok-unit-test-main`
pull-kwok-e2e-test-main | 83fc688ca872e63dca31e3eafea235fb0aef7d21 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-e2e-test-main/1649332707454357504) | true | `/test pull-kwok-e2e-test-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kwok&pr=449). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Awzshiming). 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/kwok/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 -->
k8s-ci-robot: @wzshiming: The following tests **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-kwok-verify-main | 83fc688ca872e63dca31e3eafea235fb0aef7d21 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-verify-main/1649332707332722688) | true | `/test pull-kwok-verify-main`
pull-kwok-e2e-test-main | 83fc688ca872e63dca31e3eafea235fb0aef7d21 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-e2e-test-main/1649332707454357504) | true | `/test pull-kwok-e2e-test-main`
pull-kwok-unit-test-main | 650267d69f97c030c43f74e602c2ccd414a83ad2 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-unit-test-main/1649336364946165760) | true | `/test pull-kwok-unit-test-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kwok&pr=449). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Awzshiming). 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/kwok/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 -->
k8s-ci-robot: @wzshiming: 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-kwok-unit-test-main | 81daf392c5c620b674be810412b7469a8a5f3606 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/kubernetes-sigs_kwok/449/pull-kwok-unit-test-main/1656197586597449728) | true | `/test pull-kwok-unit-test-main`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=kwok&pr=449). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Awzshiming). 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/kwok/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 -->

Patch:
diff --git a/pkg/apis/config/v1alpha1/kwok_configuration_types.go b/pkg/apis/config/v1alpha1/kwok_configuration_types.go
index ca02a5b5..15201756 100644
--- a/pkg/apis/config/v1alpha1/kwok_configuration_types.go
+++ b/pkg/apis/config/v1alpha1/kwok_configuration_types.go
@@ -110,4 +110,12 @@ type KwokConfigurationOptions struct {
 	// EnableProfiling enables /debug/pprof handler, if enableDebuggingHandlers is true.
 	// +default=true
 	EnableProfilingHandler *bool `json:"enableProfilingHandler,omitempty"`
+
+	// PodPlayStageParallelism is the number of PodPlayStages that are allowed to run in parallel.
+	// +default=4
+	PodPlayStageParallelism uint `json:"podPlayStageParallelism,omitempty"`
+
+	// NodePlayStageParallelism is the number of NodePlayStages that are allowed to run in parallel.
+	// +default=4
+	NodePlayStageParallelism uint `json:"nodePlayStageParallelism,omitempty"`
 }
diff --git a/pkg/apis/config/v1alpha1/zz_generated.defaults.go b/pkg/apis/config/v1alpha1/zz_generated.defaults.go
index 9779c8f1..63aec3c2 100644
--- a/pkg/apis/config/v1alpha1/zz_generated.defaults.go
+++ b/pkg/apis/config/v1alpha1/zz_generated.defaults.go
@@ -58,6 +58,12 @@ func SetObjectDefaults_KwokConfiguration(in *KwokConfiguration) {
 		var ptrVar1 bool = true
 		in.Options.EnableProfilingHandler = &ptrVar1
 	}
+	if in.Options.PodPlayStageParallelism == 0 {
+		in.Options.PodPlayStageParallelism = 4
+	}
+	if in.Options.NodePlayStageParallelism == 0 {
+		in.Options.NodePlayStageParallelism = 4
+	}
 }
 
 func SetObjectDefaults_KwokctlConfiguration(in *KwokctlConfiguration) {
diff --git a/pkg/apis/internalversion/kwok_configuration_types.go b/pkg/apis/internalversion/kwok_configuration_types.go
index 8842c4c7..3261bccd 100644
--- a/pkg/apis/internalversion/kwok_configuration_types.go
+++ b/pkg/apis/internalversion/kwok_configuration_types.go
@@ -79,4 +79,10 @@ type KwokConfigurationOptions struct {
 
 	// EnableProfiling enables /debug/pprof handler.
 	EnableProfilingHandler bool
+
+	// PodPlayStageParallelism is the number of PodPlayStages that are allowed to run in parallel.
+	PodPlayStageParallelism uint
+
+	// NodePlayStageParallelism is the number of NodePlayStages that are allowed to run in parallel.
+	NodePlayStageParallelism uint
 }
diff --git a/pkg/apis/internalversion/zz_generated.conversion.go b/pkg/apis/internalversion/zz_generated.conversion.go
index 0311ad31..92fc844a 100644
--- a/pkg/apis/internalversion/zz_generated.conversion.go
+++ b/pkg/apis/internalversion/zz_generated.conversion.go
@@ -1158,6 +1158,8 @@ func autoConvert_internalversion_KwokConfigurationOptions_To_v1alpha1_KwokConfig
 	if err := v1.Convert_bool_To_Pointer_bool(&in.EnableProfilingHandler, &out.EnableProfilingHandler, s); err != nil {
 		return err
 	}
+	out.PodPlayStageParallelism = in.PodPlayStageParallelism
+	out.NodePlayStageParallelism = in.NodePlayStageParallelism
 	return nil
 }
 
@@ -1193,6 +1195,8 @@ func autoConvert_v1alpha1_KwokConfigurationOptions_To_internalversion_KwokConfig
 	if err := v1.Convert_Pointer_bool_To_bool(&in.EnableProfilingHandler, &out.EnableProfilingHandler, s); err != nil {
 		return err
 	}
+	out.PodPlayStageParallelism = in.PodPlayStageParallelism
+	out.NodePlayStageParallelism = in.NodePlayStageParallelism
 	return nil
 }
 
diff --git a/pkg/kwok/cmd/root.go b/pkg/kwok/cmd/root.go
index e90f3532..ce72bfbb 100644
--- a/pkg/kwok/cmd/root.go
+++ b/pkg/kwok/cmd/root.go
@@ -166,6 +166,8 @@ func runE(ctx context.Context, flags *flagpole) error {
 		NodeIP:                                flags.Options.NodeIP,
 		NodeName:                              flags.Options.NodeName,
 		NodePort:                              flags.Options.NodePort,
+		PodPlayStageParallelism:               flags.Options.PodPlayStageParallelism,
+		NodePlayStageParallelism:              flags.Options.NodePlayStageParallelism,
 		NodeStages:                            nodeStages,
 		PodStages:                             podStages,
 	})
diff --git a/pkg/kwok/controllers/controller.go b/pkg/kwok/controllers/controller.go
index 0075d108..193432b3 100644
--- a/pkg/kwok/controllers/controller.go
+++ b/pkg/kwok/controllers/controller.go
@@ -87,6 +87,8 @@ type Config struct {
 	NodePort                              int
 	PodStages                             []*internalversion.Stage
 	NodeStages                            []*internalversion.Stage
+	PodPlayStageParallelism               uint
+	NodePlayStageParallelism              uint
 }
 
 // NewController creates a new fake kubelet controller
@@ -119,7 +121,7 @@ func NewController(conf Config) (*Controller, error) {
 	eventBroadcaster := record.NewBroadcaster()
 	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: "kwok_controller"})
 
-	var lockPodsOnNodeFunc func(ctx context.Context, nodeName string) error
+	var onNodeManagedFunc func(ctx context.Context, nodeName string) error
 
 	nodes, err := NewNodeController(NodeControllerConfig{
 		ClientSet:                             conf.ClientSet,
@@ -130,13 +132,13 @@ func NewController(conf Config) (*Controller, error) {
 		DisregardStatusWithLabelSelector:      conf.DisregardStatusWithLabelSelector,
 		ManageNodesWithLabelSelector:          conf.ManageNodesWithLabelSelector,
 		NodeSelectorFunc:                      nodeSelectorFunc,
-		LockPodsOnNodeFunc: func(ctx context.Context, nodeName string) error {
-			return lockPodsOnNodeFunc(ctx, nodeName)
+		OnNodeManagedFunc: func(ctx context.Context, nodeName string) error {
+			return onNodeManagedFunc(ctx, nodeName)
 		},
-		Stages:              conf.NodeStages,
-		LockNodeParallelism: 16,
-		FuncMap:             defaultFuncMap,
-		Recorder:            recorder,
+		Stages:               conf.NodeStages,
+		PlayStageParallelism: conf.NodePlayStageParallelism,
+		FuncMap:              defaultFuncMap,
+		Recorder:             recorder,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("failed to create nodes controller: %w", err)
@@ -150,7 +152,7 @@ func NewController(conf Config) (*Controller, error) {
 		DisregardStatusWithAnnotationSelector: conf.DisregardStatusWithAnnotationSelector,
 		DisregardStatusWithLabelSelector:      conf.DisregardStatusWithLabelSelector,
 		Stages:                                conf.PodStages,
-		LockPodParallelism:                    16,
+		PlayStageParallelism:                  conf.PodPlayStageParallelism,
 		NodeGetFunc:                           nodes.Get,
 		FuncMap:                               defaultFuncMap,
 		Recorder:                              recorder,
@@ -159,7 +161,7 @@ func NewController(conf Config) (*Controller, error) {
 		return nil, fmt.Errorf("failed to create pods controller: %w", err)
 	}
 
-	lockPodsOnNodeFunc = pods.LockPodsOnNode
+	onNodeManagedFunc = pods.PlayStagePodsOnNode
 
 	n := &Controller{
 		pods:        pods,
diff --git a/pkg/kwok/controllers/node_controller.go b/pkg/kwok/controllers/node_controller.go
index 941d408f..d92f7d59 100644
--- a/pkg/kwok/controllers/node_controller.go
+++ b/pkg/kwok/controllers/node_controller.go
@@ -94,11 +94,12 @@ type NodeController struct {
 	disregardStatusWithLabelSelector      labels.Selector
 	manageNodesWithLabelSelector          string
 	nodeSelectorFunc                      func(node *corev1.Node) bool
-	lockPodsOnNodeFunc                    func(ctx context.Context, nodeName string) error
+	onNodeManagedFunc                     func(ctx context.Context, nodeName string) error
 	nodesSets                             maps.SyncMap[string, *NodeInfo]
 	renderer                              *renderer
-	nodeChan                              chan *corev1.Node
-	parallelTasks                         *parallelTasks
+	preprocessChan                        chan *corev1.Node
+	playStageChan                         chan resourceStageJob[*corev1.Node]
+	playStageParallelism                  uint
 	lifecycle                             Lifecycle
 	cronjob                               *cron.Cron
 	delayJobs                             jobInfoMap
@@ -109,7 +110,7 @@ type NodeController struct {
 type NodeControllerConfig struct {
 	ClientSet                             kubernetes.Interface
 	NodeSelectorFunc                      func(node *corev1.Node) bool
-	LockPodsOnNodeFunc                    func(ctx context.Context, nodeName string) error
+	OnNodeManagedFunc                     func(ctx context.Context, nodeName string) error
 	DisregardStatusWithAnnotationSelector string
 	DisregardStatusWithLabelSelector      string
 	ManageNodesWithLabelSelector          string
@@ -117,7 +118,7 @@ type NodeControllerConfig struct {
 	NodeName                              string
 	NodePort                              int
 	Stages                                []*internalversion.Stage
-	LockNodeParallelism                   int
+	PlayStageParallelism                  uint
 	FuncMap                               template.FuncMap
 	Recorder                              record.EventRecorder
 }
@@ -130,6 +131,10 @@ type NodeInfo struct {
 
 // NewNodeController creates a new fake nodes controller
 func NewNodeController(conf NodeControllerConfig) (*NodeController, error) {
+	if conf.PlayStageParallelism <= 0 {
+		return nil, fmt.Errorf("playStageParallelism must be greater than 0")
+	}
+
 	disregardStatusWithAnnotationSelector, err := labelsParse(conf.DisregardStatusWithAnnotationSelector)
 	if err != nil {
 		return nil, err
@@ -151,14 +156,15 @@ func NewNodeController(conf NodeControllerConfig) (*NodeController, error) {
 		disregardStatusWithAnnotationSelector: disregardStatusWithAnnotationSelector,
 		disregardStatusWithLabelSelector:      disregardStatusWithLabelSelector,
 		manageNodesWithLabelSelector:          conf.ManageNodesWithLabelSelector,
-		lockPodsOnNodeFunc:                    conf.LockPodsOnNodeFunc,
+		onNodeManagedFunc:                     conf.OnNodeManagedFunc,
 		nodeIP:                                conf.NodeIP,
 		nodeName:                              conf.NodeName,
 		nodePort:                              conf.NodePort,
 		cronjob:                               cron.NewCron(),
 		lifecycle:                             lifecycles,
-		parallelTasks:                         newParallelTasks(conf.LockNodeParallelism),
-		nodeChan:                              make(chan *corev1.Node),
+		playStageParallelism:                  conf.PlayStageParallelism,
+		preprocessChan:                        make(chan *corev1.Node),
+		playStageChan:                         make(chan resourceStageJob[*corev1.Node]),
 		recorder:                              conf.Recorder,
 	}
 	funcMap := template.FuncMap{
@@ -179,28 +185,31 @@ func NewNodeController(conf NodeControllerConfig) (*NodeController, error) {
 // Start starts the fake nodes controller
 // if nodeSelectorFunc is not nil, it will use it to determine if the node should be managed
 func (c *NodeController) Start(ctx context.Context) error {
-	go c.LockNodes(ctx, c.nodeChan)
+	go c.preprocessWorker(ctx)
+	for i := uint(0); i < c.playStageParallelism; i++ {
+		go c.playStageWorker(ctx)
+	}
 
 	opt := metav1.ListOptions{
 		LabelSelector: c.manageNodesWithLabelSelector,
 	}
-	err := c.WatchNodes(ctx, c.nodeChan, opt)
+	err := c.watchResources(ctx, opt)
 	if err != nil {
-		return fmt.Errorf("failed watch node: %w", err)
+		return fmt.Errorf("failed watch nodes: %w", err)
 	}
 
 	logger := log.FromContext(ctx)
 	go func() {
-		err = c.ListNodes(ctx, c.nodeChan, opt)
+		err = c.listResources(ctx, opt)
 		if err != nil {
-			logger.Error("Failed list node", err)
+			logger.Error("Failed list nodes", err)
 		}
 	}()
 
 	return nil
 }
 
-func (c *NodeController) needLockNode(node *corev1.Node) bool {
+func (c *NodeController) need(node *corev1.Node) bool {
 	if !c.nodeSelectorFunc(node) {
 		return false
 	}
@@ -218,8 +227,8 @@ func (c *NodeController) needLockNode(node *corev1.Node) bool {
 	return true
 }
 
-// WatchNodes watch nodes put into the channel
-func (c *NodeController) WatchNodes(ctx context.Context, ch chan<- *corev1.Node, opt metav1.ListOptions) error {
+// watchResources watch resources and send to preprocessChan
+func (c *NodeController) watchResources(ctx context.Context, opt metav1.ListOptions) error {
 	// Watch nodes in the cluster
 	watcher, err := c.clientSet.CoreV1().Nodes().Watch(ctx, opt)
 	if err != nil {
@@ -252,13 +261,13 @@ func (c *NodeController) WatchNodes(ctx context.Context, ch chan<- *corev1.Node,
 				switch event.Type {
 				case watch.Added:
 					node := event.Object.(*corev1.Node)
-					if c.needLockNode(node) {
+					if c.need(node) {
 						c.putNodeInfo(node)
-						ch <- node
-						if c.lockPodsOnNodeFunc != nil {
-							err = c.lockPodsOnNodeFunc(ctx, node.Name)
+						c.preprocessChan <- node
+						if c.onNodeManagedFunc != nil {
+							err = c.onNodeManagedFunc(ctx, node.Name)
 							if err != nil {
-								logger.Error("Failed to lock pods on node", err,
+								logger.Error("Failed to onNodeManagedFunc", err,
 									"node", node.Name,
 								)
 							}
@@ -266,9 +275,9 @@ func (c *NodeController) WatchNodes(ctx context.Context, ch chan<- *corev1.Node,
 					}
 				case watch.Modified:
 					node := event.Object.(*corev1.Node)
-					if c.needLockNode(node) {
+					if c.need(node) {
 						c.putNodeInfo(node)
-						ch <- node
+						c.preprocessChan <- node
 					}
 				case watch.Deleted:
 					node := event.Object.(*corev1.Node)
@@ -293,36 +302,23 @@ func (c *NodeController) WatchNodes(ctx context.Context, ch chan<- *corev1.Node,
 	return nil
 }
 
-// ListNodes list nodes put into the channel
-func (c *NodeController) ListNodes(ctx context.Context, ch chan<- *corev1.Node, opt metav1.ListOptions) error {
+// listResources lists all resources and sends to preprocessChan
+func (c *NodeController) listResources(ctx context.Context, opt metav1.ListOptions) error {
 	listPager := pager.New(func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
 		return c.clientSet.CoreV1().Nodes().List(ctx, opts)
 	})
 	return listPager.EachListItem(ctx, opt, func(obj runtime.Object) error {
 		node := obj.(*corev1.Node)
-		if c.needLockNode(node) {
+		if c.need(node) {
 			c.putNodeInfo(node)
-			ch <- node
+			c.preprocessChan <- node
 		}
 		return nil
 	})
 }
 
-// LockNodes locks a nodes from the channel
-func (c *NodeController) LockNodes(ctx context.Context, nodes <-chan *corev1.Node) {
-	logger := log.FromContext(ctx)
-	for node := range nodes {
-		err := c.LockNode(ctx, node)
-		if err != nil {
-			logger.Error("Failed to lock node", err,
-				"node", node.Name,
-			)
-		}
-	}
-}
-
-// FinalizersModify modify finalizers of node
-func (c *NodeController) FinalizersModify(ctx context.Context, node *corev1.Node, finalizers *internalversion.StageFinalizers) (*corev1.Node, error) {
+// finalizersModify modify finalizers of node
+func (c *NodeController) finalizersModify(ctx context.Context, node *corev1.Node, finalizers *internalversion.StageFinalizers) (*corev1.Node, error) {
 	ops := finalizersModify(node.Finalizers, finalizers)
 	if len(ops) == 0 {
 		return nil, nil
@@ -350,8 +346,8 @@ func (c *NodeController) FinalizersModify(ctx context.Context, node *corev1.Node
 	return result, nil
 }
 
-// DeleteNode deletes a node
-func (c *NodeController) DeleteNode(ctx context.Context, node *corev1.Node) error {
+// deleteResource deletes a node
+func (c *NodeController) deleteResource(ctx context.Context, node *corev1.Node) error {
 	logger := log.FromContext(ctx)
 	logger = logger.With(
 		"node", node.Name,
@@ -371,9 +367,23 @@ func (c *NodeController) DeleteNode(ctx context.Context, node *corev1.Node) erro
 	return nil
 }
 
-// LockNode locks a given node
-func (c *NodeController) LockNode(ctx context.Context, node *corev1.Node) error {
+// preprocessWorker receives the resource from the preprocessChan and preprocess it
+func (c *NodeController) preprocessWorker(ctx context.Context) {
+	logger := log.FromContext(ctx)
+	for node := range c.preprocessChan {
+		err := c.preprocess(ctx, node)
+		if err != nil {
+			logger.Error("Failed to preprocess node", err,
+				"node", node.Name,
+			)
+		}
+	}
+}
+
+// preprocess the pod and send it to the playStageWorker
+func (c *NodeController) preprocess(ctx context.Context, node *corev1.Node) error {
 	key := node.Name
+
 	resourceJob, ok := c.delayJobs.Load(key)
 	if ok && resourceJob.ResourceVersion == node.ResourceVersion {
 		return nil
@@ -381,8 +391,9 @@ func (c *NodeController) LockNode(ctx context.Context, node *corev1.Node) error
 
 	logger := log.FromContext(ctx)
 	logger = logger.With(
-		"node", node.Name,
+		"node", key,
 	)
+
 	data, err := expression.ToJSONStandard(node)
 	if err != nil {
 		return err
@@ -414,9 +425,10 @@ func (c *NodeController) LockNode(ctx context.Context, node *corev1.Node) error
 		if ok {
 			resourceJob.Cancel()
 		}
-		c.parallelTasks.Add(func() {
-			c.playStage(ctx, node, stage)
-		})
+		c.playStageChan <- resourceStageJob[*corev1.Node]{
+			Resource: node,
+			Stage:    stage,
+		}
 	})
 	if ok {
 		resourceJob, ok := c.delayJobs.LoadOrStore(key, jobInfo{
@@ -430,6 +442,14 @@ func (c *NodeController) LockNode(ctx context.Context, node *corev1.Node) error
 	return nil
 }
 
+// playStageWorker receives the resource from the playStageChan and play the stage
+func (c *NodeController) playStageWorker(ctx context.Context) {
+	for node := range c.playStageChan {
+		c.playStage(ctx, node.Resource, node.Stage)
+	}
+}
+
+// playStage plays the stage
 func (c *NodeController) playStage(ctx context.Context, node *corev1.Node, stage *LifecycleStage) {
 	next := stage.Next()
 	logger := log.FromContext(ctx)
@@ -442,16 +462,16 @@ func (c *NodeController) playStage(ctx context.Context, node *corev1.Node, stage
 		}, next.Event.Type, next.Event.Reason, next.Event.Message)
 	}
 	if next.Finalizers != nil {
-		result, err := c.FinalizersModify(ctx, node, next.Finalizers)
+		result, err := c.finalizersModify(ctx, node, next.Finalizers)
 		if err != nil {
 			logger.Error("Failed to finalizers of node", err)
 		}
 		if result != nil && stage.ImmediateNextStage() {
-			c.nodeChan <- result
+			c.preprocessChan <- result
 		}
 	}
 	if next.Delete {
-		err := c.DeleteNode(ctx, node)
+		err := c.deleteResource(ctx, node)
 		if err != nil {
 			logger.Error("Failed to delete node", err)
 		}
@@ -466,18 +486,19 @@ func (c *NodeController) playStage(ctx context.Context, node *corev1.Node, stage
 				"reason", "do not need to modify",
 			)
 		} else {
-			result, err := c.lockNode(ctx, node, patch)
+			result, err := c.patchResource(ctx, node, patch)
 			if err != nil {
-				logger.Error("Failed to lock node", err)
+				logger.Error("Failed to patch node", err)
 			}
 			if result != nil && stage.ImmediateNextStage() {
-				c.nodeChan <- result
+				c.preprocessChan <- result
 			}
 		}
 	}
 }
 
-func (c *NodeController) lockNode(ctx context.Context, node *corev1.Node, patch []byte) (*corev1.Node, error) {
+// patchResource patches the resource
+func (c *NodeController) patchResource(ctx context.Context, node *corev1.Node, patch []byte) (*corev1.Node, error) {
 	logger := log.FromContext(ctx)
 	logger = logger.With(
 		"node", node.Name,
@@ -492,7 +513,7 @@ func (c *NodeController) lockNode(ctx context.Context, node *corev1.Node, patch
 		}
 		return nil, err
 	}
-	logger.Info("Lock node")
+	logger.Info("Patch node")
 	return result, nil
 }
 
diff --git a/pkg/kwok/controllers/pod_controller.go b/pkg/kwok/controllers/pod_controller.go
index 3518fd9c..20659e64 100644
--- a/pkg/kwok/controllers/pod_controller.go
+++ b/pkg/kwok/controllers/pod_controller.go
@@ -61,8 +61,9 @@ type PodController struct {
 	nodeGetFunc                           func(nodeName string) (*NodeInfo, bool)
 	ipPools                               maps.SyncMap[string, *ipPool]
 	renderer                              *renderer
-	lockPodChan                           chan *corev1.Pod
-	parallelTasks                         *parallelTasks
+	preprocessChan                        chan *corev1.Pod
+	playStageChan                         chan resourceStageJob[*corev1.Pod]
+	playStageParallelism                  uint
 	lifecycle                             Lifecycle
 	cronjob                               *cron.Cron
 	delayJobs                             jobInfoMap
@@ -79,13 +80,17 @@ type PodControllerConfig struct {
 	CIDR                                  string
 	NodeGetFunc                           func(nodeName string) (*NodeInfo, bool)
 	Stages                                []*internalversion.Stage
-	LockPodParallelism                    int
+	PlayStageParallelism                  uint
 	FuncMap                               template.FuncMap
 	Recorder                              record.EventRecorder
 }
 
 // NewPodController creates a new fake pods controller
 func NewPodController(conf PodControllerConfig) (*PodController, error) {
+	if conf.PlayStageParallelism <= 0 {
+		return nil, fmt.Errorf("playStageParallelism must be greater than 0")
+	}
+
 	disregardStatusWithAnnotationSelector, err := labelsParse(conf.DisregardStatusWithAnnotationSelector)
 	if err != nil {
 		return nil, err
@@ -111,8 +116,9 @@ func NewPodController(conf PodControllerConfig) (*PodController, error) {
 		nodeGetFunc:                           conf.NodeGetFunc,
 		cronjob:                               cron.NewCron(),
 		lifecycle:                             lifecycles,
-		parallelTasks:                         newParallelTasks(conf.LockPodParallelism),
-		lockPodChan:                           make(chan *corev1.Pod),
+		playStageParallelism:                  conf.PlayStageParallelism,
+		preprocessChan:                        make(chan *corev1.Pod),
+		playStageChan:                         make(chan resourceStageJob[*corev1.Pod]),
 		recorder:                              conf.Recorder,
 	}
 	funcMap := template.FuncMap{
@@ -131,19 +137,22 @@ func NewPodController(conf PodControllerConfig) (*PodController, error) {
 // Start starts the fake pod controller
 // It will modify the pods status to we want
 func (c *PodController) Start(ctx context.Context) error {
-	go c.LockPods(ctx, c.lockPodChan)
+	go c.preprocessWorker(ctx)
+	for i := uint(0); i < c.playStageParallelism; i++ {
+		go c.playStageWorker(ctx)
+	}
 
 	opt := metav1.ListOptions{
 		FieldSelector: podFieldSelector,
 	}
-	err := c.WatchPods(ctx, c.lockPodChan, opt)
+	err := c.watchResources(ctx, opt)
 	if err != nil {
 		return fmt.Errorf("failed watch pods: %w", err)
 	}
 
 	logger := log.FromContext(ctx)
 	go func() {
-		err = c.ListPods(ctx, c.lockPodChan, opt)
+		err = c.listResources(ctx, opt)
 		if err != nil {
 			logger.Error("Failed list pods", err)
 		}
@@ -151,8 +160,8 @@ func (c *PodController) Start(ctx context.Context) error {
 	return nil
 }
 
-// FinalizersModify modify the finalizers of the pod
-func (c *PodController) FinalizersModify(ctx context.Context, pod *corev1.Pod, finalizers *internalversion.StageFinalizers) (*corev1.Pod, error) {
+// finalizersModify modify the finalizers of the pod
+func (c *PodController) finalizersModify(ctx context.Context, pod *corev1.Pod, finalizers *internalversion.StageFinalizers) (*corev1.Pod, error) {
 	ops := finalizersModify(pod.Finalizers, finalizers)
 	if len(ops) == 0 {
 		return nil, nil
@@ -181,8 +190,8 @@ func (c *PodController) FinalizersModify(ctx context.Context, pod *corev1.Pod, f
 	return result, nil
 }
 
-// DeletePod deletes a pod
-func (c *PodController) DeletePod(ctx context.Context, pod *corev1.Pod) error {
+// deleteResource deletes a pod
+func (c *PodController) deleteResource(ctx context.Context, pod *corev1.Pod) error {
 	logger := log.FromContext(ctx)
 	logger = logger.With(
 		"pod", log.KObj(pod),
@@ -203,8 +212,22 @@ func (c *PodController) DeletePod(ctx context.Context, pod *corev1.Pod) error {
 	return nil
 }
 
-// LockPod locks a given pod
-func (c *PodController) LockPod(ctx context.Context, pod *corev1.Pod) error {
+// preprocessWorker receives the resource from the preprocessChan and preprocess it
+func (c *PodController) preprocessWorker(ctx context.Context) {
+	logger := log.FromContext(ctx)
+	for pod := range c.preprocessChan {
+		err := c.preprocess(ctx, pod)
+		if err != nil {
+			logger.Error("Failed to preprocess node", err,
+				"pod", log.KObj(pod),
+				"node", pod.Spec.NodeName,
+			)
+		}
+	}
+}
+
+// preprocess the pod and send it to the playStageWorker
+func (c *PodController) preprocess(ctx context.Context, pod *corev1.Pod) error {
 	key := log.KObj(pod).String()
 
 	resourceJob, ok := c.delayJobs.Load(key)
@@ -217,6 +240,7 @@ func (c *PodController) LockPod(ctx context.Context, pod *corev1.Pod) error {
 		"pod", key,
 		"node", pod.Spec.NodeName,
 	)
+
 	data, err := expression.ToJSONStandard(pod)
 	if err != nil {
 		return err
@@ -248,9 +272,10 @@ func (c *PodController) LockPod(ctx context.Context, pod *corev1.Pod) error {
 		if ok {
 			resourceJob.Cancel()
 		}
-		c.parallelTasks.Add(func() {
-			c.playStage(ctx, pod, stage)
-		})
+		c.playStageChan <- resourceStageJob[*corev1.Pod]{
+			Resource: pod,
+			Stage:    stage,
+		}
 	})
 	if ok {
 		resourceJob, ok := c.delayJobs.LoadOrStore(key, jobInfo{
@@ -265,6 +290,14 @@ func (c *PodController) LockPod(ctx context.Context, pod *corev1.Pod) error {
 	return nil
 }
 
+// playStageWorker receives the resource from the playStageChan and play the stage
+func (c *PodController) playStageWorker(ctx context.Context) {
+	for pod := range c.playStageChan {
+		c.playStage(ctx, pod.Resource, pod.Stage)
+	}
+}
+
+// playStage plays the stage
 func (c *PodController) playStage(ctx context.Context, pod *corev1.Pod, stage *LifecycleStage) {
 	next := stage.Next()
 	logger := log.FromContext(ctx)
@@ -277,21 +310,21 @@ func (c *PodController) playStage(ctx context.Context, pod *corev1.Pod, stage *L
 		}, next.Event.Type, next.Event.Reason, next.Event.Message)
 	}
 	if next.Finalizers != nil {
-		result, err := c.FinalizersModify(ctx, pod, next.Finalizers)
+		result, err := c.finalizersModify(ctx, pod, next.Finalizers)
 		if err != nil {
 			logger.Error("Failed to finalizers", err)
 		}
 		if result != nil && stage.ImmediateNextStage() {
-			c.lockPodChan <- result
+			c.preprocessChan <- result
 		}
 	}
 	if next.Delete {
-		err := c.DeletePod(ctx, pod)
+		err := c.deleteResource(ctx, pod)
 		if err != nil {
 			logger.Error("Failed to delete pod", err)
 		}
 	} else if next.StatusTemplate != "" {
-		patch, err := c.configurePod(pod, next.StatusTemplate)
+		patch, err := c.configureResource(pod, next.StatusTemplate)
 		if err != nil {
 			logger.Error("Failed to configure pod", err)
 			return
@@ -301,18 +334,19 @@ func (c *PodController) playStage(ctx context.Context, pod *corev1.Pod, stage *L
 				"reason", "do not need to modify",
 			)
 		} else {
-			result, err := c.lockPod(ctx, pod, patch)
+			result, err := c.patchResource(ctx, pod, patch)
 			if err != nil {
-				logger.Error("Failed to lock pod", err)
+				logger.Error("Failed to patch node", err)
 			}
 			if result != nil && stage.ImmediateNextStage() {
-				c.lockPodChan <- result
+				c.preprocessChan <- result
 			}
 		}
 	}
 }
 
-func (c *PodController) lockPod(ctx context.Context, pod *corev1.Pod, patch []byte) (*corev1.Pod, error) {
+// patchResource patches the resource
+func (c *PodController) patchResource(ctx context.Context, pod *corev1.Pod, patch []byte) (*corev1.Pod, error) {
 	logger := log.FromContext(ctx)
 	logger = logger.With(
 		"pod", log.KObj(pod),
@@ -328,25 +362,11 @@ func (c *PodController) lockPod(ctx context.Context, pod *corev1.Pod, patch []by
 		}
 		return nil, err
 	}
-	logger.Info("Lock pod")
+	logger.Info("Patch pod")
 	return result, nil
 }
 
-// LockPods locks a pods from the channel
-func (c *PodController) LockPods(ctx context.Context, pods <-chan *corev1.Pod) {
-	logger := log.FromContext(ctx)
-	for pod := range pods {
-		err := c.LockPod(ctx, pod)
-		if err != nil {
-			logger.Error("Failed to lock pod", err,
-				"pod", log.KObj(pod),
-				"node", pod.Spec.NodeName,
-			)
-		}
-	}
-}
-
-func (c *PodController) needLockPod(pod *corev1.Pod) bool {
+func (c *PodController) need(pod *corev1.Pod) bool {
 	if _, has := c.nodeGetFunc(pod.Spec.NodeName); !has {
 		return false
 	}
@@ -365,8 +385,8 @@ func (c *PodController) needLockPod(pod *corev1.Pod) bool {
 	return true
 }
 
-// WatchPods watch pods put into the channel
-func (c *PodController) WatchPods(ctx context.Context, lockChan chan<- *corev1.Pod, opt metav1.ListOptions) error {
+// watchResources watch resources and send to preprocessChan
+func (c *PodController) watchResources(ctx context.Context, opt metav1.ListOptions) error {
 	watcher, err := c.clientSet.CoreV1().Pods(corev1.NamespaceAll).Watch(ctx, opt)
 	if err != nil {
 		return err
@@ -399,8 +419,8 @@ func (c *PodController) WatchPods(ctx context.Context, lockChan chan<- *corev1.P
 				switch event.Type {
 				case watch.Added, watch.Modified:
 					pod := event.Object.(*corev1.Pod)
-					if c.needLockPod(pod) {
-						lockChan <- pod.DeepCopy()
+					if c.need(pod) {
+						c.preprocessChan <- pod.DeepCopy()
 					} else {
 						logger.Debug("Skip pod",
 							"reason", "not managed",
@@ -412,7 +432,7 @@ func (c *PodController) WatchPods(ctx context.Context, lockChan chan<- *corev1.P
 
 				case watch.Deleted:
 					pod := event.Object.(*corev1.Pod)
-					if c.needLockPod(pod) {
+					if c.need(pod) {
 						// Recycling PodIP
 						c.recyclingPodIP(ctx, pod)
 
@@ -435,23 +455,23 @@ func (c *PodController) WatchPods(ctx context.Context, lockChan chan<- *corev1.P
 	return nil
 }
 
-// ListPods list pods put into the channel
-func (c *PodController) ListPods(ctx context.Context, ch chan<- *corev1.Pod, opt metav1.ListOptions) error {
+// listResources lists all resources and sends to preprocessChan
+func (c *PodController) listResources(ctx context.Context, opt metav1.ListOptions) error {
 	listPager := pager.New(func(ctx context.Context, opts metav1.ListOptions) (runtime.Object, error) {
 		return c.clientSet.CoreV1().Pods(corev1.NamespaceAll).List(ctx, opts)
 	})
 	return listPager.EachListItem(ctx, opt, func(obj runtime.Object) error {
 		pod := obj.(*corev1.Pod)
-		if c.needLockPod(pod) {
-			ch <- pod.DeepCopy()
+		if c.need(pod) {
+			c.preprocessChan <- pod.DeepCopy()
 		}
 		return nil
 	})
 }
 
-// LockPodsOnNode locks pods on the node
-func (c *PodController) LockPodsOnNode(ctx context.Context, nodeName string) error {
-	return c.ListPods(ctx, c.lockPodChan, metav1.ListOptions{
+// PlayStagePodsOnNode plays stage pods on node
+func (c *PodController) PlayStagePodsOnNode(ctx context.Context, nodeName string) error {
+	return c.listResources(ctx, metav1.ListOptions{
 		FieldSelector: fields.OneTermEqualSelector("spec.nodeName", nodeName).String(),
 	})
 }
@@ -510,7 +530,7 @@ func (c *PodController) recyclingPodIP(ctx context.Context, pod *corev1.Pod) {
 	}
 }
 
-func (c *PodController) configurePod(pod *corev1.Pod, template string) ([]byte, error) {
+func (c *PodController) configureResource(pod *corev1.Pod, template string) ([]byte, error) {
 	if !c.enableCNI {
 		// Mark the pod IP that existed before the kubelet was started
 		if nodeInfo, has := c.nodeGetFunc(pod.Spec.NodeName); has {
diff --git a/pkg/kwok/controllers/utils.go b/pkg/kwok/controllers/utils.go
index c7a3950a..56ca2113 100644
--- a/pkg/kwok/controllers/utils.go
+++ b/pkg/kwok/controllers/utils.go
@@ -20,7 +20,6 @@ import (
 	"encoding/binary"
 	"net"
 	"sync"
-	"time"
 
 	"github.com/wzshiming/cron"
 	"k8s.io/apimachinery/pkg/labels"
@@ -119,50 +118,6 @@ func (i *ipPool) Use(ip string) {
 	i.used[ip] = struct{}{}
 }
 
-type parallelTasks struct {
-	wg     sync.WaitGroup
-	bucket chan struct{}
-	tasks  chan func()
-}
-
-func newParallelTasks(n int) *parallelTasks {
-	return &parallelTasks{
-		bucket: make(chan struct{}, n),
-		tasks:  make(chan func()),
-	}
-}
-
-func (p *parallelTasks) Add(fun func()) {
-	p.wg.Add(1)
-	select {
-	case p.tasks <- fun: // there are idle threads
-	case p.bucket <- struct{}{}: // there are free threads
-		go p.fork()
-		p.tasks <- fun
-	}
-}
-
-func (p *parallelTasks) fork() {
-	defer func() {
-		<-p.bucket
-	}()
-	timer := time.NewTimer(time.Second / 2)
-	for {
-		select {
-		case <-timer.C: // idle threads
-			return
-		case fun := <-p.tasks:
-			fun()
-			p.wg.Done()
-			timer.Reset(time.Second / 2)
-		}
-	}
-}
-
-func (p *parallelTasks) Wait() {
-	p.wg.Wait()
-}
-
 func labelsParse(selector string) (labels.Selector, error) {
 	if selector == "" {
 		return nil, nil
@@ -176,3 +131,8 @@ type jobInfo struct {
 	ResourceVersion string
 	Cancel          cron.DoFunc
 }
+
+type resourceStageJob[T any] struct {
+	Resource T
+	Stage    *LifecycleStage
+}


Test patch:
diff --git a/pkg/kwok/controllers/controller_test.go b/pkg/kwok/controllers/controller_test.go
index 4307041b..3777ec20 100644
--- a/pkg/kwok/controllers/controller_test.go
+++ b/pkg/kwok/controllers/controller_test.go
@@ -79,11 +79,13 @@ func TestController(t *testing.T) {
 		{
 			name: "node controller test: manage all nodes",
 			conf: Config{
-				ClientSet:      fake.NewSimpleClientset(nodes...),
-				ManageAllNodes: true,
-				NodeStages:     nodeStages,
-				PodStages:      podStages,
-				CIDR:           "10.0.0.1/24",
+				ClientSet:                fake.NewSimpleClientset(nodes...),
+				ManageAllNodes:           true,
+				NodeStages:               nodeStages,
+				PodStages:                podStages,
+				CIDR:                     "10.0.0.1/24",
+				NodePlayStageParallelism: 1,
+				PodPlayStageParallelism:  1,
 			},
 			wantNodePhase: map[string]corev1.NodePhase{
 				"node-0": corev1.NodeRunning,
@@ -100,6 +102,8 @@ func TestController(t *testing.T) {
 				NodeStages:                   nodeStages,
 				PodStages:                    podStages,
 				CIDR:                         "10.0.0.1/24",
+				NodePlayStageParallelism:     1,
+				PodPlayStageParallelism:      1,
 			},
 			wantNodePhase: map[string]corev1.NodePhase{
 				"node-0": corev1.NodeRunning,
@@ -116,6 +120,8 @@ func TestController(t *testing.T) {
 				NodeStages:                        nodeStages,
 				PodStages:                         podStages,
 				CIDR:                              "10.0.0.1/24",
+				NodePlayStageParallelism:          1,
+				PodPlayStageParallelism:           1,
 			},
 			wantNodePhase: map[string]corev1.NodePhase{
 				"node-0": corev1.NodePending,
diff --git a/pkg/kwok/controllers/node_controller_test.go b/pkg/kwok/controllers/node_controller_test.go
index bbb6c4a4..82b26485 100644
--- a/pkg/kwok/controllers/node_controller_test.go
+++ b/pkg/kwok/controllers/node_controller_test.go
@@ -70,12 +70,12 @@ func TestNodeController(t *testing.T) {
 	}
 	nodeStageStatus, _ := NewStagesFromYaml([]byte(stages.DefaultNodeStages))
 	nodes, err := NewNodeController(NodeControllerConfig{
-		ClientSet:           clientset,
-		NodeIP:              "10.0.0.1",
-		NodeSelectorFunc:    nodeSelectorFunc,
-		Stages:              nodeStageStatus,
-		FuncMap:             defaultFuncMap,
-		LockNodeParallelism: 2,
+		ClientSet:            clientset,
+		NodeIP:               "10.0.0.1",
+		NodeSelectorFunc:     nodeSelectorFunc,
+		Stages:               nodeStageStatus,
+		FuncMap:              defaultFuncMap,
+		PlayStageParallelism: 2,
 	})
 	if err != nil {
 		t.Fatal(fmt.Errorf("new nodes controller error: %w", err))
diff --git a/pkg/kwok/controllers/pod_controller_test.go b/pkg/kwok/controllers/pod_controller_test.go
index 6bf7bce8..b46607da 100644
--- a/pkg/kwok/controllers/pod_controller_test.go
+++ b/pkg/kwok/controllers/pod_controller_test.go
@@ -170,7 +170,7 @@ func TestPodController(t *testing.T) {
 		Stages:                                podStageStatus,
 		NodeGetFunc:                           nodeGetFunc,
 		FuncMap:                               defaultFuncMap,
-		LockPodParallelism:                    2,
+		PlayStageParallelism:                  2,
 	})
 	if err != nil {
 		t.Fatal(fmt.Errorf("new pods controller error: %w", err))
@@ -273,7 +273,7 @@ func TestPodController(t *testing.T) {
 				if pod.Status.Phase != corev1.PodRunning {
 					return false, fmt.Errorf("want pod %s phase is running, got %s", pod.Name, pod.Status.Phase)
 				}
-				if pods.needLockPod(&list.Items[index]) {
+				if pods.need(&list.Items[index]) {
 					if pod.Status.HostIP != nodeInfo.HostIPs[0] {
 						return false, fmt.Errorf("want pod %s hostIP=%s, got %s", pod.Name, nodeInfo.HostIPs[0], pod.Status.HostIP)
 					}
diff --git a/pkg/kwok/controllers/utils_test.go b/pkg/kwok/controllers/utils_test.go
index 8a12c792..080cff4f 100644
--- a/pkg/kwok/controllers/utils_test.go
+++ b/pkg/kwok/controllers/utils_test.go
@@ -20,28 +20,8 @@ import (
 	"net"
 	"reflect"
 	"testing"
-	"time"
 )
 
-func TestParallelTasks(t *testing.T) {
-	tasks := newParallelTasks(4)
-	startTime := time.Now()
-	for i := 0; i < 10; i++ {
-		tasks.Add(func() {
-			time.Sleep(1 * time.Second)
-		})
-	}
-
-	tasks.Wait()
-	elapsed := time.Since(startTime)
-	if elapsed >= 4*time.Second {
-		t.Fatalf("Tasks took too long to complete: %v", elapsed)
-	} else if elapsed < 3*time.Second {
-		t.Fatalf("Tasks completed too quickly: %v", elapsed)
-	}
-	t.Logf("Tasks completed in %v", elapsed)
-}
-
 func Test_parseCIDR(t *testing.T) {
 	type args struct {
 		s string