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

Optimize function latency when cache expired/invalid under high concurrency (#856)

Browse Source 
* Optimize router response time by adding update lock

In 0.9.2, the router sends multiple GetServiceForFunction requests to executor
to get the service URL. However, the response time of executor will increase
under high-concurrency situation due to too many requests are waiting for
processing.

To solve the problem, an update lock was added to the router. All of goroutines belongs
to the same function need to grab the update lock before sending the request.
Only the first goroutine which gets the update lock is allowed to send request.
In this way, we reduce the burden of executor and lower the failure rate.
This commit is contained in:
Ta-Ching Chen
2018-10-27 23:35:00 +08:00
committed by GitHub
parent 394c5b13f4
commit 29aabaabda
8 changed files with 409 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files
charts/fission-all/templates/deployment.yaml
+4
View File
@@ -200,6 +200,10 @@ spec:
value: {{ .Values.routerRoundTripKeepAliveTime | default "30s" | quote }}
- name: ROUTER_ROUND_TRIP_MAX_RETRIES
value: {{ .Values.routerRoundTripMaxRetries | default 10 | quote }}
- name: ROUTER_ROUND_TRIP_SVC_ADDRESS_MAX_RETRIES
value: {{ .Values.routerRoundTripSvcAddressMaxRetries | default 5 | quote }}
- name: ROUTER_ROUND_TRIP_SVC_ADDRESS_UPDATE_TIMEOUT
value: {{ .Values.routerRoundTripSvcAddressUpdateTimeout | default 30 | quote }}
- name: DEBUG_ENV
value: {{ .Values.debugEnv | quote }}
readinessProbe:
charts/fission-core/templates/deployment.yaml
+4
View File
@@ -197,6 +197,10 @@ spec:
value: {{ .Values.routerRoundTripKeepAliveTime | default "30s" | quote }}
- name: ROUTER_ROUND_TRIP_MAX_RETRIES
value: {{ .Values.routerRoundTripMaxRetries | default 10 | quote }}
- name: ROUTER_ROUND_TRIP_SVC_ADDRESS_MAX_RETRIES
value: {{ .Values.routerRoundTripSvcAddressMaxRetries | default 5 | quote }}
- name: ROUTER_ROUND_TRIP_SVC_ADDRESS_UPDATE_TIMEOUT
value: {{ .Values.routerRoundTripSvcAddressUpdateTimeout | default 30 | quote }}
- name: DEBUG_ENV
value: {{ .Values.debugEnv | quote }}
readinessProbe:
executor/executor.go
+7 -7
View File
@@ -142,13 +142,6 @@ func (executor *Executor) getFunctionExecutorType(meta *metav1.ObjectMeta) (fiss
func (executor *Executor) createServiceForFunction(meta *metav1.ObjectMeta) (*fscache.FuncSvc, error) {
log.Printf("[%v] No cached function service found, creating one", meta.Name)
// from Func -> get Env
log.Printf("[%v] getting environment for function", meta.Name)
env, err := executor.getFunctionEnv(meta)
if err != nil {
return nil, err
}
executorType, err := executor.getFunctionExecutorType(meta)
if err != nil {
return nil, err
@@ -161,6 +154,13 @@ func (executor *Executor) createServiceForFunction(meta *metav1.ObjectMeta) (*fs
case fission.ExecutorTypeNewdeploy:
fsvc, fsvcErr = executor.ndm.GetFuncSvc(meta)
default:
// from Func -> get Env
log.Printf("[%v] getting environment for function", meta.Name)
env, err := executor.getFunctionEnv(meta)
if err != nil {
return nil, err
}
pool, err := executor.gpm.GetPool(env)
if err != nil {
return nil, err
router/functionHandler.go
+185 -110
View File
@@ -29,8 +29,8 @@ import (
"time"
"github.com/gorilla/mux"
"github.com/pkg/errors"
"github.com/satori/go.uuid"
log "github.com/sirupsen/logrus"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -49,7 +49,19 @@ type tsRoundTripperParams struct {
timeout time.Duration
timeoutExponent int
keepAlive time.Duration
maxRetries int
// maxRetires is the max times for RetryingRoundTripper to retry a request.
// Default maxRetries is 10, which means router will retry for
// up to 10 times and abort it if still not succeeded.
maxRetries int
// svcAddrRetryCount is the max times for RetryingRoundTripper to retry with a specific service address
// Router sends requests to a specific service address for each function.
// A service address is considered as an invalid one if amount of non-network
// errors router received is higher than svcAddrRetryCount. In this situation,
// remove it from cache and try to get a new one from executor.
// Default svcAddrRetryCount is 5.
svcAddrRetryCount int
}
type functionHandler struct {
@@ -64,6 +76,7 @@ type functionHandler struct {
tsRoundTripperParams *tsRoundTripperParams
recorderName string
isDebugEnv bool
svcAddrUpdateLocks *svcAddrUpdateLocks
}
// A layer on top of http.DefaultTransport, with retries.
@@ -103,8 +116,9 @@ func init() {
// Earlier, GetServiceForFunction was called inside handler function and fission explicitly set http status code to 500
// if it returned an error.
func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *http.Response, err error) {
var needExecutor, serviceUrlFromExecutor bool
var serviceUrlFromCache bool
var serviceUrl *url.URL
var retryCounter int
// Set forwarded host header if not exists
addForwardedHostHeader(req)
@@ -130,11 +144,13 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt
}
}
fnMeta := roundTripper.funcHandler.function
// Metrics stuff
startTime := time.Now()
funcMetricLabels := &functionLabels{
namespace: roundTripper.funcHandler.function.Namespace,
name: roundTripper.funcHandler.function.Name,
namespace: fnMeta.Namespace,
name: fnMeta.Name,
}
httpMetricLabels := &httpLabels{
method: req.Method,
@@ -146,22 +162,150 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt
// set the timeout for transport context
transport := http.DefaultTransport.(*http.Transport)
// Disables caching, Please refer to issue and specifically comment: https://github.com/fission/fission/issues/723#issuecomment-398781995
transport.DisableKeepAlives = true
// cache lookup to get serviceUrl
serviceUrl, err = roundTripper.funcHandler.fmap.lookup(roundTripper.funcHandler.function)
if err != nil || serviceUrl == nil {
// cache miss or nil entry in cache
log.Printf("Setting needExecutor to true for function : %s", roundTripper.funcHandler.function.Name)
needExecutor = true
}
executingTimeout := roundTripper.funcHandler.tsRoundTripperParams.timeout
for i := 0; i < roundTripper.funcHandler.tsRoundTripperParams.maxRetries-1; i++ {
if needExecutor {
log.Printf("Calling getServiceForFunction for function: %s", roundTripper.funcHandler.function.Name)
// cache lookup to get serviceUrl
serviceUrl, err = roundTripper.funcHandler.fmap.lookup(fnMeta)
if err != nil {
e, ok := err.(fission.Error)
if (ok && e.Code != fission.ErrorNotFound) || !ok {
if ok {
err = errors.Wrap(err, fmt.Sprintf("Error getting function %v;s service entry from cache", fnMeta.Name))
} else {
err = errors.Wrap(err, "Unknown error when looking up service entry")
}
return nil, err
}
} else {
serviceUrlFromCache = true
}
if serviceUrl != nil {
// modify the request to reflect the service url
// this service url may have come from the cache lookup or from executor response
req.URL.Scheme = serviceUrl.Scheme
req.URL.Host = serviceUrl.Host
// To keep the function run container simple, it
// doesn't do any routing. In the future if we have
// multiple functions per container, we could use the
// function metadata here.
// leave the query string intact (req.URL.RawQuery)
req.URL.Path = "/"
// Overwrite request host with internal host,
// or request will be blocked in some situations
// (e.g. istio-proxy)
req.Host = serviceUrl.Host
// over-riding default settings.
transport.DialContext = (&net.Dialer{
Timeout: executingTimeout,
KeepAlive: roundTripper.funcHandler.tsRoundTripperParams.keepAlive,
}).DialContext
overhead := time.Since(startTime)
// tapService before invoking roundTrip for the serviceUrl
if serviceUrlFromCache {
go roundTripper.funcHandler.tapService(serviceUrl)
}
// forward the request to the function service
resp, err = transport.RoundTrip(req)
if err == nil {
// Track metrics
httpMetricLabels.code = resp.StatusCode
funcMetricLabels.cached = serviceUrlFromCache
functionCallCompleted(funcMetricLabels, httpMetricLabels,
overhead, time.Since(startTime), resp.ContentLength)
if len(roundTripper.funcHandler.recorderName) > 0 {
if roundTripper.funcHandler.httpTrigger != nil {
trigger := roundTripper.funcHandler.httpTrigger.Metadata.Name
redis.Record(
trigger,
roundTripper.funcHandler.recorderName,
req.Header.Get("X-Fission-ReqUID"), req, originalUrl, postedBody, resp, fnMeta.Namespace,
time.Now().UnixNano(),
)
} else {
log.Println("No http trigger attached for recorder: %v", roundTripper.funcHandler.recorderName)
}
}
// return response back to user
return resp, nil
}
// if transport.RoundTrip returns a non-network dial error, then relay it back to user
if !fission.IsNetworkDialError(err) {
err = errors.Wrapf(err, "Error sending request to function %v", fnMeta.Name)
return resp, err
}
// dial timeout or dial network errors goes here
// The reason for request failure may vary from case to case.
// After some investigation, found most of the failure are due to
// network timeout or target function is under heavy workload. In
// such cases, if router keeps trying to get new function service
// will increase executor burden and cause 502 error.
//
// The "retryCounter" was introduced to solve this problem by retrying
// requests for "limited threshold". Once a request's retryCounter higher
// than the predefined threshold, reset retryCounter and remove service
// cache, then retry to get new svc record from executor again.
if retryCounter < roundTripper.funcHandler.tsRoundTripperParams.svcAddrRetryCount {
retryCounter++
executingTimeout = executingTimeout * time.Duration(roundTripper.funcHandler.tsRoundTripperParams.timeoutExponent)
log.Printf("request to %s errored out. backing off for %v before retrying",
req.URL.Host, executingTimeout)
time.Sleep(executingTimeout)
if serviceUrlFromCache {
continue
}
} else {
// if transport.RoundTrip returns a network dial error and serviceUrl was from cache,
// it means, the entry in router cache is stale, so invalidate it.
log.Printf("request to %s errored out. removing function : %s from router's cache "+
"and requesting a new service for function",
req.URL.Host, fnMeta.Name)
roundTripper.funcHandler.fmap.remove(fnMeta)
retryCounter = 0
}
}
// break directly if we still fail at the last round
if i >= roundTripper.funcHandler.tsRoundTripperParams.maxRetries-1 {
break
}
// cache miss or nil entry in cache
lock, ableToUpdateCache := roundTripper.funcHandler.grabUpdateEntryLock(fnMeta)
if !ableToUpdateCache {
// This goroutine wait for update of service map to finish.
err = lock.Wait()
if err != nil {
log.Println(errors.Wrap(err,
fmt.Sprintf("Error updating service address entry for function %v_%v", fnMeta.Name, fnMeta.Namespace)))
}
} else {
// This goroutine is the first one to grab update lock
log.Printf("Calling getServiceForFunction for function: %s", fnMeta.Name)
// send a request to executor to specialize a new pod
service, err := roundTripper.funcHandler.executor.GetServiceForFunction(
@@ -169,7 +313,7 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt
if err != nil {
statusCode, errMsg := fission.GetHTTPError(err)
log.Printf("Err from GetServiceForFunction : %v : %v", statusCode, errMsg)
log.Printf("Err from GetServiceForFunction for function (%v): %v : %v", roundTripper.funcHandler.function, statusCode, errMsg)
// We might want a specific error code or header for fission failures as opposed to
// user function bugs.
@@ -186,118 +330,37 @@ func (roundTripper RetryingRoundTripper) RoundTrip(req *http.Request) (resp *htt
}, nil
}
roundTripper.funcHandler.releaseUpdateEntryLock(fnMeta)
return nil, err
}
// parse the address into url
serviceUrl, err = url.Parse(fmt.Sprintf("http://%v", service))
if err != nil {
log.Printf("Error parsing service url (%v): %v", serviceUrl, err)
roundTripper.funcHandler.releaseUpdateEntryLock(fnMeta)
return nil, err
}
// add the address in router's cache
log.Printf("assigning serviceUrl : %s for function : %s", serviceUrl, roundTripper.funcHandler.function.Name)
log.Printf("Assigning serviceUrl : %s for function : %s", serviceUrl, roundTripper.funcHandler.function.Name)
roundTripper.funcHandler.fmap.assign(roundTripper.funcHandler.function, serviceUrl)
// flag denotes that service was not obtained from cache, instead, created just now by executor
serviceUrlFromExecutor = true
}
serviceUrlFromCache = false
// modify the request to reflect the service url
// this service url may have come from the cache lookup or from executor response
req.URL.Scheme = serviceUrl.Scheme
req.URL.Host = serviceUrl.Host
// To keep the function run container simple, it
// doesn't do any routing. In the future if we have
// multiple functions per container, we could use the
// function metadata here.
// leave the query string intact (req.URL.RawQuery)
req.URL.Path = "/"
// Overwrite request host with internal host,
// or request will be blocked in some situations
// (e.g. istio-proxy)
req.Host = serviceUrl.Host
// over-riding default settings.
transport.DialContext = (&net.Dialer{
Timeout: executingTimeout,
KeepAlive: roundTripper.funcHandler.tsRoundTripperParams.keepAlive,
}).DialContext
overhead := time.Since(startTime)
// tapService before invoking roundTrip for the serviceUrl
if !serviceUrlFromExecutor {
go roundTripper.funcHandler.tapService(serviceUrl)
}
// forward the request to the function service
resp, err = transport.RoundTrip(req)
if err == nil {
// Track metrics
httpMetricLabels.code = resp.StatusCode
funcMetricLabels.cached = !serviceUrlFromExecutor
functionCallCompleted(funcMetricLabels, httpMetricLabels,
overhead, time.Since(startTime), resp.ContentLength)
// if transport.RoundTrip succeeds and it was a cached entry, then tapService
if !serviceUrlFromExecutor {
go roundTripper.funcHandler.tapService(serviceUrl)
}
trigger := ""
if roundTripper.funcHandler.httpTrigger != nil {
trigger = roundTripper.funcHandler.httpTrigger.Metadata.Name
} else {
log.Println("No trigger attached.") // Wording?
}
if len(roundTripper.funcHandler.recorderName) > 0 {
redis.Record(
trigger,
roundTripper.funcHandler.recorderName,
req.Header.Get("X-Fission-ReqUID"), req, originalUrl, postedBody, resp, roundTripper.funcHandler.function.Namespace,
time.Now().UnixNano(),
)
}
// return response back to user
return resp, nil
}
// if transport.RoundTrip returns a non-network dial error, then relay it back to user
if !fission.IsNetworkDialError(err) {
return resp, err
}
// means its a newly created service and it returned a network dial error.
// just retry after backing off for timeout period.
if serviceUrlFromExecutor {
log.Printf("request to %s errored out. backing off for %v before retrying",
req.URL.Host, executingTimeout)
executingTimeout = executingTimeout * time.Duration(roundTripper.funcHandler.tsRoundTripperParams.timeoutExponent)
time.Sleep(executingTimeout)
needExecutor = false
continue
} else {
// if transport.RoundTrip returns a network dial error and serviceUrl was from cache,
// it means, the entry in router cache is stale, so invalidate it.
// also set needExecutor to true so a new service can be requested for function.
log.Printf("request to %s errored out. removing function : %s from router's cache "+
"and requesting a new service for function",
req.URL.Host, roundTripper.funcHandler.function.Name)
roundTripper.funcHandler.fmap.remove(roundTripper.funcHandler.function)
needExecutor = true
roundTripper.funcHandler.releaseUpdateEntryLock(fnMeta)
}
}
// finally, one more retry with the default timeout
return http.DefaultTransport.RoundTrip(req)
resp, err = http.DefaultTransport.RoundTrip(req)
if err != nil {
log.Printf("Error getting response from function %v: %v",
fnMeta.Name, err)
}
return resp, err
}
func (fh *functionHandler) tapService(serviceUrl *url.URL) {
@@ -429,3 +492,15 @@ func addForwardedHostHeader(req *http.Request) {
req.Header.Set(FORWARDED, host)
req.Header.Set(X_FORWARDED_HOST, req.Host)
}
// grabUpdateEntryLock helps goroutine to grab update lock for updating function service cache.
// If the update lock exists, return old svcAddrUpdateLock.
func (fh *functionHandler) grabUpdateEntryLock(fnMeta *metav1.ObjectMeta) (lock *svcAddrUpdateLock, ableToUpdateCache bool) {
return fh.svcAddrUpdateLocks.Get(fnMeta)
}
// releaseUpdateEntryLock release update lock so that other goroutines can take over the responsibility
// of updating the service map.
func (fh *functionHandler) releaseUpdateEntryLock(fnMeta *metav1.ObjectMeta) {
fh.svcAddrUpdateLocks.Delete(fnMeta)
}
router/httpTriggers.go
+13 -1
View File
@@ -53,10 +53,12 @@ type HTTPTriggerSet struct {
updateRouterRequestChannel chan struct{}
tsRoundTripperParams *tsRoundTripperParams
isDebugEnv bool
svcAddrUpdateLocks *svcAddrUpdateLocks
}
func makeHTTPTriggerSet(fmap *functionServiceMap, frmap *functionRecorderMap, trmap *triggerRecorderMap, fissionClient *crd.FissionClient,
kubeClient *kubernetes.Clientset, executor *executorClient.Client, crdClient *rest.RESTClient, params *tsRoundTripperParams, isDebugEnv bool) (*HTTPTriggerSet, k8sCache.Store, k8sCache.Store) {
kubeClient *kubernetes.Clientset, executor *executorClient.Client, crdClient *rest.RESTClient, params *tsRoundTripperParams, isDebugEnv bool, locks *svcAddrUpdateLocks) (*HTTPTriggerSet, k8sCache.Store, k8sCache.Store) {
httpTriggerSet := &HTTPTriggerSet{
functionServiceMap: fmap,
triggers: []crd.HTTPTrigger{},
@@ -67,6 +69,7 @@ func makeHTTPTriggerSet(fmap *functionServiceMap, frmap *functionRecorderMap, tr
updateRouterRequestChannel: make(chan struct{}),
tsRoundTripperParams: params,
isDebugEnv: isDebugEnv,
svcAddrUpdateLocks: locks,
}
var tStore, fnStore, rStore k8sCache.Store
var tController, fnController k8sCache.Controller
@@ -153,8 +156,16 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router {
tsRoundTripperParams: ts.tsRoundTripperParams,
recorderName: recorderName,
isDebugEnv: ts.isDebugEnv,
svcAddrUpdateLocks: ts.svcAddrUpdateLocks,
}
// The functionHandler for HTTP trigger with fn reference type "FunctionReferenceTypeFunctionName",
// it's function metadata is set here.
// The functionHandler For HTTP trigger with fn reference type "FunctionReferenceTypeFunctionWeights",
// it's function metadata is decided dynamically before proxying the request in order to support canary
// deployment. For more details, please check "handler" function of functionHandler.
if rr.resolveResultType == resolveResultSingleFunction {
for _, metadata := range fh.functionMetadataMap {
fh.function = metadata
@@ -201,6 +212,7 @@ func (ts *HTTPTriggerSet) getRouter() *mux.Router {
tsRoundTripperParams: ts.tsRoundTripperParams,
recorderName: recorderName,
isDebugEnv: ts.isDebugEnv,
svcAddrUpdateLocks: ts.svcAddrUpdateLocks,
}
muxRouter.HandleFunc(fission.UrlForFunction(function.Metadata.Name, function.Metadata.Namespace), fh.handler)
}
router/router.go
+21 -5
View File
@@ -129,12 +129,28 @@ func Start(port int, executorUrl string) {
log.Fatalf("Failed to parse DEBUG_ENV: %v", err)
}
// svcAddrRetryCount is the max times for RetryingRoundTripper to retry with a specific service address
svcAddrRetryCount, err := strconv.Atoi(os.Getenv("ROUTER_ROUND_TRIP_SVC_ADDRESS_MAX_RETRIES"))
if err != nil {
svcAddrRetryCount = 5
log.Printf("Failed to parse svc address retry conunt, set it to default value(5): %v", err)
}
// svcAddrUpdateTimeout is the timeout setting for a goroutine to wait for the update of a service entry.
// If the update process cannot be done within the timeout window, consider it failed.
svcAddrUpdateTimeout, err := time.ParseDuration(os.Getenv("ROUTER_ROUND_TRIP_SVC_ADDRESS_UPDATE_TIMEOUT"))
if err != nil {
svcAddrUpdateTimeout = 30 * time.Second
log.Printf("Failed to parse svc address update timeout, set it to default value(30): %v", err)
}
triggers, _, fnStore := makeHTTPTriggerSet(fmap, frmap, trmap, fissionClient, kubeClient, executor, restClient, &tsRoundTripperParams{
timeout: timeout,
timeoutExponent: timeoutExponent,
keepAlive: keepAlive,
maxRetries: maxRetries,
}, isDebugEnv)
timeout: timeout,
timeoutExponent: timeoutExponent,
keepAlive: keepAlive,
maxRetries: maxRetries,
svcAddrRetryCount: svcAddrRetryCount,
}, isDebugEnv, MakeUpdateLocks(svcAddrUpdateTimeout))
resolver := makeFunctionReferenceResolver(fnStore)
router/router_test.go
+1 -1
View File
@@ -57,7 +57,7 @@ func TestRouter(t *testing.T) {
timeoutExponent: 2,
keepAlive: 30 * time.Second,
maxRetries: 10,
}, false)
}, false, MakeUpdateLocks(30*time.Second))
triggerUrl := "/foo"
triggers.triggers = append(triggers.triggers,
crd.HTTPTrigger{
router/svcAddrUpdateLock.go
+174
View File
@@ -0,0 +1,174 @@
/*
Copyright 2018 The Fission Authors.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package router
import (
"errors"
"sync"
"time"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/fission/fission/crd"
)
type svcAddrUpdateOperation int
const (
GET svcAddrUpdateOperation = iota
DELETE
EXPIRE
)
type (
// svcAddrUpdateLock is the lock that will be used when
// gorountine tries to update service address entry.
svcAddrUpdateLock struct {
wg *sync.WaitGroup
ctimestamp time.Time // creation time of lock
timeExpiry time.Duration
}
svcAddrUpdateLocks struct {
requestChan chan *svcAddrUpdateRequest
locks map[string]*svcAddrUpdateLock
lockTimeExpiry time.Duration
}
svcAddrUpdateRequest struct {
requestType svcAddrUpdateOperation
responseChan chan *svcAddrUpdateResponse
fnMeta *metav1.ObjectMeta
}
svcAddrUpdateResponse struct {
lock *svcAddrUpdateLock
loaded bool // denote the lock for same function already exists
}
)
func (l *svcAddrUpdateLock) isOld() bool {
return time.Since(l.ctimestamp) > l.timeExpiry
}
func (l *svcAddrUpdateLock) Wait() error {
ch := make(chan struct{})
go func(wg *sync.WaitGroup, ch chan struct{}) {
wg.Wait()
close(ch)
}(l.wg, ch)
select {
case <-ch:
return nil
case <-time.After(l.timeExpiry):
return errors.New("Error waiting for svcAddrUpdateLock to be released: Exceeded timeout")
}
}
func MakeUpdateLocks(timeExpiry time.Duration) *svcAddrUpdateLocks {
locks := &svcAddrUpdateLocks{
requestChan: make(chan *svcAddrUpdateRequest),
locks: make(map[string]*svcAddrUpdateLock),
lockTimeExpiry: timeExpiry,
}
go locks.service()
go locks.expiryService()
return locks
}
func (ul *svcAddrUpdateLocks) service() {
for {
req := <-ul.requestChan
switch req.requestType {
case GET:
key := crd.CacheKey(req.fnMeta)
lock, ok := ul.locks[key]
if ok && !lock.isOld() {
req.responseChan <- &svcAddrUpdateResponse{
lock: lock, loaded: false,
}
continue
} else if ok && lock.isOld() {
// in case that one goroutine occupy the update lock for long time
lock.wg.Done()
}
lock = &svcAddrUpdateLock{
wg: &sync.WaitGroup{},
ctimestamp: time.Now(),
timeExpiry: ul.lockTimeExpiry,
}
lock.wg.Add(1)
ul.locks[key] = lock
req.responseChan <- &svcAddrUpdateResponse{
lock: lock, loaded: true,
}
case DELETE:
key := crd.CacheKey(req.fnMeta)
lock, ok := ul.locks[key]
if ok {
lock.wg.Done()
delete(ul.locks, key)
}
case EXPIRE:
for k, v := range ul.locks {
if v.isOld() {
delete(ul.locks, k)
v.wg.Done()
}
}
}
}
}
func (locks *svcAddrUpdateLocks) Get(fnMeta *metav1.ObjectMeta) (lock *svcAddrUpdateLock, ableToUpdate bool) {
ch := make(chan *svcAddrUpdateResponse)
locks.requestChan <- &svcAddrUpdateRequest{
requestType: GET,
responseChan: ch,
fnMeta: fnMeta,
}
resp := <-ch
return resp.lock, resp.loaded
}
func (locks *svcAddrUpdateLocks) Delete(fnMeta *metav1.ObjectMeta) {
locks.requestChan <- &svcAddrUpdateRequest{
requestType: DELETE,
fnMeta: fnMeta,
}
}
// expiryService periodically expires time-out locks.
// Normally, we don't need to do this just in case any of goroutine didn't release lock.
func (locks *svcAddrUpdateLocks) expiryService() {
for {
time.Sleep(time.Minute)
locks.requestChan <- &svcAddrUpdateRequest{
requestType: EXPIRE,
}
}
}