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 <sanketsudake@gmail.com>
Co-authored-by: Sanket Sudake <sanketsudake@gmail.com>

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

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=

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
+}

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 {

pkg/executor/executor.go

@@ -60,7 +60,6 @@ type (
 		requestChan chan *createFuncServiceRequest
 		fsCreateWg  sync.Map
 	}
-
 	createFuncServiceRequest struct {
 		context  context.Context
 		function *fv1.Function

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)

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)
 

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")

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",

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) {

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

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")
 	}

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,
 	}
 }

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)))
+		})
+	}
 }

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()
+}

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())
+	}
+}