Prevent the router from deadlocking itself when calling Commit()

Backport of #13717

The router reload function (Commit()) was changed so that it could be
rate limited.  The rate limiter tracks changes in a kcache.Queue
object.  It will coalesce changes to the same call so that if three
calls to Invoke() the rate limited function happen before the next
time it is allowed to process, then only one will occur.

Our problem was that we were doing:
 Thread 1 (the rate-limiter background process):
   - Wake up
   - Ratelimit sees there is work to be done and calls fifo.go's Pop() function
   - Fifo.go acquires a fifo lock and call the processing function
   - Router.go's commitAndReload() function acquires a lock on the router object
 Thread 2 (triggered by the event handler that commit's changes to the router):
   - Get the event and process it
   - Since there are changes to be made, call router.Commit()
   - Commit() grabs a lock on the router object
   - Then calls the rate-limiter wrapper around commitAndReload() using Invoke() to queue work
   - In order to queue the work... it acquires a lock on the fifo

So thread 1 locks: fifo then router; thread 2 locks: router then fifo.
If you get unlucky, those threads deadlock and you never process
another event.

The fix is to release the lock on the router object in our Commit()
function before we call Invoke on the rate limited function.  The lock
is not actually protecting anything at that point since the rate
limited function does its own locking, and is run in a separate thread
anyway.

Bug 1440977 (https://bugzilla.redhat.com/show_bug.cgi?id=1440977)

Author: knobunc

pkg/router/template/fake.go

@@ -13,6 +13,17 @@ func NewFakeTemplateRouter() *templateRouter {
 	}
 }
 
+// FakeReloadHandler implements the minimal changes needed to make the locking behavior work
+// This MUST match the behavior with the stateChanged of commitAndReload
+func (r *templateRouter) FakeReloadHandler() {
+	r.lock.Lock()
+	defer r.lock.Unlock()
+
+	r.stateChanged = false
+
+	return
+}
+
 // fakeCertWriter is a certificate writer that records actions but is a no-op
 type fakeCertWriter struct {
 	addedCerts   []string

pkg/router/template/router.go

@@ -366,19 +366,21 @@ func (r *templateRouter) readState() error {
 // Commit applies the changes made to the router configuration - persists
 // the state and refresh the backend. This is all done in the background
 // so that we can rate limit + coalesce multiple changes.
+// Note: If this is changed FakeCommit() in fake.go should also be updated
 func (r *templateRouter) Commit() {
 	r.lock.Lock()
-	defer r.lock.Unlock()
 
 	if !r.synced {
 		glog.V(4).Infof("Router state synchronized for the first time")
 		r.synced = true
 		r.stateChanged = true
 	}
 
-	if r.stateChanged {
+	needsCommit := r.stateChanged
+	r.lock.Unlock()
+
+	if needsCommit {
 		r.rateLimitedCommitFunction.Invoke(r.rateLimitedCommitFunction)
-		r.stateChanged = false
 	}
 }
 
@@ -387,6 +389,8 @@ func (r *templateRouter) commitAndReload() error {
 	r.lock.Lock()
 	defer r.lock.Unlock()
 
+	r.stateChanged = false
+
 	glog.V(4).Infof("Writing the router state")
 	if err := r.writeState(); err != nil {
 		return err

test/integration/router_stress_test.go

@@ -112,7 +112,7 @@ func stressRouter(t *testing.T, namespaceCount, routesPerNamespace, routerCount,
 	// Create the routers
 	for i := int32(0); i < routerCount; i++ {
 		routerName := fmt.Sprintf("router_%d", i)
-		router := launchRouter(oc, kc, maxRouterDelay, routerName, reloadInterval, reloadedMap)
+		router := launchRouter(t, oc, kc, maxRouterDelay, routerName, reloadInterval, reloadedMap)
 		plugins = append(plugins, router)
 	}
 
@@ -148,12 +148,12 @@ func stressRouter(t *testing.T, namespaceCount, routesPerNamespace, routerCount,
 		}
 	}
 
-	for _, reloadCount := range reloadedMap {
+	for routerName, reloadCount := range reloadedMap {
 		if reloadCount > 1 {
 			// If a router reloads more than once, post-sync watch
 			// events resulting from route status updates are
 			// incorrectly updating router state.
-			t.Fatalf("One or more routers reloaded more than once")
+			t.Fatalf("One or more routers reloaded more than once (%s reloaded %d times)", routerName, reloadCount)
 		}
 	}
 
@@ -281,12 +281,14 @@ func (p *DelayPlugin) Commit() error {
 
 // launchRouter launches a template router that communicates with the
 // api via the provided clients.
-func launchRouter(oc osclient.Interface, kc kclientset.Interface, maxDelay int32, name string, reloadInterval int, reloadedMap map[string]int) (templatePlugin *templateplugin.TemplatePlugin) {
+func launchRouter(t *testing.T, oc osclient.Interface, kc kclientset.Interface, maxDelay int32, name string, reloadInterval int, reloadedMap map[string]int) (templatePlugin *templateplugin.TemplatePlugin) {
 	r := templateplugin.NewFakeTemplateRouter()
 
 	reloadedMap[name] = 0
 	r.EnableRateLimiter(reloadInterval, func() error {
 		reloadedMap[name] += 1
+		t.Logf("Router %s reloaded (%d times)\n", name, reloadedMap[name])
+		r.FakeReloadHandler()
 		return nil
 	})