Updated FCFS scheduler and Goroutine spawning

Author: PythonHacker24

cmd/laclm/main.go

@@ -14,6 +14,7 @@ import (
 	"github.com/joho/godotenv"
 	"github.com/spf13/cobra"
 	"go.uber.org/zap"
+	"go.uber.org/automaxprocs/maxprocs"
 
 	"github.com/PythonHacker24/linux-acl-management-backend/api/routes"
 	"github.com/PythonHacker24/linux-acl-management-backend/config"
@@ -90,6 +91,13 @@ func exec() error {
 	/* zap.L() can be used all over the code for global level logging */
 	zap.L().Info("Logger Initiated ...")
 
+	/* calculate max procs accurately (runtime.GOMAXPROCS(0)) */
+	if _, err := maxprocs.Set(); err != nil {
+		zap.L().Error("automaxproces: failed to set GOMAXPROCS",
+			zap.Error(err),
+		)
+	}
+
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 

config/app.go

@@ -13,6 +13,7 @@ type App struct {
 	DebugMode      bool   `yaml:"debug_mode,omitempty"`
 	SessionTimeout int    `yaml:"session_timeout,omitempty"`
 	BasePath       string `yaml:"base_path,omitempty"`
+	MaxWorkers	   int 	  `yaml:"max_workers,omitempty"`
 }
 
 /* normalization function */
@@ -47,5 +48,7 @@ func (a *App) Normalize() error {
 		`))
 	}
 
+	/* max_workers can be zero - it will be adjusted scheduler */
+
 	return nil
 }

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/joho/godotenv v1.5.1 // indirect
 	github.com/spf13/cobra v1.9.1 // indirect
 	github.com/spf13/pflag v1.0.6 // indirect
+	go.uber.org/automaxprocs v1.6.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/crypto v0.37.0 // indirect

go.sum

@@ -24,6 +24,8 @@ github.com/spf13/cobra v1.9.1 h1:CXSaggrXdbHK9CF+8ywj8Amf7PBRmPCOJugH954Nnlo=
 github.com/spf13/cobra v1.9.1/go.mod h1:nDyEzZ8ogv936Cinf6g1RU9MRY64Ir93oCnqb9wxYW0=
 github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o=
 github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+go.uber.org/automaxprocs v1.6.0 h1:O3y2/QNTOdbF+e/dpXNNW7Rx2hZ4sTIPyybbxyNqTUs=
+go.uber.org/automaxprocs v1.6.0/go.mod h1:ifeIMSnPZuznNm6jmdzmU3/bfk01Fe2fotchwEFJ8r8=
 go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
 go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
 go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=

internal/scheduler/fcfs/fcfs.go

@@ -3,16 +3,31 @@ package fcfs
 import (
 	"runtime"
 
+	"github.com/PythonHacker24/linux-acl-management-backend/config"
 	"github.com/PythonHacker24/linux-acl-management-backend/internal/session"
 )
 
 /* spawns a new FCFS scheduler */
 func NewFCFSScheduler(sm *session.Manager) *FCFSScheduler {
 	/* calculate max workers */
-	/* TODO: make it configurable and dynamic */
-	maxWorkers := runtime.NumCPU()
+	maxProcs := runtime.GOMAXPROCS(0)
+	maxWorkers := config.BackendConfig.AppInfo.MaxWorkers 
+
+	/* 
+		incase of maxWorkers set less than or equal to 0, 
+		use 75% of GOMAXPROCS to prevent starvation to other processes 
+	*/
+	if maxWorkers <= 0 {
+		maxWorkers = int(float64(maxProcs) * 0.75)
+	}
+
+	/* Prevent over-allocation */
+	if maxWorkers > maxProcs {
+		maxWorkers = maxProcs
+	}
+
 	return &FCFSScheduler{
-		sessionManager: sm,
+		curSessionManager: sm,
 		maxWorkers: maxWorkers,
 		semaphore: make(chan struct{}, maxWorkers),
 	}

internal/scheduler/fcfs/model.go

@@ -7,6 +7,12 @@ import (
 	"go.uber.org/zap"
 )
 
+/*
+	Notes: the structure of scheduler is very modular
+	Docs must be updated for replacing a certain scheduler module with another
+	This includes installation of prebuilt module or developing a module
+*/
+
 /* FCFS Scheduler attached with curSession.Manager */
 type FCFSScheduler struct {
 	curSessionManager *session.Manager 
@@ -29,6 +35,7 @@ func (f *FCFSScheduler) Run(ctx context.Context) error {
 		default:
 			/* RULE: ctx is propogates all over the coming functions */
 
+			/* get next session in the queue (round robin manner) */
 			curSession := f.curSessionManager.GetNextSession()
 			if curSession == nil {
 				/* might need a delay of 10 ms */
@@ -54,8 +61,13 @@ func (f *FCFSScheduler) Run(ctx context.Context) error {
 				/* defer clearing the semaphore channel */
 				defer func() { <-f.semaphore }()
 
-				/* process the transaction */
-				if err := f.processTransaction; err != nil {
+				/* 
+					process the transaction 
+					* processTransaction handles transaction processing completely
+					* now it is responsible now responsible to execute it
+					* role of scheduler in handling transactions ends here
+				*/
+				if err := f.processTransaction(ctx, curSession, transaction); err != nil {
 					zap.L().Error("Faild to process transaction", 
 						zap.Error(err),
 					)