github/kubernetes
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Avoid unnecessary copies in cacher

Browse Source
This commit is contained in:
Wojciech Tyczynski
2016-10-18 10:42:00 +02:00
parent 6a1e8c98e0
commit 9895f337ee
2 changed files with 13 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
pkg/storage/cacher.go
View File

@@ -707,7 +707,8 @@ func (c *cacheWatcher) add(event *watchCacheEvent) {
} }
} }
func (c *cacheWatcher) sendWatchCacheEvent(event watchCacheEvent) { // NOTE: sendWatchCacheEvent is assumed to not modify <event> !!!
func (c *cacheWatcher) sendWatchCacheEvent(event *watchCacheEvent) {
curObjPasses := event.Type != watch.Deleted && c.filter(event.Object) curObjPasses := event.Type != watch.Deleted && c.filter(event.Object)
oldObjPasses := false oldObjPasses := false
if event.PrevObject != nil { if event.PrevObject != nil {
@@ -752,7 +753,7 @@ func (c *cacheWatcher) process(initEvents []watchCacheEvent, resourceVersion uin
const initProcessThreshold = 500 * time.Millisecond const initProcessThreshold = 500 * time.Millisecond
startTime := time.Now() startTime := time.Now()
for _, event := range initEvents { for _, event := range initEvents {
c.sendWatchCacheEvent(event) c.sendWatchCacheEvent(&event)
} }
processingTime := time.Since(startTime) processingTime := time.Since(startTime)
if processingTime > initProcessThreshold { if processingTime > initProcessThreshold {
@@ -772,7 +773,7 @@ func (c *cacheWatcher) process(initEvents []watchCacheEvent, resourceVersion uin
} }
// only send events newer than resourceVersion // only send events newer than resourceVersion
if event.ResourceVersion > resourceVersion { if event.ResourceVersion > resourceVersion {
c.sendWatchCacheEvent(event) c.sendWatchCacheEvent(&event)
} }
} }
} }

13
pkg/storage/watch_cache.go
View File

@@ -181,23 +181,28 @@ func (w *watchCache) processEvent(event watch.Event, resourceVersion uint64, upd
if exists { if exists {
prevObject = previous.(runtime.Object) prevObject = previous.(runtime.Object)
} }
watchCacheEvent := watchCacheEvent{event.Type, event.Object, prevObject, resourceVersion} watchCacheEvent := watchCacheEvent{
Type: event.Type,
Object: event.Object,
PrevObject: prevObject,
ResourceVersion: resourceVersion,
}
if w.onEvent != nil { if w.onEvent != nil {
w.onEvent(watchCacheEvent) w.onEvent(watchCacheEvent)
} }
w.updateCache(resourceVersion, watchCacheEvent) w.updateCache(resourceVersion, &watchCacheEvent)
w.resourceVersion = resourceVersion w.resourceVersion = resourceVersion
w.cond.Broadcast() w.cond.Broadcast()
return updateFunc(event.Object) return updateFunc(event.Object)
} }
// Assumes that lock is already held for write. // Assumes that lock is already held for write.
func (w *watchCache) updateCache(resourceVersion uint64, event watchCacheEvent) { func (w *watchCache) updateCache(resourceVersion uint64, event *watchCacheEvent) {
if w.endIndex == w.startIndex+w.capacity { if w.endIndex == w.startIndex+w.capacity {
// Cache is full - remove the oldest element. // Cache is full - remove the oldest element.
w.startIndex++ w.startIndex++
} }
w.cache[w.endIndex%w.capacity] = watchCacheElement{resourceVersion, event} w.cache[w.endIndex%w.capacity] = watchCacheElement{resourceVersion, *event}
w.endIndex++ w.endIndex++
} }