Debug

kubernetes-sigs · Nov 27, 2024 · 60d25d0 · 60d25d0
1 parent c55737a
commit 60d25d0
Show file tree

Hide file tree

Showing 5 changed files with 16 additions and 7 deletions.
diff --git a/pkg/kwok/controllers/controller.go b/pkg/kwok/controllers/controller.go
@@ -95,6 +95,9 @@ type Controller struct {
 
 	podOnNodeManageQueue queue.Queue[string]
 	nodeManageQueue      queue.Queue[string]
+
+	podOnNodeManageQueueParallelism *queue.Parallelism[string]
+	nodeManageQueueParallelism      *queue.Parallelism[string]
 }
 
 // Config is the configuration for the controller
@@ -295,6 +298,7 @@ func (c *Controller) initNodeLeaseController(ctx context.Context) error {
 		c.nodeLeases.ReleaseHold(nodeName)
 	}
 
+	c.nodeManageQueueParallelism = queue.NewParallelismQueue(ctx, c.nodeManageQueue, c.nodeLeaseSyncWorker)
 	go c.nodeLeaseSyncWorker(ctx)
 
 	err = c.nodeLeases.Start(ctx)
@@ -307,7 +311,7 @@ func (c *Controller) initNodeLeaseController(ctx context.Context) error {
 func (c *Controller) nodeLeaseSyncWorker(ctx context.Context) {
 	logger := log.FromContext(ctx)
 	for ctx.Err() == nil {
-		nodeName, ok := c.nodeManageQueue.GetOrWaitWithDone(ctx.Done())
+		nodeName, ok := c.nodeManageQueueParallelism.GetOrWaitWithDone(ctx.Done())
 		if !ok {
 			return
 		}
@@ -339,6 +343,7 @@ func (c *Controller) startStageController(ctx context.Context, ref internalversi
 			return fmt.Errorf("failed to init pod controller: %w", err)
 		}
 
+		c.podOnNodeManageQueueParallelism = queue.NewParallelismQueue(ctx, c.podOnNodeManageQueue, c.podsOnNodeSyncWorker)
 		go c.podsOnNodeSyncWorker(ctx)
 
 	case nodeRef:
@@ -559,7 +564,7 @@ func (c *Controller) Start(ctx context.Context) error {
 func (c *Controller) podsOnNodeSyncWorker(ctx context.Context) {
 	logger := log.FromContext(ctx)
 	for ctx.Err() == nil {
-		nodeName, ok := c.podOnNodeManageQueue.GetOrWaitWithDone(ctx.Done())
+		nodeName, ok := c.podOnNodeManageQueueParallelism.GetOrWaitWithDone(ctx.Done())
 		if !ok {
 			return
 		}

diff --git a/pkg/kwok/controllers/node_controller.go b/pkg/kwok/controllers/node_controller.go
@@ -145,6 +145,9 @@ func NewNodeController(conf NodeControllerConfig) (*NodeController, error) {
 func (c *NodeController) Start(ctx context.Context, events <-chan informer.Event[*corev1.Node]) error {
 	go c.preprocessWorker(ctx)
 	c.parallelism = queue.NewParallelismQueue(ctx, c.delayQueue, c.playStageWorker)
+	for i := uint(0); i < c.playStageParallelism; i++ {
+		go c.playStageWorker(ctx)
+	}
 	go c.watchResources(ctx, events)
 	return nil
 }

diff --git a/pkg/kwok/controllers/node_lease_controller.go b/pkg/kwok/controllers/node_lease_controller.go
@@ -101,6 +101,9 @@ func NewNodeLeaseController(conf NodeLeaseControllerConfig) (*NodeLeaseControlle
 // Start starts the NodeLeaseController
 func (c *NodeLeaseController) Start(ctx context.Context) error {
 	c.parallelism = queue.NewParallelismQueue(ctx, c.delayQueue, c.syncWorker)
+	for i := uint(0); i < c.leaseParallelism; i++ {
+		go c.syncWorker(ctx)
+	}
 	return nil
 }
 

diff --git a/pkg/kwok/controllers/pod_controller.go b/pkg/kwok/controllers/pod_controller.go
@@ -150,6 +150,9 @@ func NewPodController(conf PodControllerConfig) (*PodController, error) {
 func (c *PodController) Start(ctx context.Context, events <-chan informer.Event[*corev1.Pod]) error {
 	go c.preprocessWorker(ctx)
 	c.parallelism = queue.NewParallelismQueue(ctx, c.delayQueue, c.playStageWorker)
+	for i := uint(0); i < c.playStageParallelism; i++ {
+		go c.playStageWorker(ctx)
+	}
 	go c.watchResources(ctx, events)
 	return nil
 }

diff --git a/pkg/utils/queue/parallelism.go b/pkg/utils/queue/parallelism.go
@@ -29,11 +29,6 @@ type Parallelism[T any] struct {
 }
 
 func NewParallelismQueue[T any](ctx context.Context, q Queue[T], startFunc func(ctx context.Context)) *Parallelism[T] {
-	go startFunc(ctx)
-	go startFunc(ctx)
-	go startFunc(ctx)
-	go startFunc(ctx)
-
 	return &Parallelism[T]{
 		ctx:         ctx,
 		startFunc:   startFunc,