From 4f7c4a700cde0752e024b25a8c581873e136bea4 Mon Sep 17 00:00:00 2001 From: amirhnajafiz Date: Wed, 6 Aug 2025 10:51:07 -0400 Subject: [PATCH 1/7] Init: worker module in autopilot-daemon/pkg Signed-off-by: amirhnajafiz --- autopilot-daemon/pkg/worker/enum.go | 10 ++++++++++ autopilot-daemon/pkg/worker/pool.go | 24 ++++++++++++++++++++++++ autopilot-daemon/pkg/worker/worker.go | 14 ++++++++++++++ 3 files changed, 48 insertions(+) create mode 100644 autopilot-daemon/pkg/worker/enum.go create mode 100644 autopilot-daemon/pkg/worker/pool.go create mode 100644 autopilot-daemon/pkg/worker/worker.go diff --git a/autopilot-daemon/pkg/worker/enum.go b/autopilot-daemon/pkg/worker/enum.go new file mode 100644 index 0000000..b2ec229 --- /dev/null +++ b/autopilot-daemon/pkg/worker/enum.go @@ -0,0 +1,10 @@ +package worker + +// TaskType represents the type of task to be processed by the worker pool. +// it is either a periodic check or an invasive check. +type TaskType int + +const ( + TaskPeriodicCheck TaskType = iota + 1 + TaskInvasiveCheck +) diff --git a/autopilot-daemon/pkg/worker/pool.go b/autopilot-daemon/pkg/worker/pool.go new file mode 100644 index 0000000..d982a9a --- /dev/null +++ b/autopilot-daemon/pkg/worker/pool.go @@ -0,0 +1,24 @@ +package worker + +// WorkerPool manages a pool of go-routines that process tasks concurrently. +type WorkerPool struct { + taskChannel chan TaskType +} + +// CreateWorkerPool initializes a new WorkerPool with a specified number of workers. +func CreateWorkerPool(numberOfWorkers int) *WorkerPool { + taskChannel := make(chan TaskType) + + for i := 0; i < numberOfWorkers; i++ { + go worker(taskChannel) + } + + return &WorkerPool{ + taskChannel: taskChannel, + } +} + +// Submit adds a task to the worker pool for processing. +func (wp *WorkerPool) Submit(task TaskType) { + wp.taskChannel <- task +} diff --git a/autopilot-daemon/pkg/worker/worker.go b/autopilot-daemon/pkg/worker/worker.go new file mode 100644 index 0000000..66b25f1 --- /dev/null +++ b/autopilot-daemon/pkg/worker/worker.go @@ -0,0 +1,14 @@ +package worker + +func worker(c chan TaskType) { + for task := range c { + switch task { + case TaskPeriodicCheck: + // Handle periodic check + case TaskInvasiveCheck: + // Handle invasive check + default: + // Handle unknown task type + } + } +} From 25e0a88c10d2bd15ce79663112a5e6b398315375 Mon Sep 17 00:00:00 2001 From: amirhnajafiz Date: Wed, 6 Aug 2025 11:05:01 -0400 Subject: [PATCH 2/7] Add: sync.Map in the worker pool struct Signed-off-by: amirhnajafiz --- autopilot-daemon/pkg/worker/pool.go | 25 ++++++++++++++++++++----- autopilot-daemon/pkg/worker/worker.go | 10 +++++++++- 2 files changed, 29 insertions(+), 6 deletions(-) diff --git a/autopilot-daemon/pkg/worker/pool.go b/autopilot-daemon/pkg/worker/pool.go index d982a9a..0452677 100644 --- a/autopilot-daemon/pkg/worker/pool.go +++ b/autopilot-daemon/pkg/worker/pool.go @@ -1,24 +1,39 @@ package worker +import "sync" + // WorkerPool manages a pool of go-routines that process tasks concurrently. type WorkerPool struct { - taskChannel chan TaskType + // runningTasks keeps track of tasks currently being processed + runningTasks *sync.Map + // taskQueue is a channel where tasks are submitted for processing + taskQueue chan TaskType } // CreateWorkerPool initializes a new WorkerPool with a specified number of workers. func CreateWorkerPool(numberOfWorkers int) *WorkerPool { - taskChannel := make(chan TaskType) + syncMap := &sync.Map{} + taskQueue := make(chan TaskType) + // start the specified number of workers for i := 0; i < numberOfWorkers; i++ { - go worker(taskChannel) + go worker(taskQueue, syncMap) } return &WorkerPool{ - taskChannel: taskChannel, + runningTasks: syncMap, + taskQueue: taskQueue, } } // Submit adds a task to the worker pool for processing. func (wp *WorkerPool) Submit(task TaskType) { - wp.taskChannel <- task + // check if the task is running + if _, exists := wp.runningTasks.Load(task); exists { + return // task is already running, do not submit again + } + + // mark the task as running + wp.runningTasks.Store(task, struct{}{}) + wp.taskQueue <- task } diff --git a/autopilot-daemon/pkg/worker/worker.go b/autopilot-daemon/pkg/worker/worker.go index 66b25f1..2e56ef5 100644 --- a/autopilot-daemon/pkg/worker/worker.go +++ b/autopilot-daemon/pkg/worker/worker.go @@ -1,6 +1,11 @@ package worker -func worker(c chan TaskType) { +import "sync" + +// worker is a function that processes tasks from the task queue. +// it runs in a separate goroutine and listens for tasks to process. +// it removes the task from the runningTasks map once completed. +func worker(c chan TaskType, sm *sync.Map) { for task := range c { switch task { case TaskPeriodicCheck: @@ -10,5 +15,8 @@ func worker(c chan TaskType) { default: // Handle unknown task type } + + // mark the task as completed + sm.Delete(task) } } From 9315ef65c9714110ce7b6ce8ac961b2552143ec4 Mon Sep 17 00:00:00 2001 From: amirhnajafiz Date: Wed, 6 Aug 2025 11:10:30 -0400 Subject: [PATCH 3/7] Feat: worker pool module Signed-off-by: amirhnajafiz --- autopilot-daemon/pkg/worker/pool.go | 4 +++- autopilot-daemon/pkg/worker/worker.go | 14 ++++++++++---- 2 files changed, 13 insertions(+), 5 deletions(-) diff --git a/autopilot-daemon/pkg/worker/pool.go b/autopilot-daemon/pkg/worker/pool.go index 0452677..759e1b7 100644 --- a/autopilot-daemon/pkg/worker/pool.go +++ b/autopilot-daemon/pkg/worker/pool.go @@ -1,6 +1,8 @@ package worker -import "sync" +import ( + "sync" +) // WorkerPool manages a pool of go-routines that process tasks concurrently. type WorkerPool struct { diff --git a/autopilot-daemon/pkg/worker/worker.go b/autopilot-daemon/pkg/worker/worker.go index 2e56ef5..ff07532 100644 --- a/autopilot-daemon/pkg/worker/worker.go +++ b/autopilot-daemon/pkg/worker/worker.go @@ -1,6 +1,12 @@ package worker -import "sync" +import ( + "sync" + + "github.com/IBM/autopilot/pkg/healthcheck" + + "k8s.io/klog/v2" +) // worker is a function that processes tasks from the task queue. // it runs in a separate goroutine and listens for tasks to process. @@ -9,11 +15,11 @@ func worker(c chan TaskType, sm *sync.Map) { for task := range c { switch task { case TaskPeriodicCheck: - // Handle periodic check + healthcheck.PeriodicCheck() case TaskInvasiveCheck: - // Handle invasive check + healthcheck.InvasiveCheck() default: - // Handle unknown task type + klog.Errorf("Unknown task type: %v", task) } // mark the task as completed From b2b8458e8e99736341d566cb9325cef66796f02e Mon Sep 17 00:00:00 2001 From: amirhnajafiz Date: Wed, 6 Aug 2025 11:19:50 -0400 Subject: [PATCH 4/7] Update: changed health check execution from sequential to workpool in autopilot-daemon/pkg/cmd/main.go Signed-off-by: amirhnajafiz --- autopilot-daemon/pkg/cmd/main.go | 13 ++++++++++--- autopilot-daemon/pkg/worker/pool.go | 5 +++-- 2 files changed, 13 insertions(+), 5 deletions(-) diff --git a/autopilot-daemon/pkg/cmd/main.go b/autopilot-daemon/pkg/cmd/main.go index aed09c2..e487b1f 100644 --- a/autopilot-daemon/pkg/cmd/main.go +++ b/autopilot-daemon/pkg/cmd/main.go @@ -6,11 +6,13 @@ import ( "fmt" "net/http" "os" + "runtime" "time" "github.com/IBM/autopilot/pkg/handler" "github.com/IBM/autopilot/pkg/healthcheck" "github.com/IBM/autopilot/pkg/utils" + "github.com/IBM/autopilot/pkg/worker" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promhttp" "k8s.io/klog/v2" @@ -122,8 +124,13 @@ func main() { // Create a Watcher over nodes. Needed to export metrics from data created by external jobs (i.e., dcgm Jobs) go utils.WatchNode() + // Create a WorkerPool to handle tasks concurrently + numCPU := runtime.NumCPU() + workerPool := worker.CreateWorkerPool(2 * numCPU) // use 2 workers per CPU core + klog.Infof("Starting WorkerPool with %d workers", 2*numCPU) + // Run the health checks at startup, then start the timer - healthcheck.PeriodicCheck() + workerPool.Submit(worker.TaskPeriodicCheck) // Parse the repeat and invasive intervals to durations repeatDuration, err := utils.ParseInterval(*repeat) @@ -144,10 +151,10 @@ func main() { for { select { case <-periodicChecksTicker.C: - healthcheck.PeriodicCheck() + workerPool.Submit(worker.TaskPeriodicCheck) case <-invasiveChecksTicker.C: if invasiveDuration > 0 { - healthcheck.InvasiveCheck() + workerPool.Submit(worker.TaskInvasiveCheck) } } } diff --git a/autopilot-daemon/pkg/worker/pool.go b/autopilot-daemon/pkg/worker/pool.go index 759e1b7..df6c78d 100644 --- a/autopilot-daemon/pkg/worker/pool.go +++ b/autopilot-daemon/pkg/worker/pool.go @@ -19,7 +19,7 @@ func CreateWorkerPool(numberOfWorkers int) *WorkerPool { // start the specified number of workers for i := 0; i < numberOfWorkers; i++ { - go worker(taskQueue, syncMap) + go worker(taskQueue, syncMap) // start a worker goroutine, see worker.go for implementation } return &WorkerPool{ @@ -35,7 +35,8 @@ func (wp *WorkerPool) Submit(task TaskType) { return // task is already running, do not submit again } - // mark the task as running + // mark the task as running, so it won't be submitted again + // the worker will remove it from runningTasks when done wp.runningTasks.Store(task, struct{}{}) wp.taskQueue <- task } From 47a76ba8793d5a20892bd045b3781f02c8384805 Mon Sep 17 00:00:00 2001 From: amirhnajafiz Date: Wed, 6 Aug 2025 11:49:25 -0400 Subject: [PATCH 5/7] Add: worker limit in configs for health check go-routines Signed-off-by: amirhnajafiz --- autopilot-daemon/pkg/cmd/main.go | 6 ++++++ helm-charts/autopilot/templates/autopilot.yaml | 2 +- helm-charts/autopilot/values.yaml | 3 +++ 3 files changed, 10 insertions(+), 1 deletion(-) diff --git a/autopilot-daemon/pkg/cmd/main.go b/autopilot-daemon/pkg/cmd/main.go index e487b1f..bfc678d 100644 --- a/autopilot-daemon/pkg/cmd/main.go +++ b/autopilot-daemon/pkg/cmd/main.go @@ -25,6 +25,7 @@ func main() { v := flag.String("loglevel", "2", "Log level") repeat := flag.String("w", "24h", "Run all tests periodically on each node. Time set in interval format. Defaults to 24h") invasive := flag.String("invasive-check-timer", "4h", "Run invasive checks (e.g., dcgmi level 3) on each node when GPUs are free. Time set in interval format. Defaults to 4h. Set to 0 to avoid invasive checks") + workersLimit := flag.Int("workers", 0, "Number of workers to use for concurrent tasks. Defaults to 0 which uses 2*number_of_logical_CPU_cores") flag.Parse() @@ -126,6 +127,11 @@ func main() { // Create a WorkerPool to handle tasks concurrently numCPU := runtime.NumCPU() + if *workersLimit > 0 { + // if user has set a limit, use it + numCPU = *workersLimit + } + workerPool := worker.CreateWorkerPool(2 * numCPU) // use 2 workers per CPU core klog.Infof("Starting WorkerPool with %d workers", 2*numCPU) diff --git a/helm-charts/autopilot/templates/autopilot.yaml b/helm-charts/autopilot/templates/autopilot.yaml index 9b042a4..6f3de7d 100644 --- a/helm-charts/autopilot/templates/autopilot.yaml +++ b/helm-charts/autopilot/templates/autopilot.yaml @@ -51,7 +51,7 @@ spec: - sh - -c - | - /usr/local/bin/autopilot --port {{ .Values.service.port }} --loglevel={{ .Values.loglevel }} --bw {{ .Values.PCIeBW }} --w {{ .Values.repeat }} --invasive-check-timer {{ .Values.invasive }} + /usr/local/bin/autopilot --port {{ .Values.service.port }} --loglevel={{ .Values.loglevel }} --bw {{ .Values.PCIeBW }} --w {{ .Values.repeat }} --invasive-check-timer {{ .Values.invasive }} --workers {{ .Values.workers }} imagePullPolicy: {{ .Values.image.pullPolicy }} name: autopilot securityContext: diff --git a/helm-charts/autopilot/values.yaml b/helm-charts/autopilot/values.yaml index 29e36f3..689aae9 100644 --- a/helm-charts/autopilot/values.yaml +++ b/helm-charts/autopilot/values.yaml @@ -5,6 +5,9 @@ image: repository: quay.io/autopilot/autopilot pullPolicy: Always +# Workers for concurrent tasks. Defaults to 0 which uses 2*number_of_logical_CPU_cores (depends on the resource limits). +workers: 2 + # Bandwidth threshold below which PCIe links are considered defective (Gb/s) # It is recommended to set a threshold that is 25% or lower of the expected peak PCIe bandwidth capability, which maps to maximum peak from 16 lanes to 4 lanes. For example, for a PCIe Gen4x16, reported peak bandwidth is 63GB/s. A degradation at 25% is 15.75GB/s, which corresponds to PCIe Gen4x4. The measured bandwidth is expected to be at least 80% of the expected peak PCIe generation bandwidth. PCIeBW: 4 From 6fea7a20f8283429fc74507fe92597aaaf69b737 Mon Sep 17 00:00:00 2001 From: amirhnajafiz Date: Wed, 6 Aug 2025 12:09:50 -0400 Subject: [PATCH 6/7] Fix: pool size value issue Signed-off-by: amirhnajafiz --- autopilot-daemon/pkg/cmd/main.go | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/autopilot-daemon/pkg/cmd/main.go b/autopilot-daemon/pkg/cmd/main.go index bfc678d..864c482 100644 --- a/autopilot-daemon/pkg/cmd/main.go +++ b/autopilot-daemon/pkg/cmd/main.go @@ -25,7 +25,7 @@ func main() { v := flag.String("loglevel", "2", "Log level") repeat := flag.String("w", "24h", "Run all tests periodically on each node. Time set in interval format. Defaults to 24h") invasive := flag.String("invasive-check-timer", "4h", "Run invasive checks (e.g., dcgmi level 3) on each node when GPUs are free. Time set in interval format. Defaults to 4h. Set to 0 to avoid invasive checks") - workersLimit := flag.Int("workers", 0, "Number of workers to use for concurrent tasks. Defaults to 0 which uses 2*number_of_logical_CPU_cores") + poolSizeInput := flag.Int("workers", 0, "Number of workers to use for concurrent health checks. Defaults to 0 which uses 2*number_of_logical_CPU_cores") flag.Parse() @@ -125,15 +125,16 @@ func main() { // Create a Watcher over nodes. Needed to export metrics from data created by external jobs (i.e., dcgm Jobs) go utils.WatchNode() - // Create a WorkerPool to handle tasks concurrently - numCPU := runtime.NumCPU() - if *workersLimit > 0 { + // Set the pool size based on the number of CPU cores + poolSize := runtime.NumCPU() * 2 // use 2 workers per CPU core + if *poolSizeInput > 0 { // if user has set a limit, use it - numCPU = *workersLimit + poolSize = *poolSizeInput } - workerPool := worker.CreateWorkerPool(2 * numCPU) // use 2 workers per CPU core - klog.Infof("Starting WorkerPool with %d workers", 2*numCPU) + // Create a WorkerPool to handle tasks concurrently + workerPool := worker.CreateWorkerPool(poolSize) + klog.Infof("Starting WorkerPool with %d workers", poolSize) // Run the health checks at startup, then start the timer workerPool.Submit(worker.TaskPeriodicCheck) From d72300fec79de7dc90899889345cc9d4bb7af7ae Mon Sep 17 00:00:00 2001 From: amirhnajafiz Date: Fri, 8 Aug 2025 14:16:53 -0400 Subject: [PATCH 7/7] Add: tracing logs in worker module Signed-off-by: amirhnajafiz --- autopilot-daemon/pkg/worker/enum.go | 12 ++++++++++++ autopilot-daemon/pkg/worker/pool.go | 5 +++++ autopilot-daemon/pkg/worker/worker.go | 3 +++ 3 files changed, 20 insertions(+) diff --git a/autopilot-daemon/pkg/worker/enum.go b/autopilot-daemon/pkg/worker/enum.go index b2ec229..c01ec13 100644 --- a/autopilot-daemon/pkg/worker/enum.go +++ b/autopilot-daemon/pkg/worker/enum.go @@ -8,3 +8,15 @@ const ( TaskPeriodicCheck TaskType = iota + 1 TaskInvasiveCheck ) + +// String returns a string representation of the TaskType. +func (t TaskType) String() string { + switch t { + case TaskPeriodicCheck: + return "Periodic Check" + case TaskInvasiveCheck: + return "Invasive Check" + default: + return "Unknown Task Type" + } +} diff --git a/autopilot-daemon/pkg/worker/pool.go b/autopilot-daemon/pkg/worker/pool.go index df6c78d..5c129e8 100644 --- a/autopilot-daemon/pkg/worker/pool.go +++ b/autopilot-daemon/pkg/worker/pool.go @@ -2,6 +2,8 @@ package worker import ( "sync" + + "k8s.io/klog/v2" ) // WorkerPool manages a pool of go-routines that process tasks concurrently. @@ -32,6 +34,7 @@ func CreateWorkerPool(numberOfWorkers int) *WorkerPool { func (wp *WorkerPool) Submit(task TaskType) { // check if the task is running if _, exists := wp.runningTasks.Load(task); exists { + klog.InfoS("Task already running, skipping submission", "task", task.String()) return // task is already running, do not submit again } @@ -39,4 +42,6 @@ func (wp *WorkerPool) Submit(task TaskType) { // the worker will remove it from runningTasks when done wp.runningTasks.Store(task, struct{}{}) wp.taskQueue <- task + + klog.InfoS("Task submitted to worker pool", "task", task.String()) } diff --git a/autopilot-daemon/pkg/worker/worker.go b/autopilot-daemon/pkg/worker/worker.go index ff07532..9017511 100644 --- a/autopilot-daemon/pkg/worker/worker.go +++ b/autopilot-daemon/pkg/worker/worker.go @@ -13,6 +13,8 @@ import ( // it removes the task from the runningTasks map once completed. func worker(c chan TaskType, sm *sync.Map) { for task := range c { + klog.InfoS("Processing task", "task", task.String()) + switch task { case TaskPeriodicCheck: healthcheck.PeriodicCheck() @@ -24,5 +26,6 @@ func worker(c chan TaskType, sm *sync.Map) { // mark the task as completed sm.Delete(task) + klog.InfoS("Task completed", "task", task.String()) } }