From ce4d3ac42d7291d60ca34dc71204989ab21b50a9 Mon Sep 17 00:00:00 2001 From: yang qf Date: Mon, 19 Jun 2017 20:56:38 -0500 Subject: [PATCH] Fix creation of redundant pods on heavy load cold start (#232) Modifies function specialization so that concurrent requests for the same function cause only one pod to be created. --- poolmgr/api.go | 101 ++++++++++++++++++++++++++++++----- poolmgr/gp.go | 5 +- router/functionServiceMap.go | 7 ++- 3 files changed, 96 insertions(+), 17 deletions(-) diff --git a/poolmgr/api.go b/poolmgr/api.go index 4408848f..854a5ceb 100644 --- a/poolmgr/api.go +++ b/poolmgr/api.go @@ -24,6 +24,7 @@ import ( "net/http" "os" "strings" + "sync" "time" "github.com/gorilla/handlers" @@ -44,22 +45,89 @@ type funcSvc struct { atime time.Time } +type createFuncServiceRequest struct { + funcMeta *fission.Metadata + respChan chan *createFuncServiceResponse +} +type createFuncServiceResponse struct { + address string + err error +} + type API struct { - poolMgr *GenericPoolManager - functionEnv *cache.Cache // map[fission.Metadata]fission.Environment - fsCache *functionServiceCache - controller *controllerclient.Client + poolMgr *GenericPoolManager + functionEnv *cache.Cache // map[fission.Metadata]fission.Environment + fsCache *functionServiceCache + controller *controllerclient.Client + fsCreateChannels map[fission.Metadata]*sync.WaitGroup + requestChan chan *createFuncServiceRequest //functionService *cache.Cache // map[fission.Metadata]*funcSvc //urlFuncSvc *cache.Cache // map[string]*funcSvc } func MakeAPI(gpm *GenericPoolManager, controller *controllerclient.Client, fsCache *functionServiceCache) *API { - return &API{ - poolMgr: gpm, - functionEnv: cache.MakeCache(time.Minute, 0), - fsCache: fsCache, - controller: controller, + api := API{ + poolMgr: gpm, + functionEnv: cache.MakeCache(time.Minute, 0), + fsCache: fsCache, + controller: controller, + fsCreateChannels: make(map[fission.Metadata]*sync.WaitGroup), + requestChan: make(chan *createFuncServiceRequest), + } + go api.serveCreateFuncServices() + return &api +} + +// All non-cached function service requests go through this goroutine +// serially. It parallelizes requests for different functions, and +// ensures that for a given function, only one request causes a pod to +// get specialized. In other words, it ensures that when there's an +// ongoing request for a certain function, all other requests wait for +// that request to complete. +func (api *API) serveCreateFuncServices() { + for { + req := <-api.requestChan + m := req.funcMeta + + // Cache miss -- is this first one to request the func? + wg, found := api.fsCreateChannels[*m] + if !found { + // create a waitgroup for other requests for + // the same function to wait on + wg := &sync.WaitGroup{} + wg.Add(1) + api.fsCreateChannels[*m] = wg + + // launch a goroutine for each request, to parallelize + // the specialization of different functions + go func() { + address, err := api.createServiceForFunction(m) + req.respChan <- &createFuncServiceResponse{ + address: address, + err: err, + } + delete(api.fsCreateChannels, *m) + wg.Done() + }() + } else { + // There's an existing request for this function, wait for it to finish + go func() { + log.Printf("Waiting for concurrent request for the same function: %v", m) + wg.Wait() + + // get the function service from the cache + fsvc, err := api.fsCache.GetByFunction(m) + address := "" + if err == nil { + address = fsvc.address + } + req.respChan <- &createFuncServiceResponse{ + address: address, + err: err, + } + }() + } } } @@ -135,6 +203,16 @@ func (api *API) getServiceForFunction(m *fission.Metadata) (string, error) { return fsvc.address, nil } + respChan := make(chan *createFuncServiceResponse) + api.requestChan <- &createFuncServiceRequest{ + funcMeta: m, + respChan: respChan, + } + resp := <-respChan + return resp.address, resp.err +} + +func (api *API) createServiceForFunction(m *fission.Metadata) (string, error) { // None exists, so create a new funcSvc: log.Printf("[%v] No cached function service found, creating one", m.Name) @@ -155,12 +233,11 @@ func (api *API) getServiceForFunction(m *fission.Metadata) (string, error) { // from GenericPool -> get one function container // (this also adds to the cache) log.Printf("[%v] getting function service from pool", m.Name) - funcSvc, err := pool.GetFuncSvc(m) + fsvc, err := pool.GetFuncSvc(m) if err != nil { return "", err } - - return funcSvc.address, nil + return fsvc.address, nil } // find funcSvc and update its atime diff --git a/poolmgr/gp.go b/poolmgr/gp.go index 7938af77..3036bc1c 100644 --- a/poolmgr/gp.go +++ b/poolmgr/gp.go @@ -438,7 +438,7 @@ func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) { if gp.useSvc { svcName := fmt.Sprintf("svc-%v", m.Name) if len(m.Uid) > 0 { - svcName += ("-" + m.Uid) + svcName += "-" + m.Uid } labels := gp.labelsForFunction(m) @@ -472,8 +472,7 @@ func (gp *GenericPool) GetFuncSvc(m *fission.Metadata) (*funcSvc, error) { err, existingFsvc := gp.fsCache.Add(*fsvc) if err != nil { // Some other thread beat us to it -- return the other thread's fsvc and clean up - // our own. TODO: this is grossly inefficient, improve it with some sort of state - // machine + // our own. log.Printf("func svc already exists: %v", existingFsvc.podName) go func() { gp.kubernetesClient.Core().Pods(gp.namespace).Delete(fsvc.podName, nil) diff --git a/router/functionServiceMap.go b/router/functionServiceMap.go index bb9c8470..6b08adf7 100644 --- a/router/functionServiceMap.go +++ b/router/functionServiceMap.go @@ -45,9 +45,12 @@ func (fmap *functionServiceMap) lookup(f *fission.Metadata) (*url.URL, error) { } func (fmap *functionServiceMap) assign(f *fission.Metadata, serviceUrl *url.URL) { - err, _ := fmap.cache.Set(*f, serviceUrl) + err, old := fmap.cache.Set(*f, serviceUrl) if err != nil { - log.Printf("error caching service url for function: %v", err) + if *serviceUrl == *(old.(*url.URL)) { + return + } + log.Printf("error caching service url for function with a different value: %v", err) // ignore error } }