Nail/fission-src
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add funcSvcGroup type in pool cache for grouping of function services (#2728)

Browse Source 
* add specfic structs for function svc and group

* remove unused var in funcSvcGroup

* format poolcache
This commit is contained in:
Pranoy Kumar Kundu
2023-02-28 09:06:16 +05:30
committed by GitHub
parent 1cb18a78a6
commit 0de8923ea8
1 changed files with 37 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/executor/fscache/poolcache.go
+37 -28
View File
@@ -39,17 +39,21 @@ const (
) )
type ( type (
// value used as "value" in cache funcSvcInfo struct {
value struct {
val *FuncSvc val *FuncSvc
activeRequests int // number of requests served by function pod activeRequests int // number of requests served by function pod
currentCPUUsage resource.Quantity // current cpu usage of the specialized function pod currentCPUUsage resource.Quantity // current cpu usage of the specialized function pod
cpuLimit resource.Quantity // if currentCPUUsage is more than cpuLimit cache miss occurs in getValue request cpuLimit resource.Quantity // if currentCPUUsage is more than cpuLimit cache miss occurs in getValue request
} }
funcSvcGroup struct {
svcs map[string]*funcSvcInfo
}
// PoolCache implements a simple cache implementation having values mapped by two keys [function][address]. // PoolCache implements a simple cache implementation having values mapped by two keys [function][address].
// As of now PoolCache is only used by poolmanager executor // As of now PoolCache is only used by poolmanager executor
PoolCache struct { PoolCache struct {
cache map[string]map[string]*value cache map[string]*funcSvcGroup
requestChannel chan *request requestChannel chan *request
logger *zap.Logger logger *zap.Logger
} }
@@ -76,7 +80,7 @@ type (
func NewPoolCache(logger *zap.Logger) *PoolCache { func NewPoolCache(logger *zap.Logger) *PoolCache {
c := &PoolCache{ c := &PoolCache{
cache: make(map[string]map[string]*value), cache: make(map[string]*funcSvcGroup),
requestChannel: make(chan *request), requestChannel: make(chan *request),
logger: logger, logger: logger,
} }
@@ -90,20 +94,21 @@ func (c *PoolCache) service() {
resp := &response{} resp := &response{}
switch req.requestType { switch req.requestType {
case getValue: case getValue:
values, ok := c.cache[req.function] funcSvcGroup, ok := c.cache[req.function]
found := false found := false
if !ok { if !ok {
resp.error = ferror.MakeError(ferror.ErrorNotFound, resp.error = ferror.MakeError(ferror.ErrorNotFound,
fmt.Sprintf("function Name '%v' not found", req.function)) fmt.Sprintf("function Name '%v' not found", req.function))
} else { } else {
for addr := range values { for addr := range funcSvcGroup.svcs {
if values[addr].activeRequests < req.requestsPerPod && values[addr].currentCPUUsage.Cmp(values[addr].cpuLimit) < 1 { if funcSvcGroup.svcs[addr].activeRequests < req.requestsPerPod &&
funcSvcGroup.svcs[addr].currentCPUUsage.Cmp(funcSvcGroup.svcs[addr].cpuLimit) < 1 {
// mark active // mark active
values[addr].activeRequests++ funcSvcGroup.svcs[addr].activeRequests++
if c.logger.Core().Enabled(zap.DebugLevel) { 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", values[addr].activeRequests)) 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 = values[addr].val resp.value = funcSvcGroup.svcs[addr].val
found = true found = true
break break
} }
@@ -111,26 +116,28 @@ func (c *PoolCache) service() {
if !found { if !found {
resp.error = ferror.MakeError(ferror.ErrorNotFound, fmt.Sprintf("function '%v' all functions are busy", req.function)) resp.error = ferror.MakeError(ferror.ErrorNotFound, fmt.Sprintf("function '%v' all functions are busy", req.function))
} }
resp.totalActive = len(values) resp.totalActive = len(funcSvcGroup.svcs)
} }
req.responseChannel <- resp req.responseChannel <- resp
case setValue: case setValue:
if _, ok := c.cache[req.function]; !ok { if _, ok := c.cache[req.function]; !ok {
c.cache[req.function] = make(map[string]*value) c.cache[req.function] = &funcSvcGroup{
svcs: make(map[string]*funcSvcInfo),
}
} }
if _, ok := c.cache[req.function][req.address]; !ok { if _, ok := c.cache[req.function].svcs[req.address]; !ok {
c.cache[req.function][req.address] = &value{} c.cache[req.function].svcs[req.address] = &funcSvcInfo{}
} }
c.cache[req.function][req.address].val = req.value c.cache[req.function].svcs[req.address].val = req.value
c.cache[req.function][req.address].activeRequests++ c.cache[req.function].svcs[req.address].activeRequests++
if c.logger.Core().Enabled(zap.DebugLevel) { 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][req.address].activeRequests)) 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))
} }
c.cache[req.function][req.address].cpuLimit = req.cpuUsage c.cache[req.function].svcs[req.address].cpuLimit = req.cpuUsage
case listAvailableValue: case listAvailableValue:
vals := make([]*FuncSvc, 0) vals := make([]*FuncSvc, 0)
for key1, values := range c.cache { for key1, values := range c.cache {
for key2, value := range values { for key2, value := range values.svcs {
debugLevel := c.logger.Core().Enabled(zap.DebugLevel) debugLevel := c.logger.Core().Enabled(zap.DebugLevel)
if debugLevel { if debugLevel {
otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Reading active requests", zap.String("function", key1), zap.String("address", key2), zap.Int("activeRequests", value.activeRequests)) otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Reading active requests", zap.String("function", key1), zap.String("address", key2), zap.Int("activeRequests", value.activeRequests))
@@ -147,26 +154,28 @@ func (c *PoolCache) service() {
req.responseChannel <- resp req.responseChannel <- resp
case setCPUUtilization: case setCPUUtilization:
if _, ok := c.cache[req.function]; !ok { if _, ok := c.cache[req.function]; !ok {
c.cache[req.function] = make(map[string]*value) c.cache[req.function] = &funcSvcGroup{
svcs: make(map[string]*funcSvcInfo),
}
} }
if _, ok := c.cache[req.function][req.address]; ok { if _, ok := c.cache[req.function].svcs[req.address]; ok {
c.cache[req.function][req.address].currentCPUUsage = req.cpuUsage c.cache[req.function].svcs[req.address].currentCPUUsage = req.cpuUsage
} }
case markAvailable: case markAvailable:
if _, ok := c.cache[req.function]; ok { if _, ok := c.cache[req.function]; ok {
if _, ok = c.cache[req.function][req.address]; ok { if _, ok = c.cache[req.function].svcs[req.address]; ok {
if c.cache[req.function][req.address].activeRequests > 0 { if c.cache[req.function].svcs[req.address].activeRequests > 0 {
c.cache[req.function][req.address].activeRequests-- c.cache[req.function].svcs[req.address].activeRequests--
if c.logger.Core().Enabled(zap.DebugLevel) { if c.logger.Core().Enabled(zap.DebugLevel) {
otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Decrease active requests", zap.String("function", req.function), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function][req.address].activeRequests)) otelUtils.LoggerWithTraceID(req.ctx, c.logger).Debug("Decrease active requests", zap.String("function", req.function), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests))
} }
} else { } else {
otelUtils.LoggerWithTraceID(req.ctx, c.logger).Error("Invalid request to decrease active requests", zap.String("function", req.function), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function][req.address].activeRequests)) otelUtils.LoggerWithTraceID(req.ctx, c.logger).Error("Invalid request to decrease active requests", zap.String("function", req.function), zap.String("address", req.address), zap.Int("activeRequests", c.cache[req.function].svcs[req.address].activeRequests))
} }
} }
} }
case deleteValue: case deleteValue:
delete(c.cache[req.function], req.address) delete(c.cache[req.function].svcs, req.address)
req.responseChannel <- resp req.responseChannel <- resp
default: default:
resp.error = ferror.MakeError(ferror.ErrorInvalidArgument, resp.error = ferror.MakeError(ferror.ErrorInvalidArgument,