Unload taskListManager by instance, not taskListID (uber#5101)

service/matching/matchingEngine.go

@@ -1,22 +1,24 @@
-// Copyright (c) 2017-2020 Uber Technologies, Inc.
-//
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+
+// Portions of the Software are attributed to Copyright (c) 2020 Temporal Technologies Inc.
+
 // Permission is hereby granted, free of charge, to any person obtaining a copy
 // of this software and associated documentation files (the "Software"), to deal
 // in the Software without restriction, including without limitation the rights
 // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 // copies of the Software, and to permit persons to whom the Software is
 // furnished to do so, subject to the following conditions:
 //
-// The above copyright notice and this permission notice shall be included in
-// all copies or substantial portions of the Software.
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
 //
 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-// THE SOFTWARE.
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
 
 package matching
 
@@ -253,11 +255,14 @@ func (e *matchingEngineImpl) updateTaskList(taskList *taskListID, mgr taskListMa
 	e.taskLists[*taskList] = mgr
 }
 
-func (e *matchingEngineImpl) removeTaskListManager(id *taskListID) {
+func (e *matchingEngineImpl) removeTaskListManager(tlMgr taskListManager) {
+	id := tlMgr.TaskListID()
 	e.taskListsLock.Lock()
-	defer e.taskListsLock.Unlock()
-
-	delete(e.taskLists, *id)
+	currentTlMgr, ok := e.taskLists[*id]
+	if ok && tlMgr == currentTlMgr {
+		delete(e.taskLists, *id)
+	}
+	e.taskListsLock.Unlock()
 	e.metricsClient.Scope(metrics.MatchingTaskListMgrScope).UpdateGauge(
 		metrics.TaskListManagersGauge,
 		float64(len(e.taskLists)),
@@ -836,16 +841,17 @@ func (e *matchingEngineImpl) getTask(
 	return tlMgr.GetTask(ctx, maxDispatchPerSecond)
 }
 
-func (e *matchingEngineImpl) unloadTaskList(id *taskListID) {
+func (e *matchingEngineImpl) unloadTaskList(tlMgr taskListManager) {
+	id := tlMgr.TaskListID()
 	e.taskListsLock.Lock()
-	tlMgr, ok := e.taskLists[*id]
-	if ok {
-		delete(e.taskLists, *id)
+	currentTlMgr, ok := e.taskLists[*id]
+	if !ok || tlMgr != currentTlMgr {
+		e.taskListsLock.Unlock()
+		return
 	}
+	delete(e.taskLists, *id)
 	e.taskListsLock.Unlock()
-	if ok {
-		tlMgr.Stop()
-	}
+	tlMgr.Stop()
 }
 
 // Populate the decision task response based on context and scheduled/started events.

service/matching/matchingEngine_test.go

@@ -1,22 +1,24 @@
-// Copyright (c) 2017 Uber Technologies, Inc.
-//
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+
+// Portions of the Software are attributed to Copyright (c) 2020 Temporal Technologies Inc.
+
 // Permission is hereby granted, free of charge, to any person obtaining a copy
 // of this software and associated documentation files (the "Software"), to deal
 // in the Software without restriction, including without limitation the rights
 // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 // copies of the Software, and to permit persons to whom the Software is
 // furnished to do so, subject to the following conditions:
 //
-// The above copyright notice and this permission notice shall be included in
-// all copies or substantial portions of the Software.
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
 //
 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
-// THE SOFTWARE.
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
 
 package matching
 
@@ -175,6 +177,39 @@ func (s *matchingEngineSuite) TestPollForDecisionTasksEmptyResultWithShortContex
 	s.PollForTasksEmptyResultTest(callContext, persistence.TaskListTypeDecision)
 }
 
+func (s *matchingEngineSuite) TestOnlyUnloadMatchingInstance() {
+	taskListID := newTestTaskListID(
+		uuid.New(),
+		"makeToast",
+		persistence.TaskListTypeActivity)
+	tlKind := types.TaskListKindNormal
+	tlm, err := s.matchingEngine.getTaskListManager(taskListID, &tlKind)
+	s.Require().NoError(err)
+
+	tlm2, err := newTaskListManager(
+		s.matchingEngine,
+		taskListID, // same taskListID as above
+		&tlKind,
+		s.matchingEngine.config)
+	s.Require().NoError(err)
+
+	// try to unload a different tlm instance with the same taskListID
+	s.matchingEngine.unloadTaskList(tlm2)
+
+	got, err := s.matchingEngine.getTaskListManager(taskListID, &tlKind)
+	s.Require().NoError(err)
+	s.Require().Same(tlm, got,
+		"Unload call with non-matching taskListManager should not cause unload")
+
+	// this time unload the right tlm
+	s.matchingEngine.unloadTaskList(tlm)
+
+	got, err = s.matchingEngine.getTaskListManager(taskListID, &tlKind)
+	s.Require().NoError(err)
+	s.Require().NotSame(tlm, got,
+		"Unload call with matching incarnation should have caused unload")
+}
+
 func (s *matchingEngineSuite) TestPollForDecisionTasks() {
 	s.PollForDecisionTasksResultTest()
 }
@@ -1496,7 +1531,7 @@ func (s *matchingEngineSuite) TestTaskListManagerGetTaskBatch() {
 	s.True(0 < len(tasks) && len(tasks) <= rangeSize)
 	s.True(isReadBatchDone)
 
-	tlMgr.engine.removeTaskListManager(tlMgr.taskListID)
+	tlMgr.engine.removeTaskListManager(tlMgr)
 }
 
 func (s *matchingEngineSuite) TestTaskListManagerGetTaskBatch_ReadBatchDone() {

service/matching/taskListManager.go

@@ -83,6 +83,7 @@ type (
 		DescribeTaskList(includeTaskListStatus bool) *types.DescribeTaskListResponse
 		String() string
 		GetTaskListKind() types.TaskListKind
+		TaskListID() *taskListID
 	}
 
 	// Single task list in memory state
@@ -204,7 +205,7 @@ func (c *taskListManagerImpl) Stop() {
 	close(c.shutdownCh)
 	c.taskWriter.Stop()
 	c.taskReader.Stop()
-	c.engine.removeTaskListManager(c.taskListID)
+	c.engine.removeTaskListManager(c)
 	c.logger.Info("Task list manager state changed", tag.LifeCycleStopped)
 }
 
@@ -428,6 +429,10 @@ func (c *taskListManagerImpl) GetTaskListKind() types.TaskListKind {
 	return c.taskListKind
 }
 
+func (c *taskListManagerImpl) TaskListID() *taskListID {
+	return c.taskListID
+}
+
 // completeTask marks a task as processed. Only tasks created by taskReader (i.e. backlog from db) reach
 // here. As part of completion:
 //   - task is deleted from the database when err is nil