diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index e25b2ae0..fb48c7d0 100644 --- a/pkg/executor/executor.go +++ b/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 diff --git a/pkg/executor/executortype/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go index d65dd923..a4627b89 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/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) diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index fe00b4e0..5c6eb696 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/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", diff --git a/pkg/executor/fscache/functionServiceCache.go b/pkg/executor/fscache/functionServiceCache.go index 578b4740..0bf416fa 100644 --- a/pkg/executor/fscache/functionServiceCache.go +++ b/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