Introduce pool level mutex for thread safety and corresponding tests (#14)

* introduced pool level mutex for thread safety and correspoding tests
* corrected feature name typo
* added a new integration test for concurrent cpu moves

* improved thread safety with cpu.consolidate()

* Address race-condition issue in new Checkmarx scan

This PR will make use of channels to pass go routines results back to doConcurrentMoveCPUSetProfile

* Bumping go to 1.21 to keep library in sync with manager

---------

Co-authored-by: Lukasz Danilczuk <lukasz.danilczuk@intel.com>

Author: madalazar

go.mod

@@ -1,6 +1,6 @@
 module github.com/intel/power-optimization-library
 
-go 1.20
+go 1.21
 
 require (
 	github.com/go-logr/logr v1.2.4

pkg/power/c_states.go

@@ -43,7 +43,7 @@ func initCStates() featureStatus {
 	driver = strings.TrimSuffix(driver, "\n")
 	feature.driver = driver
 	if err != nil {
-		feature.err = fmt.Errorf("failed to determine driver")
+		feature.err = fmt.Errorf("failed to determine driver: %w", err)
 		return feature
 	}
 	if !isSupportedCStatesDriver(driver) {

pkg/power/cpu.go

@@ -16,6 +16,7 @@ type Cpu interface {
 	getPool() Pool
 	doSetPool(pool Pool) error
 	consolidate() error
+	consolidate_unsafe() error
 
 	// C-States stuff
 	SetCStates(cStates CStates) error
@@ -26,7 +27,7 @@ type Cpu interface {
 
 type cpuImpl struct {
 	id    uint
-	mutex *sync.Mutex
+	mutex sync.Locker
 	pool  Pool
 	core  *cpuCore
 	// C-States properties
@@ -44,6 +45,11 @@ func newCpu(coreID uint, core *cpuCore) (Cpu, error) {
 }
 
 func (cpu *cpuImpl) consolidate() error {
+	cpu.mutex.Lock()
+	defer cpu.mutex.Unlock()
+	return cpu.consolidate_unsafe()
+}
+func (cpu *cpuImpl) consolidate_unsafe() error {
 	if err := cpu.updateFrequencies(); err != nil {
 		return err
 	}
@@ -53,6 +59,9 @@ func (cpu *cpuImpl) consolidate() error {
 	return nil
 }
 
+
+
+
 // SetPool moves current core to a specified target pool
 // allowed movements are reservedPoolType <-> sharedPoolType and sharedPoolType <-> any exclusive pool
 func (cpu *cpuImpl) SetPool(targetPool Pool) error {
@@ -77,6 +86,9 @@ func (cpu *cpuImpl) SetPool(targetPool Pool) error {
 	}
 
 	log.Info("Set pool", "cpu", cpu.id, "source pool", cpu.pool.Name(), "target pool", targetPool.Name())
+	cpu.mutex.Lock()
+	defer cpu.mutex.Unlock()
+
 	if cpu.pool == targetPool { // case 0,1,5
 		return nil
 	}
@@ -102,17 +114,19 @@ func (cpu *cpuImpl) SetPool(targetPool Pool) error {
 }
 
 func (cpu *cpuImpl) doSetPool(pool Pool) error {
-	log.V(4).Info("mutex locking cpu", "coreID", cpu.id)
-	cpu.mutex.Lock()
-
-	defer func() {
-		log.V(4).Info("mutex unlocking cpu", "coreID", cpu.id)
-		cpu.mutex.Unlock()
-	}()
+	cpu.pool.poolMutex().Lock()
+	pool.poolMutex().Lock()
+	log.V(4).Info("acquired mutexes", "source", cpu.pool.Name(), "target", pool.Name(), "cpu")
 
 	origPool := cpu.pool
 	cpu.pool = pool
 
+	defer func() {
+		log.V(4).Info("releasing mutexes", "source", origPool.Name(), "target", pool.Name())
+		origPool.poolMutex().Unlock()
+		pool.poolMutex().Unlock()
+	}()
+
 	origPoolCpus := origPool.Cpus()
 	log.V(4).Info("removing cpu from pool", "pool", origPool.Name(), "coreID", cpu.id)
 	if err := origPoolCpus.remove(cpu); err != nil {
@@ -121,7 +135,7 @@ func (cpu *cpuImpl) doSetPool(pool Pool) error {
 	}
 
 	log.V(4).Info("starting consolidation of cpu", "coreID", cpu.id)
-	if err := cpu.consolidate(); err != nil {
+	if err := cpu.consolidate_unsafe(); err != nil {
 		cpu.pool = origPool
 		origPoolCpus.add(cpu)
 		return err