Fix potential deadlock of exited thread delay queue.

Signed-off-by: yaofighting <[email protected]>
KindlingProject · Nov 28, 2022 · 57d1bfe · 57d1bfe
1 parent d4e8e64
commit 57d1bfe
Show file tree

Hide file tree

Showing 4 changed files with 10 additions and 19 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -18,7 +18,7 @@
 ### Bug fixes
 - 
 - 
-- 
+- Fix potential deadlock of exited thread delay queue. ([#366](https://github.com/CloudDectective-Harmonycloud/kindling/pull/366))
 - Fix the bug that cpuEvent cache size continuously increases even if trace profiling is not enabled.([#362](https://github.com/CloudDectective-Harmonycloud/kindling/pull/362))
 - Fix the bug that duplicate CPU events are indexed into Elasticsearch. ([#359](https://github.com/KindlingProject/kindling/pull/359))
 - Implement the delay queue for exited thread, so as to avoid losing the data in the period before the thread exits. ([#365](https://github.com/CloudDectective-Harmonycloud/kindling/pull/365))

diff --git a/collector/pkg/component/analyzer/cpuanalyzer/cpu_analyzer.go b/collector/pkg/component/analyzer/cpuanalyzer/cpu_analyzer.go
@@ -242,6 +242,8 @@ func (ca *CpuAnalyzer) PutEventToSegments(pid uint32, tid uint32, threadName str
 func (ca *CpuAnalyzer) trimExitedThread(pid uint32, tid uint32) {
 	ca.lock.Lock()
 	defer ca.lock.Unlock()
+	ca.tidExpiredQueue.queueMutex.Lock()
+	defer ca.tidExpiredQueue.queueMutex.Unlock()
 	tidEventsMap := ca.cpuPidEvents[pid]
 	if tidEventsMap == nil {
 		return

diff --git a/collector/pkg/component/analyzer/cpuanalyzer/delete_tid.go b/collector/pkg/component/analyzer/cpuanalyzer/delete_tid.go
@@ -40,23 +40,17 @@ func (dq *tidDeleteQueue) Pop() {
 //Add procexit tid
 func (ca *CpuAnalyzer) AddTidToDeleteCache(curTime time.Time, pid uint32, tid uint32) {
 	cacheElem := deleteTid{pid: pid, tid: tid, exitTime: curTime}
-	ca.tidExpiredQueue.queueMutex.Lock()
-	defer ca.tidExpiredQueue.queueMutex.Unlock()
 	ca.tidExpiredQueue.Push(cacheElem)
 }
 
-func (ca *CpuAnalyzer) DeleteTid(tidEventsMap map[uint32]*TimeSegments, tid uint32) {
-	ca.lock.Lock()
-	defer ca.lock.Unlock()
-	delete(tidEventsMap, tid)
-}
-
 func (ca *CpuAnalyzer) TidDelete(interval time.Duration, expiredDuration time.Duration) {
 	for {
 		select {
 		case <-time.After(interval):
 			now := time.Now()
 			func() {
+				ca.lock.Lock()
+				defer ca.lock.Unlock()
 				ca.tidExpiredQueue.queueMutex.Lock()
 				defer ca.tidExpiredQueue.queueMutex.Unlock()
 				for {
@@ -72,7 +66,7 @@ func (ca *CpuAnalyzer) TidDelete(interval time.Duration, expiredDuration time.Du
 						}
 						ca.telemetry.Logger.Debugf("Delete expired thread... pid=%d, tid=%d", elem.pid, elem.tid)
 						//fmt.Printf("Go Test: Delete expired thread... pid=%d, tid=%d\n", elem.pid, elem.tid)
-						ca.DeleteTid(tidEventsMap, elem.tid)
+						delete(tidEventsMap, elem.tid)
 						ca.tidExpiredQueue.Pop()
 					} else {
 						break

diff --git a/collector/pkg/component/analyzer/cpuanalyzer/delete_tid_test.go b/collector/pkg/component/analyzer/cpuanalyzer/delete_tid_test.go
@@ -42,21 +42,16 @@ func TestDeleteQueue(t *testing.T) {
 		//check tid which exist in queue but not in the map
 		if i%4 != 0 {
 			ca.PutEventToSegments(uint32(i), uint32(i)+5, "threadname"+strconv.Itoa(i+100), ev)
+			time.Sleep(timeDuration)
 		}
 
-		var queueLen int
-
 		func() {
+			cacheElem := deleteTid{uint32(i), uint32(i) + 5, curTime.Add(timeDuration)}
+			ca.trimExitedThread(uint32(i), uint32(i)+5)
+
 			ca.tidExpiredQueue.queueMutex.Lock()
 			defer ca.tidExpiredQueue.queueMutex.Unlock()
-			queueLen = len(ca.tidExpiredQueue.queue)
-
-			cacheElem := deleteTid{uint32(i), uint32(i) + 5, curTime.Add(timeDuration)}
-			ca.tidExpiredQueue.Push(cacheElem)
 			visit = append(visit, cacheElem)
-			if len(ca.tidExpiredQueue.queue) != queueLen+1 {
-				t.Errorf("the length of queue is not added, expection: %d but: %d\n", queueLen+1, len(ca.tidExpiredQueue.queue))
-			}
 		}()
 
 		t.Logf("pid=%d, tid=%d enter time=%s\n", uint32(i), uint32(i)+5, curTime.Format("2006-01-02 15:04:05.000"))