Bugfix: deadlock when domain failover (#919)

* fix deadlock issue in timer queue processor failover * fix domain cache copy without locking issue
cadence-workflow · Jul 3, 2018 · 2b92fd6 · 2b92fd6
1 parent 7eaf360
commit 2b92fd6
Show file tree

Hide file tree

Showing 3 changed files with 8 additions and 7 deletions.
diff --git a/common/cache/domainCache.go b/common/cache/domainCache.go
@@ -173,8 +173,11 @@ func (c *domainCache) GetAllDomain() map[string]*DomainCacheEntry {
 	for ite.HasNext() {
 		entry := ite.Next()
 		id := entry.Key().(string)
-		domainCacheEntry := entry.Value().(*DomainCacheEntry).duplicate()
-		result[id] = domainCacheEntry
+		domainCacheEntry := entry.Value().(*DomainCacheEntry)
+		domainCacheEntry.RLock()
+		dup := domainCacheEntry.duplicate()
+		domainCacheEntry.RUnlock()
+		result[id] = dup
 	}
 	return result
 }
@@ -185,13 +188,13 @@ func (c *domainCache) GetAllDomain() map[string]*DomainCacheEntry {
 // afterCallback will be invoked when NOT holding the domain cache lock.
 func (c *domainCache) RegisterDomainChangeCallback(shard int, initialNotificationVersion int64, beforeCallback CallbackFn, afterCallback CallbackFn) {
 	c.Lock()
-	defer c.Unlock()
-
 	c.beforeCallbacks[shard] = beforeCallback
 	c.afterCallbacks[shard] = afterCallback
+	domainNotificationVersion := c.domainNotificationVersion
+	c.Unlock()
 
 	// this section is trying to make the shard catch up with domain changes
-	if c.domainNotificationVersion > initialNotificationVersion {
+	if domainNotificationVersion > initialNotificationVersion {
 		domains := DomainCacheEntries{}
 		for _, domain := range c.GetAllDomain() {
 			domains = append(domains, domain)

diff --git a/service/history/timerQueueProcessor.go b/service/history/timerQueueProcessor.go
@@ -143,7 +143,6 @@ func (t *timerQueueProcessorImpl) FailoverDomain(domainID string) {
 	failoverTimerProcessor := newTimerQueueFailoverProcessor(t.shard, t.historyService, domainID,
 		standbyClusterName, minLevel, maxLevel, t.matchingClient, t.logger)
 	failoverTimerProcessor.Start()
-	failoverTimerProcessor.timerQueueProcessorBase.readAndFanoutTimerTasks()
 }
 
 func (t *timerQueueProcessorImpl) getTimerFiredCount(clusterName string) uint64 {

diff --git a/service/history/transferQueueProcessor.go b/service/history/transferQueueProcessor.go
@@ -157,7 +157,6 @@ func (t *transferQueueProcessorImpl) FailoverDomain(domainID string) {
 		domainID, standbyClusterName, minLevel, maxLevel, t.logger,
 	)
 	failoverTaskProcessor.Start()
-	failoverTaskProcessor.notifyNewTask()
 }
 
 func (t *transferQueueProcessorImpl) completeTransferLoop() {