Covert maps to sync maps in executor fscache to avoid dirty reads (#2105)

Signed-off-by: Sanket Sudake <sanketsudake@gmail.com>

pkg/executor/executor.go

@@ -57,7 +57,7 @@ type (
 		fissionClient *crd.FissionClient
 
 		requestChan chan *createFuncServiceRequest
-		fsCreateWg  map[string]*sync.WaitGroup
+		fsCreateWg  sync.Map
 	}
 
 	createFuncServiceRequest struct {
@@ -82,7 +82,6 @@ func MakeExecutor(ctx context.Context, logger *zap.Logger, cms *cms.ConfigSecret
 		executorTypes: types,
 
 		requestChan: make(chan *createFuncServiceRequest),
-		fsCreateWg:  make(map[string]*sync.WaitGroup),
 	}
 
 	// Run all informers
@@ -138,13 +137,13 @@ func (executor *Executor) serveCreateFuncServices() {
 		}
 
 		// Cache miss -- is this first one to request the func?
-		wg, found := executor.fsCreateWg[crd.CacheKey(fnMetadata)]
+		wg, found := executor.fsCreateWg.Load(crd.CacheKey(fnMetadata))
 		if !found {
 			// create a waitgroup for other requests for
 			// the same function to wait on
 			wg := &sync.WaitGroup{}
 			wg.Add(1)
-			executor.fsCreateWg[crd.CacheKey(fnMetadata)] = wg
+			executor.fsCreateWg.Store(crd.CacheKey(fnMetadata), wg)
 
 			// launch a goroutine for each request, to parallelize
 			// the specialization of different functions
@@ -176,7 +175,7 @@ func (executor *Executor) serveCreateFuncServices() {
 					funcSvc: fsvc,
 					err:     err,
 				}
-				delete(executor.fsCreateWg, crd.CacheKey(fnMetadata))
+				executor.fsCreateWg.Delete(crd.CacheKey(fnMetadata))
 				wg.Done()
 			}()
 		} else {
@@ -184,6 +183,14 @@ func (executor *Executor) serveCreateFuncServices() {
 			go func() {
 				executor.logger.Debug("waiting for concurrent request for the same function",
 					zap.Any("function", fnMetadata))
+				wg, ok := wg.(*sync.WaitGroup)
+				if !ok {
+					err := fmt.Errorf("could not convert value to workgroup for function %v in namespace %v", fnMetadata.Name, fnMetadata.Namespace)
+					req.respChan <- &createFuncServiceResponse{
+						funcSvc: nil,
+						err:     err,
+					}
+				}
 				wg.Wait()
 
 				// get the function service from the cache

pkg/executor/executortype/poolmgr/gp.go

@@ -713,11 +713,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac
 		Atime:             time.Now(),
 	}
 
-	if gp.fsCache.PodToFsvc == nil {
-		gp.fsCache.PodToFsvc = make(map[string]*fscache.FuncSvc)
-	}
-	gp.fsCache.PodToFsvc[pod.GetObjectMeta().GetName()] = fsvc
-
+	gp.fsCache.PodToFsvc.Store(pod.GetObjectMeta().GetName(), fsvc)
 	gp.podFSVCMap.Store(pod.ObjectMeta.Name, []interface{}{crd.CacheKey(fsvc.Function), fsvc.Address})
 	gp.fsCache.AddFunc(*fsvc)
 

pkg/executor/executortype/poolmgr/gpm.go

@@ -644,7 +644,7 @@ func (gpm *GenericPoolManager) idleObjectReaper() {
 				continue
 			}
 
-			if _, ok := gpm.fsCache.WebsocketFsvc[fsvc.Name]; ok {
+			if _, ok := gpm.fsCache.WebsocketFsvc.Load(fsvc.Name); ok {
 				continue
 			}
 			// For function with the environment that no longer exists, executor
@@ -725,8 +725,13 @@ func (gpm *GenericPoolManager) WebsocketStartEventChecker(kubeClient *kubernetes
 				zap.String("Pod name", mObj.GetName()))
 
 			podName := strings.SplitAfter(mObj.GetName(), ".")
-			if fsvc, ok := gpm.fsCache.PodToFsvc[strings.TrimSuffix(podName[0], ".")]; ok {
-				gpm.fsCache.WebsocketFsvc[fsvc.Name] = true
+			if fsvc, ok := gpm.fsCache.PodToFsvc.Load(strings.TrimSuffix(podName[0], ".")); ok {
+				fsvc, ok := fsvc.(*fscache.FuncSvc)
+				if !ok {
+					gpm.logger.Error("could not covert item from PodToFsvc")
+					return
+				}
+				gpm.fsCache.WebsocketFsvc.Store(fsvc.Name, true)
 			}
 		},
 	})
@@ -761,8 +766,12 @@ func (gpm *GenericPoolManager) NoActiveConnectionEventChecker(kubeClient *kubern
 				zap.String("Pod name", mObj.GetName()))
 
 			podName := strings.SplitAfter(mObj.GetName(), ".")
-			if fsvc, ok := gpm.fsCache.PodToFsvc[strings.TrimSuffix(podName[0], ".")]; ok {
-
+			if fsvc, ok := gpm.fsCache.PodToFsvc.Load(strings.TrimSuffix(podName[0], ".")); ok {
+				fsvc, ok := fsvc.(*fscache.FuncSvc)
+				if !ok {
+					gpm.logger.Error("could not covert value from PodToFsvc")
+					return
+				}
 				gpm.fsCache.DeleteFunctionSvc(fsvc)
 				for i := range fsvc.KubernetesObjects {
 					gpm.logger.Info("release idle function resources due to  inactivity",

pkg/executor/fscache/functionServiceCache.go

@@ -18,6 +18,7 @@ package fscache
 
 import (
 	"fmt"
+	"sync"
 	"time"
 
 	"github.com/pkg/errors"
@@ -68,8 +69,8 @@ type (
 		byAddress         *cache.Cache     // address      -> function : map[string]metav1.ObjectMeta
 		byFunctionUID     *cache.Cache     // function uid -> function : map[string]metav1.ObjectMeta
 		connFunctionCache *poolcache.Cache // function-key -> funcSvc : map[string]*funcSvc
-		PodToFsvc         map[string]*FuncSvc
-		WebsocketFsvc     map[string]bool
+		PodToFsvc         sync.Map         // pod-name -> funcSvc: map[string]*FuncSvc
+		WebsocketFsvc     sync.Map         // funcSvc-name -> bool: map[string]bool
 		requestChannel    chan *fscRequest
 	}
 
@@ -111,8 +112,6 @@ func MakeFunctionServiceCache(logger *zap.Logger) *FunctionServiceCache {
 		byFunctionUID:     cache.MakeCache(0, 0),
 		connFunctionCache: poolcache.NewPoolCache(),
 		requestChannel:    make(chan *fscRequest),
-		PodToFsvc:         make(map[string]*FuncSvc),
-		WebsocketFsvc:     make(map[string]bool),
 	}
 	go fsc.service()
 	return fsc