Started working on transprocessor and implementation of adaptable process for future upgrades

Author: PythonHacker24

internal/scheduler/fcfs/fcfs.go

@@ -1,14 +1,18 @@
 package fcfs
 
 import (
+	"context"
 	"runtime"
 
+	"go.uber.org/zap"
+
 	"github.com/PythonHacker24/linux-acl-management-backend/config"
 	"github.com/PythonHacker24/linux-acl-management-backend/internal/session"
+	"github.com/PythonHacker24/linux-acl-management-backend/internal/transprocessor"
 )
 
 /* spawns a new FCFS scheduler */
-func NewFCFSScheduler(sm *session.Manager) *FCFSScheduler {
+func NewFCFSScheduler(sm *session.Manager, processor transprocessor.TransactionProcessor) *FCFSScheduler {
 	/* calculate max workers */
 	maxProcs := runtime.GOMAXPROCS(0)
 	maxWorkers := config.BackendConfig.AppInfo.MaxWorkers 
@@ -30,5 +34,61 @@ func NewFCFSScheduler(sm *session.Manager) *FCFSScheduler {
 		curSessionManager: sm,
 		maxWorkers: maxWorkers,
 		semaphore: make(chan struct{}, maxWorkers),
+		processor: processor,
 	}
 }
+
+/* run the fcfs scheduler with context */
+func (f *FCFSScheduler) Run(ctx context.Context) error {
+	for {
+		select {
+
+		/* check if ctx is done - catchable if default is not working hard (ideal scheduler) */
+		case <-ctx.Done():
+			return nil
+
+		/* in case default is working hard - ctx is passed here so it must attempt to quit */
+		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 */
+				continue
+			}
+			
+			/* check if transaction queue of the session is empty */
+			curSession.Mutex.Lock()
+			if curSession.TransactionQueue.Len() == 0 {
+				curSession.Mutex.Unlock()
+				continue
+			}
+
+			/* get a transaction from the session to process */
+			transaction := curSession.TransactionQueue.Remove(curSession.TransactionQueue.Front())
+			curSession.Mutex.Unlock()
+
+			/* block if all workers are busy */
+			f.semaphore <- struct{}{}
+
+			/* go routine is available to be spawned */
+			go func(curSession *session.Session, transaction interface{}) {
+				/* defer clearing the semaphore channel */
+				defer func() { <-f.semaphore }()
+
+				/* 
+					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.processor.Process(ctx, curSession, transaction); err != nil {
+					zap.L().Error("Faild to process transaction", 
+						zap.Error(err),
+					)
+				}
+			}(curSession, transaction)
+		}
+	}	
+}

internal/scheduler/fcfs/model.go

@@ -1,10 +1,8 @@
 package fcfs
 
 import (
-	"context"
-
 	"github.com/PythonHacker24/linux-acl-management-backend/internal/session"
-	"go.uber.org/zap"
+	"github.com/PythonHacker24/linux-acl-management-backend/internal/transprocessor"
 )
 
 /*
@@ -20,59 +18,5 @@ type FCFSScheduler struct {
 
 	/* for limiting spawning of goroutines */
 	semaphore chan struct{}
-}
-
-/* run the fcfs scheduler with context */
-func (f *FCFSScheduler) Run(ctx context.Context) error {
-	for {
-		select {
-
-		/* check if ctx is done - catchable if default is not working hard (ideal scheduler) */
-		case <-ctx.Done():
-			return nil
-
-		/* in case default is working hard - ctx is passed here so it must attempt to quit */
-		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 */
-				continue
-			}
-			
-			/* check if transaction queue of the session is empty */
-			curSession.Mutex.Lock()
-			if curSession.TransactionQueue.Len() == 0 {
-				curSession.Mutex.Unlock()
-				continue
-			}
-
-			/* get a transaction from the session to process */
-			transaction := curSession.TransactionQueue.Remove(curSession.TransactionQueue.Front())
-			curSession.Mutex.Unlock()
-
-			/* block if all workers are busy */
-			f.semaphore <- struct{}{}
-
-			/* go routine is available to be spawned */
-			go func(curSession *session.Session, transaction interface{}) {
-				/* defer clearing the semaphore channel */
-				defer func() { <-f.semaphore }()
-
-				/* 
-					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),
-					)
-				}
-			}(curSession, transaction)
-		}
-	}	
+	processor transprocessor.TransactionProcessor
 }

internal/scheduler/handler.go

@@ -1,3 +1,5 @@
 package scheduler
 
 /* contains handlers related to monitoring the schedular */
+
+/* TODO: Implementing a watchdog */

internal/session/handler.go

@@ -1 +1,3 @@
 package session
+
+/* TODO: watchdog for session */

internal/transprocessor/model.go

@@ -0,0 +1,14 @@
+package transprocessor
+
+/*
+	transprocessor implements the transactions structure that whole project complies with
+*/
+
+/* universal transaction structure */
+type Transaction struct {
+	ID      string
+	/* command structure would be added later */
+}
+
+/* permissions processor */
+type PermProcessor struct{}

internal/transprocessor/perm-processor.go

@@ -0,0 +1,43 @@
+package transprocessor
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/PythonHacker24/linux-acl-management-backend/internal/session"
+	"go.uber.org/zap"
+)
+
+/* instanciate new permission processor */
+func NewPermProcessor() *PermProcessor {
+	return &PermProcessor{}	
+}
+
+/* processor for permissions manager */
+func (p *PermProcessor) Process(ctx context.Context, curSession *session.Session, tx interface{}) error {
+	transaction, ok := tx.(*Transaction) 
+	if !ok {
+		return fmt.Errorf("invalid transaction type")
+	}
+
+	/* add complete information here + persistent logging in database */
+	zap.L().Info("Processing Transaction",
+		zap.String("user", curSession.Username),
+	)
+	
+	select {
+		case <-ctx.Done():
+			/* 
+				store this into persistent storage too!
+				make sure database connections are closed after scheduler shutsdown
+			*/
+			zap.L().Warn("Transaction process stopped due to shutdown",
+				zap.String("user", curSession.Username),	
+			)
+			return ctx.Err()
+		default:
+			_ = transaction		
+	}
+
+	return nil	
+}

internal/transprocessor/processor.go

@@ -0,0 +1,20 @@
+package transprocessor
+
+import (
+	"context"
+
+	"github.com/PythonHacker24/linux-acl-management-backend/internal/session"
+)
+
+/*
+	the job of scheduler was to handle how transactions are allocated to executors
+	transprocessor's job is to open the content of transactions and take care of them onwards
+	so work of transactions was irrelevant to scheduler - transprocessor is responsible
+
+	also, this archirecture allows us to create mulitple processors in case we plan to extend in future
+*/
+
+/* transaction processor - pluggable to any scheduler */
+type TransactionProcessor interface {
+	Process(ctx context.Context, curSession *session.Session, transaction interface{}) error
+}