more

Author: deads2k

Diff for: staging/src/k8s.io/apiserver/pkg/storage/cacher/cache_watcher.go

@@ -124,24 +124,31 @@ func (c *cacheWatcher) Stop() {
 
 // we rely on the fact that stopLocked is actually protected by Cacher.Lock()
 func (c *cacheWatcher) stopLocked() {
+	fmt.Printf("#### 4a groupResource=%v \n", c.groupResource)
 	if !c.stopped {
+		fmt.Printf("#### 4b groupResource=%v \n", c.groupResource)
 		c.stopped = true
 		// stop without draining the input channel was requested.
 		if !c.drainInputBuffer {
+			fmt.Printf("#### 4c groupResource=%v \n", c.groupResource)
 			close(c.done)
 		}
+		fmt.Printf("#### 4d groupResource=%v \n", c.groupResource)
 		close(c.input)
 	}
 
+	fmt.Printf("#### 4e groupResource=%v \n", c.groupResource)
 	// Even if the watcher was already stopped, if it previously was
 	// using draining mode and it's not using it now we need to
 	// close the done channel now. Otherwise we could leak the
 	// processing goroutine if it will be trying to put more objects
 	// into result channel, the channel will be full and there will
 	// already be noone on the processing the events on the receiving end.
 	if !c.drainInputBuffer && !c.isDoneChannelClosedLocked() {
+		fmt.Printf("#### 4f groupResource=%v \n", c.groupResource)
 		close(c.done)
 	}
+	fmt.Printf("#### 4g groupResource=%v \n", c.groupResource)
 }
 
 func (c *cacheWatcher) nonblockingAdd(event *watchCacheEvent) bool {

Diff for: staging/src/k8s.io/apiserver/pkg/storage/cacher/cacher.go

@@ -661,6 +661,7 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions
 	}
 
 	addedWatcher := false
+	fmt.Printf("#### 1n groupResource=%v \n", c.groupResource)
 	func() {
 		c.Lock()
 		defer c.Unlock()
@@ -673,19 +674,24 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions
 			return
 		}
 
+		fmt.Printf("#### 1p groupResource=%v \n", c.groupResource)
 		// Update watcher.forget function once we can compute it.
 		watcher.forget = forgetWatcher(c, watcher, c.watcherIdx, scope, triggerValue, triggerSupported)
 		// Update the bookMarkAfterResourceVersion
 		watcher.setBookmarkAfterResourceVersion(bookmarkAfterResourceVersionFn())
 		c.watchers.addWatcher(watcher, c.watcherIdx, scope, triggerValue, triggerSupported)
 		addedWatcher = true
 
+		fmt.Printf("#### 1q groupResource=%v \n", c.groupResource)
 		// Add it to the queue only when the client support watch bookmarks.
 		if watcher.allowWatchBookmarks {
 			c.bookmarkWatchers.addWatcherThreadUnsafe(watcher)
 		}
 		c.watcherIdx++
+
+		fmt.Printf("#### 1r groupResource=%v \n", c.groupResource)
 	}()
+	fmt.Printf("#### 1s groupResource=%v \n", c.groupResource)
 
 	if !addedWatcher {
 		fmt.Printf("#### 1x groupResource=%v returning the immediate closer thing\n", c.groupResource)
@@ -696,6 +702,7 @@ func (c *Cacher) Watch(ctx context.Context, key string, opts storage.ListOptions
 		return newImmediateCloseWatcher(), nil
 	}
 
+	fmt.Printf("#### 1y groupResource=%v \n", c.groupResource)
 	go watcher.processInterval(ctx, cacheInterval, requiredResourceVersion)
 	return watcher, nil
 }
@@ -1342,13 +1349,18 @@ func forgetWatcher(c *Cacher, w *cacheWatcher, index int, scope namespacedName,
 		c.Lock()
 		defer c.Unlock()
 
+		fmt.Printf("#### 3a groupResource=%v \n", c.groupResource)
+
 		w.setDrainInputBufferLocked(drainWatcher)
+		fmt.Printf("#### 3b groupResource=%v \n", c.groupResource)
 
 		// It's possible that the watcher is already not in the structure (e.g. in case of
 		// simultaneous Stop() and terminateAllWatchers(), but it is safe to call stopLocked()
 		// on a watcher multiple times.
 		c.watchers.deleteWatcher(index, scope, triggerValue, triggerSupported)
+		fmt.Printf("#### 3c groupResource=%v \n", c.groupResource)
 		c.stopWatcherLocked(w)
+		fmt.Printf("#### 3d groupResource=%v \n", c.groupResource)
 	}
 }