From 715ef8267e380f621f64e0f556bc22469e537889 Mon Sep 17 00:00:00 2001 From: Pranoy Kumar Kundu Date: Thu, 30 Mar 2023 20:19:51 +0530 Subject: [PATCH] Improve poolmanager concurrency handling with virtual capacity (#2737) * add functionality to wait for specialization by keeping track of incoming requests * format executor package * fix required capacity to specialise new pod condition * move handling concurrency logic into pool cache from executor * remove unused methods and structs * implement queue in to store the svc wait * create a queue struct and its methods to handle concurrent inputs * use newly created queue to store waiting for svc requests * add waiting requests in queue and use them when a svc is ready * set function to request in queue if the context is still alive * remove concurrency approach to set svc for waiting requests * update the active requests whenever requests from pool are assigned a svc * add doc to define why the conditions exist * remove unwanted params in strcut and clean up code * set error while getting svc value if sum of specialization in progress and specialized is only more than concurrency limit * remove duplicate functions and unnecessary values in struct * close svc channel on set value and create constants for default concurrency and rpp * get next value in queue in case context is timed out for fetched value * remove specializationInProgress counter from pool cache * return in case the queue is empty wihle setting func to svc * test getSvcVaue and setSvcValue in poolcache * add unit tests for GetConcurrent and GetRequestsPerPod methods * reorder imports * add fuzzy testing for getSVCValue and setSVCValue in poolcache * restructure go mod file and update pool cache test cases * Add tests and bug fixes * refactor code and add test cases * add svcWaiting check while setting svc value --------- Signed-off-by: Sanket Sudake Co-authored-by: Sanket Sudake --- go.mod | 2 +- go.sum | 3 +- pkg/apis/core/v1/types.go | 19 ++ pkg/executor/api.go | 29 ++- pkg/executor/executor.go | 1 - .../executortype/container/containermgr.go | 6 - pkg/executor/executortype/executortype.go | 3 - .../executortype/newdeploy/newdeploymgr.go | 6 - pkg/executor/executortype/poolmgr/gp.go | 3 +- pkg/executor/executortype/poolmgr/gpm.go | 9 +- pkg/executor/fscache/functionServiceCache.go | 12 +- .../fscache/functionServiceCache_test.go | 10 +- pkg/executor/fscache/poolcache.go | 127 ++++++++++--- pkg/executor/fscache/poolcache_test.go | 168 +++++++++++++++--- pkg/executor/fscache/queue.go | 45 +++++ pkg/executor/fscache/queue_test.go | 115 ++++++++++++ 16 files changed, 454 insertions(+), 104 deletions(-) create mode 100644 pkg/executor/fscache/queue.go create mode 100644 pkg/executor/fscache/queue_test.go diff --git a/go.mod b/go.mod index 7c0ab10b..d1a4dc9a 100644 --- a/go.mod +++ b/go.mod @@ -155,7 +155,7 @@ require ( github.com/rogpeppe/go-internal v1.9.0 // indirect github.com/russross/blackfriday/v2 v2.1.0 // indirect github.com/sergi/go-diff v1.1.0 // indirect - github.com/sirupsen/logrus v1.8.1 // indirect + github.com/sirupsen/logrus v1.9.0 // indirect github.com/ulikunitz/xz v0.5.9 // indirect github.com/xanzy/ssh-agent v0.3.2 // indirect github.com/xi2/xz v0.0.0-20171230120015-48954b6210f8 // indirect diff --git a/go.sum b/go.sum index ae7674b8..233b1b94 100644 --- a/go.sum +++ b/go.sum @@ -631,8 +631,9 @@ github.com/sirupsen/logrus v1.4.1/go.mod h1:ni0Sbl8bgC9z8RoU9G6nDWqqs/fq4eDPysMB github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE= github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88= github.com/sirupsen/logrus v1.7.0/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0= -github.com/sirupsen/logrus v1.8.1 h1:dJKuHgqk1NNQlqoA6BTlM1Wf9DOH3NBjQyu0h9+AZZE= github.com/sirupsen/logrus v1.8.1/go.mod h1:yWOB1SBYBC5VeMP7gHvWumXLIWorT60ONWic61uBYv0= +github.com/sirupsen/logrus v1.9.0 h1:trlNQbNUG3OdDrDil03MCb1H2o9nJ1x4/5LYw7byDE0= +github.com/sirupsen/logrus v1.9.0/go.mod h1:naHLuLoDiP4jHNo9R0sCBMtWGeIprob74mVsIT4qYEQ= github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM= github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA= github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ= diff --git a/pkg/apis/core/v1/types.go b/pkg/apis/core/v1/types.go index f39706f6..a1a858de 100644 --- a/pkg/apis/core/v1/types.go +++ b/pkg/apis/core/v1/types.go @@ -22,6 +22,11 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) +const ( + DefaultConcurrency = 500 + DefaultRequestsPerPod = 1 +) + // // To add a Fission CRD type: // 1. Create a "spec" type, for everything in the type except metadata @@ -863,3 +868,17 @@ type ( func (a Archive) IsEmpty() bool { return len(a.Literal) == 0 && len(a.URL) == 0 } + +func (fn Function) GetConcurrency() int { + if fn.Spec.Concurrency == 0 { + return DefaultConcurrency + } + return fn.Spec.Concurrency +} + +func (fn Function) GetRequestPerPod() int { + if fn.Spec.RequestsPerPod == 0 { + return DefaultRequestsPerPod + } + return fn.Spec.RequestsPerPod +} diff --git a/pkg/executor/api.go b/pkg/executor/api.go index b4a7b032..8e8823ac 100644 --- a/pkg/executor/api.go +++ b/pkg/executor/api.go @@ -63,19 +63,10 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt zap.String("function_name", fn.ObjectMeta.Name), zap.String("function_namespace", fn.ObjectMeta.Namespace)) if t == fv1.ExecutorTypePoolmgr && !fn.Spec.OnceOnly { - concurrency := fn.Spec.Concurrency - if concurrency == 0 { - concurrency = 500 - } - requestsPerpod := fn.Spec.RequestsPerPod - if requestsPerpod == 0 { - requestsPerpod = 1 - } - fsvc, active, err := et.GetFuncSvcFromPoolCache(ctx, fn, requestsPerpod) + fsvc, err := et.GetFuncSvcFromCache(ctx, fn) // check if its a cache hit (check if there is already specialized function pod that can serve another request) if err == nil { // if a pod is already serving request then it already exists else validated - logger.Debug("from cache", zap.Int("active", active)) if et.IsValid(ctx, fsvc) { // Cached, return svc address logger.Debug("served from cache", zap.String("name", fsvc.Name), zap.String("address", fsvc.Address)) @@ -87,15 +78,19 @@ func (executor *Executor) getServiceForFunctionAPI(w http.ResponseWriter, r *htt zap.String("function_namespace", fn.ObjectMeta.Namespace), zap.String("address", fsvc.Address)) et.DeleteFuncSvcFromCache(ctx, fsvc) - active-- + } else { + code, msg := ferror.GetHTTPError(err) + if code == http.StatusNotFound { + logger.Debug("cache miss", zap.String("function_name", fn.ObjectMeta.Name)) + } else { + logger.Error("error getting service for function", + zap.Error(err), + zap.String("function_name", fn.ObjectMeta.Name)) + http.Error(w, msg, code) + return + } } - if active >= concurrency { - errMsg := fmt.Sprintf("max concurrency reached for %v. All %v instance are active", fn.ObjectMeta.Name, concurrency) - logger.Error("error occurred", zap.String("error", errMsg)) - http.Error(w, html.EscapeString(errMsg), http.StatusTooManyRequests) - return - } } else if t == fv1.ExecutorTypeNewdeploy || t == fv1.ExecutorTypeContainer { fsvc, err := et.GetFuncSvcFromCache(ctx, fn) if err == nil { diff --git a/pkg/executor/executor.go b/pkg/executor/executor.go index bcea0ea7..3e33fbe7 100644 --- a/pkg/executor/executor.go +++ b/pkg/executor/executor.go @@ -60,7 +60,6 @@ type ( requestChan chan *createFuncServiceRequest fsCreateWg sync.Map } - createFuncServiceRequest struct { context context.Context function *fv1.Function diff --git a/pkg/executor/executortype/container/containermgr.go b/pkg/executor/executortype/container/containermgr.go index 1b931da3..60e5a7c1 100644 --- a/pkg/executor/executortype/container/containermgr.go +++ b/pkg/executor/executortype/container/containermgr.go @@ -193,12 +193,6 @@ func (caaf *Container) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscache caaf.fsCache.DeleteEntry(fsvc) } -// GetFuncSvcFromPoolCache has not been implemented for Container Functions -func (caaf *Container) GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) { - // Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added. - return nil, 0, nil -} - // TapService makes a TouchByAddress request to the cache. func (caaf *Container) TapService(ctx context.Context, svcHost string) error { err := caaf.fsCache.TouchByAddress(svcHost) diff --git a/pkg/executor/executortype/executortype.go b/pkg/executor/executortype/executortype.go index 2e1d7eb8..03fe9d89 100644 --- a/pkg/executor/executortype/executortype.go +++ b/pkg/executor/executortype/executortype.go @@ -38,9 +38,6 @@ type ExecutorType interface { // GetFuncSvcFromCache retrieves function service from cache. GetFuncSvcFromCache(context.Context, *fv1.Function) (*fscache.FuncSvc, error) - // GetFuncSvcFromPoolCache retrieves function service and number of active instances after filtering on requestsPerPod and CPULimit - GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) - // DeleteFuncSvcFromCache deletes function service entry in cache. DeleteFuncSvcFromCache(context.Context, *fscache.FuncSvc) diff --git a/pkg/executor/executortype/newdeploy/newdeploymgr.go b/pkg/executor/executortype/newdeploy/newdeploymgr.go index efc841bd..da8e5c88 100644 --- a/pkg/executor/executortype/newdeploy/newdeploymgr.go +++ b/pkg/executor/executortype/newdeploy/newdeploymgr.go @@ -199,12 +199,6 @@ func (deploy *NewDeploy) UnTapService(ctx context.Context, key string, svcHost s // Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added. } -// GetFuncSvcFromPoolCache has not been implemented for NewDeployment -func (deploy *NewDeploy) GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) { - // Not Implemented for NewDeployment. Will be used when support of concurrent specialization of same function is added. - return nil, 0, nil -} - // TapService makes a TouchByAddress request to the cache. func (deploy *NewDeploy) TapService(ctx context.Context, svcHost string) error { otelUtils.SpanTrackEvent(ctx, "TapService") diff --git a/pkg/executor/executortype/poolmgr/gp.go b/pkg/executor/executortype/poolmgr/gp.go index 38acee36..bf4184c9 100644 --- a/pkg/executor/executortype/poolmgr/gp.go +++ b/pkg/executor/executortype/poolmgr/gp.go @@ -612,8 +612,7 @@ func (gp *GenericPool) getFuncSvc(ctx context.Context, fn *fv1.Function) (*fscac gp.fsCache.PodToFsvc.Store(pod.GetObjectMeta().GetName(), fsvc) gp.podFSVCMap.Store(pod.ObjectMeta.Name, []interface{}{crd.CacheKey(fsvc.Function), fsvc.Address}) - gp.fsCache.AddFunc(ctx, *fsvc) - + gp.fsCache.AddFunc(ctx, *fsvc, fn.GetRequestPerPod()) metrics.ColdStarts.WithLabelValues(fn.ObjectMeta.Name, fn.ObjectMeta.Namespace).Inc() logger.Info("added function service", diff --git a/pkg/executor/executortype/poolmgr/gpm.go b/pkg/executor/executortype/poolmgr/gpm.go index c543b599..9ea0b181 100644 --- a/pkg/executor/executortype/poolmgr/gpm.go +++ b/pkg/executor/executortype/poolmgr/gpm.go @@ -189,6 +189,7 @@ func (gpm *GenericPoolManager) GetTypeName(ctx context.Context) fv1.ExecutorType func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { otelUtils.SpanTrackEvent(ctx, "GetFuncSvc", otelUtils.GetAttributesForFunction(fn)...) logger := otelUtils.LoggerWithTraceID(ctx, gpm.logger) + // from Func -> get Env logger.Debug("getting environment for function", zap.String("function", fn.ObjectMeta.Name)) env, err := gpm.getFunctionEnv(ctx, fn) @@ -212,12 +213,8 @@ func (gpm *GenericPoolManager) GetFuncSvc(ctx context.Context, fn *fv1.Function) } func (gpm *GenericPoolManager) GetFuncSvcFromCache(ctx context.Context, fn *fv1.Function) (*fscache.FuncSvc, error) { - return nil, nil -} - -func (gpm *GenericPoolManager) GetFuncSvcFromPoolCache(ctx context.Context, fn *fv1.Function, requestsPerPod int) (*fscache.FuncSvc, int, error) { - otelUtils.SpanTrackEvent(ctx, "GetFuncSvcFromPoolCache", otelUtils.GetAttributesForFunction(fn)...) - return gpm.fsCache.GetFuncSvc(ctx, &fn.ObjectMeta, requestsPerPod) + otelUtils.SpanTrackEvent(ctx, "GetFuncSvcFromCache", otelUtils.GetAttributesForFunction(fn)...) + return gpm.fsCache.GetFuncSvc(ctx, &fn.ObjectMeta, fn.GetRequestPerPod(), fn.GetConcurrency()) } func (gpm *GenericPoolManager) DeleteFuncSvcFromCache(ctx context.Context, fsvc *fscache.FuncSvc) { diff --git a/pkg/executor/fscache/functionServiceCache.go b/pkg/executor/fscache/functionServiceCache.go index 34656179..fddac1d5 100644 --- a/pkg/executor/fscache/functionServiceCache.go +++ b/pkg/executor/fscache/functionServiceCache.go @@ -188,20 +188,20 @@ func (fsc *FunctionServiceCache) GetByFunction(m *metav1.ObjectMeta) (*FuncSvc, } // GetFuncSvc gets a function service from pool cache using function key and returns number of active instances of function pod -func (fsc *FunctionServiceCache) GetFuncSvc(ctx context.Context, m *metav1.ObjectMeta, requestsPerPod int) (*FuncSvc, int, error) { +func (fsc *FunctionServiceCache) GetFuncSvc(ctx context.Context, m *metav1.ObjectMeta, requestsPerPod int, concurrency int) (*FuncSvc, error) { key := crd.CacheKey(m) - fsvc, active, err := fsc.connFunctionCache.GetSvcValue(ctx, key, requestsPerPod) + fsvc, err := fsc.connFunctionCache.GetSvcValue(ctx, key, requestsPerPod, concurrency) if err != nil { fsc.logger.Info("Not found in Cache") - return nil, active, err + return nil, err } // update atime fsvc.Atime = time.Now() fsvcCopy := *fsvc - return &fsvcCopy, active, nil + return &fsvcCopy, nil } // GetByFunctionUID gets a function service from cache using function UUID. @@ -227,8 +227,8 @@ func (fsc *FunctionServiceCache) GetByFunctionUID(uid types.UID) (*FuncSvc, erro } // AddFunc adds a function service to pool cache. -func (fsc *FunctionServiceCache) AddFunc(ctx context.Context, fsvc FuncSvc) { - fsc.connFunctionCache.SetSvcValue(ctx, crd.CacheKey(fsvc.Function), fsvc.Address, &fsvc, fsvc.CPULimit) +func (fsc *FunctionServiceCache) AddFunc(ctx context.Context, fsvc FuncSvc, requestsPerPod int) { + fsc.connFunctionCache.SetSvcValue(ctx, crd.CacheKey(fsvc.Function), fsvc.Address, &fsvc, fsvc.CPULimit, requestsPerPod) now := time.Now() fsvc.Ctime = now fsvc.Atime = now diff --git a/pkg/executor/fscache/functionServiceCache_test.go b/pkg/executor/fscache/functionServiceCache_test.go index 4fb705dd..0f70a8d7 100644 --- a/pkg/executor/fscache/functionServiceCache_test.go +++ b/pkg/executor/fscache/functionServiceCache_test.go @@ -187,19 +187,17 @@ func TestFunctionServiceNewCache(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - fsc.AddFunc(ctx, *fsvc) - _, active, err := fsc.GetFuncSvc(ctx, fsvc.Function, 5) + fsc.AddFunc(ctx, *fsvc, 10) + concurrency := 10 + _, err = fsc.GetFuncSvc(ctx, fsvc.Function, 5, concurrency) if err != nil { logger.Panic("received error while retrieving value from cache") } - if active != 1 { - logger.Panic(fmt.Sprintln("active instances not matched expected 1, found ", active)) - } key := fmt.Sprintf("%v_%v", fn.ObjectMeta.UID, fn.ObjectMeta.ResourceVersion) fsc.MarkAvailable(key, fsvc.Address) - _, _, err = fsc.GetFuncSvc(ctx, fsvc.Function, 5) + _, err = fsc.GetFuncSvc(ctx, fsvc.Function, 5, concurrency) if err != nil { logger.Panic("received error while retrieving value from cache") } diff --git a/pkg/executor/fscache/poolcache.go b/pkg/executor/fscache/poolcache.go index 45073f7c..7e2d0930 100644 --- a/pkg/executor/fscache/poolcache.go +++ b/pkg/executor/fscache/poolcache.go @@ -47,7 +47,9 @@ type ( } funcSvcGroup struct { - svcs map[string]*funcSvcInfo + svcWaiting int + svcs map[string]*funcSvcInfo + queue *Queue } // PoolCache implements a simple cache implementation having values mapped by two keys [function][address]. @@ -67,12 +69,17 @@ type ( requestsPerPod int cpuUsage resource.Quantity responseChannel chan *response + concurrency int } response struct { error - allValues []*FuncSvc - value *FuncSvc - totalActive int + allValues []*FuncSvc + value *FuncSvc + svcWaitValue *svcWait + } + svcWait struct { + svcChannel chan *FuncSvc + ctx context.Context } ) @@ -88,6 +95,13 @@ func NewPoolCache(logger *zap.Logger) *PoolCache { return c } +func NewFuncSvcGroup() *funcSvcGroup { + return &funcSvcGroup{ + svcs: make(map[string]*funcSvcInfo), + queue: NewQueue(), + } +} + func (c *PoolCache) service() { for { req := <-c.requestChannel @@ -95,41 +109,87 @@ func (c *PoolCache) service() { switch req.requestType { case getValue: funcSvcGroup, ok := c.cache[req.function] - found := false if !ok { + c.cache[req.function] = NewFuncSvcGroup() + c.cache[req.function].svcWaiting++ resp.error = ferror.MakeError(ferror.ErrorNotFound, fmt.Sprintf("function Name '%v' not found", req.function)) - } else { - for addr := range funcSvcGroup.svcs { - if funcSvcGroup.svcs[addr].activeRequests < req.requestsPerPod && - funcSvcGroup.svcs[addr].currentCPUUsage.Cmp(funcSvcGroup.svcs[addr].cpuLimit) < 1 { - // mark active - funcSvcGroup.svcs[addr].activeRequests++ - if c.logger.Core().Enabled(zap.DebugLevel) { - otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with getValue", zap.String("function", req.function), zap.String("address", addr), zap.Int("activeRequests", funcSvcGroup.svcs[addr].activeRequests)) - } - resp.value = funcSvcGroup.svcs[addr].val - found = true - break + req.responseChannel <- resp + continue + } + found := false + totalActiveRequests := 0 + for addr := range funcSvcGroup.svcs { + totalActiveRequests += funcSvcGroup.svcs[addr].activeRequests + if funcSvcGroup.svcs[addr].activeRequests < req.requestsPerPod && + funcSvcGroup.svcs[addr].currentCPUUsage.Cmp(funcSvcGroup.svcs[addr].cpuLimit) < 1 { + // mark active + funcSvcGroup.svcs[addr].activeRequests++ + if c.logger.Core().Enabled(zap.DebugLevel) { + otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with getValue", zap.String("function", req.function), zap.String("address", addr), zap.Int("activeRequests", funcSvcGroup.svcs[addr].activeRequests)) } + resp.value = funcSvcGroup.svcs[addr].val + found = true + break } - if !found { - resp.error = ferror.MakeError(ferror.ErrorNotFound, fmt.Sprintf("function '%v' all functions are busy", req.function)) + } + if found { + req.responseChannel <- resp + continue + } + specializationInProgress := funcSvcGroup.svcWaiting - funcSvcGroup.queue.Len() + capacity := ((specializationInProgress + len(funcSvcGroup.svcs)) * req.requestsPerPod) - (totalActiveRequests + funcSvcGroup.svcWaiting) + if capacity > 0 { + funcSvcGroup.svcWaiting++ + svcWait := &svcWait{ + svcChannel: make(chan *FuncSvc), + ctx: req.ctx, } - resp.totalActive = len(funcSvcGroup.svcs) + resp.svcWaitValue = svcWait + funcSvcGroup.queue.Push(svcWait) + req.responseChannel <- resp + continue + } + + // concurrency should not be set to zero and + //sum of specialization in progress and specialized pods should be less then req.concurrency + if req.concurrency > 0 && (specializationInProgress+len(funcSvcGroup.svcs)) >= req.concurrency { + resp.error = ferror.MakeError(ferror.ErrorTooManyRequests, fmt.Sprintf("function '%s' concurrency '%d' limit reached.", req.function, req.concurrency)) + } else { + funcSvcGroup.svcWaiting++ + resp.error = ferror.MakeError(ferror.ErrorNotFound, fmt.Sprintf("function '%s' all functions are busy", req.function)) } req.responseChannel <- resp case setValue: if _, ok := c.cache[req.function]; !ok { - c.cache[req.function] = &funcSvcGroup{ - svcs: make(map[string]*funcSvcInfo), - } + c.cache[req.function] = NewFuncSvcGroup() } if _, ok := c.cache[req.function].svcs[req.address]; !ok { c.cache[req.function].svcs[req.address] = &funcSvcInfo{} } c.cache[req.function].svcs[req.address].val = req.value c.cache[req.function].svcs[req.address].activeRequests++ + if c.cache[req.function].svcWaiting > 0 { + c.cache[req.function].svcWaiting-- + svcCapacity := req.requestsPerPod - c.cache[req.function].svcs[req.address].activeRequests + queueLen := c.cache[req.function].queue.Len() + if svcCapacity > queueLen { + svcCapacity = queueLen + } + for i := 0; i <= svcCapacity; { + popped := c.cache[req.function].queue.Pop() + if popped == nil { + break + } + if popped.ctx.Err() == nil { + popped.svcChannel <- req.value + c.cache[req.function].svcs[req.address].activeRequests++ + i++ + } + close(popped.svcChannel) + c.cache[req.function].svcWaiting-- + } + } if c.logger.Core().Enabled(zap.DebugLevel) { otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Increase active requests with setValue", zap.String("function", req.function), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests)) } @@ -154,9 +214,7 @@ func (c *PoolCache) service() { req.responseChannel <- resp case setCPUUtilization: if _, ok := c.cache[req.function]; !ok { - c.cache[req.function] = &funcSvcGroup{ - svcs: make(map[string]*funcSvcInfo), - } + c.cache[req.function] = NewFuncSvcGroup() } if _, ok := c.cache[req.function].svcs[req.address]; ok { c.cache[req.function].svcs[req.address].currentCPUUsage = req.cpuUsage @@ -186,17 +244,27 @@ func (c *PoolCache) service() { } // GetValue returns a function service with status in Active else return error -func (c *PoolCache) GetSvcValue(ctx context.Context, function string, requestsPerPod int) (*FuncSvc, int, error) { +func (c *PoolCache) GetSvcValue(ctx context.Context, function string, requestsPerPod int, concurrency int) (*FuncSvc, error) { respChannel := make(chan *response) c.requestChannel <- &request{ ctx: ctx, requestType: getValue, function: function, + concurrency: concurrency, requestsPerPod: requestsPerPod, responseChannel: respChannel, } resp := <-respChannel - return resp.value, resp.totalActive, resp.error + + if resp.svcWaitValue != nil { + select { + case <-ctx.Done(): + return resp.value, ctx.Err() + case funcSvc := <-resp.svcWaitValue.svcChannel: + return funcSvc, nil + } + } + return resp.value, resp.error } // ListAvailableValue returns a list of the available function services stored in the Cache @@ -211,7 +279,7 @@ func (c *PoolCache) ListAvailableValue() []*FuncSvc { } // SetValue marks the value at key [function][address] as active(begin used) -func (c *PoolCache) SetSvcValue(ctx context.Context, function, address string, value *FuncSvc, cpuLimit resource.Quantity) { +func (c *PoolCache) SetSvcValue(ctx context.Context, function, address string, value *FuncSvc, cpuLimit resource.Quantity, requestsPerPod int) { respChannel := make(chan *response) c.requestChannel <- &request{ ctx: ctx, @@ -220,6 +288,7 @@ func (c *PoolCache) SetSvcValue(ctx context.Context, function, address string, v address: address, value: value, cpuUsage: cpuLimit, + requestsPerPod: requestsPerPod, responseChannel: respChannel, } } diff --git a/pkg/executor/fscache/poolcache_test.go b/pkg/executor/fscache/poolcache_test.go index f61e2adb..687916a0 100644 --- a/pkg/executor/fscache/poolcache_test.go +++ b/pkg/executor/fscache/poolcache_test.go @@ -2,11 +2,17 @@ package fscache import ( "context" + "fmt" "log" + "net/http" + "sync" + "sync/atomic" "testing" + "github.com/stretchr/testify/require" "k8s.io/apimachinery/pkg/api/resource" + ferror "github.com/fission/fission/pkg/error" "github.com/fission/fission/pkg/utils/loggerfactory" ) @@ -21,18 +27,44 @@ func TestPoolCache(t *testing.T) { defer cancel() logger := loggerfactory.GetLogger() c := NewPoolCache(logger) + concurrency := 5 + requestsPerPod := 2 + + // should return err since no svc is present + _, err := c.GetSvcValue(ctx, "func", requestsPerPod, concurrency) + if err == nil { + log.Panicf("found value when expected it to be nil") + } c.SetSvcValue(ctx, "func", "ip", &FuncSvc{ Name: "value", - }, resource.MustParse("45m")) + }, resource.MustParse("45m"), 10) + + // should not return any error since we added a svc + _, err = c.GetSvcValue(ctx, "func", requestsPerPod, concurrency) + checkErr(err) + + c.SetSvcValue(ctx, "func", "ip", &FuncSvc{ + Name: "value", + }, resource.MustParse("45m"), 10) + + // should return err since all functions are busy + _, err = c.GetSvcValue(ctx, "func", requestsPerPod, concurrency) + if err == nil { + log.Panicf("found value when expected it to be nil") + } + + c.SetSvcValue(ctx, "func", "ip", &FuncSvc{ + Name: "value", + }, resource.MustParse("45m"), 10) c.SetSvcValue(ctx, "func2", "ip2", &FuncSvc{ Name: "value2", - }, resource.MustParse("50m")) + }, resource.MustParse("50m"), 10) c.SetSvcValue(ctx, "func2", "ip22", &FuncSvc{ Name: "value22", - }, resource.MustParse("33m")) + }, resource.MustParse("33m"), 10) checkErr(c.DeleteValue(ctx, "func2", "ip2")) @@ -43,30 +75,126 @@ func TestPoolCache(t *testing.T) { c.MarkAvailable("func", "ip") - _, active, err := c.GetSvcValue(ctx, "func", 5) - if active != 1 { - log.Panicln("Expected 1 active, found", active) - } - checkErr(err) - checkErr(c.DeleteValue(ctx, "func", "ip")) - _, _, err = c.GetSvcValue(ctx, "func", 5) + _, err = c.GetSvcValue(ctx, "func", requestsPerPod, concurrency) if err == nil { log.Panicf("found deleted element") } c.SetSvcValue(ctx, "cpulimit", "100", &FuncSvc{ Name: "value", - }, resource.MustParse("3m")) + }, resource.MustParse("3m"), 10) c.SetCPUUtilization("cpulimit", "100", resource.MustParse("4m")) - - _, _, err = c.GetSvcValue(ctx, "cpulimit", 5) - - if err == nil { - log.Panicf("received pod address with higher CPU usage than limit") - } - c.SetCPUUtilization("cpulimit", "100", resource.MustParse("2m")) - _, _, err = c.GetSvcValue(ctx, "cpulimit", 5) - checkErr(err) +} + +func TestPoolCacheRequests(t *testing.T) { + + type structForTest struct { + name string + requests int + concurrency int + rpp int + simultaneous int + failedRequests int + } + + for _, tt := range []structForTest{ + { + name: "test1", + requests: 1, + concurrency: 1, + rpp: 1, + }, + { + name: "test2", + requests: 2, + concurrency: 2, + rpp: 1, + }, + { + name: "test3", + requests: 300, + concurrency: 5, + rpp: 60, + }, + + { + name: "test4", + requests: 6, + concurrency: 1, + rpp: 5, + failedRequests: 1, + }, + { + name: "test5", + requests: 6, + concurrency: 5, + rpp: 1, + failedRequests: 1, + }, + { + name: "test6", + requests: 300, + concurrency: 5, + rpp: 60, + simultaneous: 30, + }, + { + name: "test7", + requests: 310, + concurrency: 5, + rpp: 60, + simultaneous: 30, + failedRequests: 10, + }, + { + name: "test8", + requests: 10, + concurrency: 10, + rpp: 1, + simultaneous: 10, + }, + } { + t.Run(fmt.Sprintf("scenario-%s", tt.name), func(t *testing.T) { + var failedRequests, svcCounter uint64 + p := NewPoolCache(loggerfactory.GetLogger()) + wg := sync.WaitGroup{} + simultaneous := tt.simultaneous + if simultaneous == 0 { + simultaneous = 1 + } + for i := 1; i <= tt.requests; i++ { + wg.Add(1) + go func(reqno int) { + defer wg.Done() + svc, err := p.GetSvcValue(context.Background(), "func", tt.rpp, tt.concurrency) + if err != nil { + code, _ := ferror.GetHTTPError(err) + if code == http.StatusNotFound { + p.SetSvcValue(context.Background(), "func", fmt.Sprintf("svc-%d", svcCounter), &FuncSvc{ + Name: "value", + }, resource.MustParse("45m"), tt.rpp) + atomic.AddUint64(&svcCounter, 1) + } else { + t.Log(reqno, "=>", err) + atomic.AddUint64(&failedRequests, 1) + } + } else { + if svc == nil { + t.Log(reqno, "=>", "svc is nil") + atomic.AddUint64(&failedRequests, 1) + } + } + }(i) + if i%simultaneous == 0 { + wg.Wait() + } + } + wg.Wait() + + require.Equal(t, tt.failedRequests, int(atomic.LoadUint64(&failedRequests))) + require.Equal(t, tt.concurrency, int(atomic.LoadUint64(&svcCounter))) + }) + } } diff --git a/pkg/executor/fscache/queue.go b/pkg/executor/fscache/queue.go new file mode 100644 index 00000000..d27266bf --- /dev/null +++ b/pkg/executor/fscache/queue.go @@ -0,0 +1,45 @@ +package fscache + +import ( + "container/list" + "sync" +) + +type Queue struct { + items *list.List + mutex sync.Mutex +} + +func NewQueue() *Queue { + return &Queue{ + items: list.New(), + } +} + +func (q *Queue) Push(item *svcWait) { + q.mutex.Lock() + defer q.mutex.Unlock() + q.items.PushBack(item) +} + +func (q *Queue) Pop() *svcWait { + q.mutex.Lock() + defer q.mutex.Unlock() + + item := q.items.Front() + if item == nil { + return nil + } + q.items.Remove(item) + svcWait, ok := item.Value.(*svcWait) + if !ok { + return nil + } + return svcWait +} + +func (q *Queue) Len() int { + q.mutex.Lock() + defer q.mutex.Unlock() + return q.items.Len() +} diff --git a/pkg/executor/fscache/queue_test.go b/pkg/executor/fscache/queue_test.go new file mode 100644 index 00000000..55144aaa --- /dev/null +++ b/pkg/executor/fscache/queue_test.go @@ -0,0 +1,115 @@ +package fscache + +import ( + "sync" + "testing" +) + +func TestNewQueue(t *testing.T) { + q := NewQueue() + if q == nil { + t.Error("NewQueue returned nil") + } +} + +func TestQueuePushWithSingleRequest(t *testing.T) { + q := NewQueue() + item := &svcWait{ + svcChannel: make(chan *FuncSvc), + ctx: nil, + } + q.Push(item) + if q.Len() != 1 { + t.Errorf("Expected queue length to be 1, got %d", q.Len()) + } +} + +func TestQueuePopWithSingleRequest(t *testing.T) { + q := NewQueue() + item := &svcWait{ + svcChannel: make(chan *FuncSvc), + ctx: nil, + } + q.Push(item) + popped := q.Pop() + if popped == nil { + t.Error("Expected Pop to return a non-nil value") + } + if popped != item { + t.Error("Expected Pop to return the same element that was pushed") + } + if q.Len() != 0 { + t.Errorf("Expected queue length to be 0, got %d", q.Len()) + } +} + +func TestQueuePushWithConcurrentRequest(t *testing.T) { + q := NewQueue() + noOfRequests := 20 + var wg sync.WaitGroup + wg.Add(noOfRequests) + for i := 0; i < noOfRequests; i++ { + go func() { + defer wg.Done() + item := &svcWait{ + svcChannel: make(chan *FuncSvc), + ctx: nil, + } + q.Push(item) + }() + } + + wg.Wait() + + if q.Len() != noOfRequests { + t.Errorf("Expected queue length to be 20, got %d", q.Len()) + } +} + +func TestQueuePopWithConcurrentRequest(t *testing.T) { + q := NewQueue() + noOfPush := 20 + noOfPop := 15 + + var wg sync.WaitGroup + wg.Add(noOfPush + noOfPop) + + for i := 0; i < noOfPush; i++ { + go func() { + defer wg.Done() + item := &svcWait{ + svcChannel: make(chan *FuncSvc), + ctx: nil, + } + q.Push(item) + }() + } + + for i := 0; i < noOfPop; i++ { + go func() { + defer wg.Done() + q.Pop() + }() + } + + wg.Wait() + + if q.Len() != 5 { + t.Errorf("Expected queue length to be 5, got %d", q.Len()) + } +} + +func TestQueueLen(t *testing.T) { + q := NewQueue() + if q.Len() != 0 { + t.Errorf("Expected queue length to be 0, got %d", q.Len()) + } + item := &svcWait{ + svcChannel: make(chan *FuncSvc), + ctx: nil, + } + q.Push(item) + if q.Len() != 1 { + t.Errorf("Expected queue length to be 1, got %d", q.Len()) + } +}