Fix potential deadlock of exited thread delay queue.

CHANGELOG.md

@@ -18,7 +18,7 @@
 ### Bug fixes
 - 
 - 
-- 
+- Fix potential deadlock of exited thread delay queue. ([#373](https://github.com/CloudDectective-Harmonycloud/kindling/pull/373))
 - 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))

collector/pkg/component/analyzer/cpuanalyzer/cpu_analyzer.go

@@ -240,12 +240,10 @@ 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()
-	tidEventsMap := ca.cpuPidEvents[pid]
-	if tidEventsMap == nil {
-		return
-	}
+	ca.tidExpiredQueue.queueMutex.Lock()
+	defer ca.tidExpiredQueue.queueMutex.Unlock()
 	ca.telemetry.Logger.Debugf("Receive a procexit pid=%d, tid=%d, which will be deleted from map after 10 seconds. ", pid, tid)
-	ca.AddTidToDeleteCache(time.Now(), pid, tid)
+
+	cacheElem := deleteTid{pid: pid, tid: tid, exitTime: time.Now()}
+	ca.tidExpiredQueue.Push(cacheElem)
 }

collector/pkg/component/analyzer/cpuanalyzer/delete_tid.go

@@ -37,20 +37,6 @@ 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 {
@@ -64,19 +50,23 @@ func (ca *CpuAnalyzer) TidDelete(interval time.Duration, expiredDuration time.Du
 					if elem == nil {
 						break
 					}
-					if elem.exitTime.Add(expiredDuration).Before(now) {
+					if elem.exitTime.Add(expiredDuration).After(now) {
+						break
+					}
+					//Delete expired threads (current_time >= thread_exit_time + interval_time).
+					func() {
+						ca.lock.Lock()
+						defer ca.lock.Unlock()
 						tidEventsMap := ca.cpuPidEvents[elem.pid]
 						if tidEventsMap == nil {
 							ca.tidExpiredQueue.Pop()
-							continue
+						} else {
+							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)
+							delete(tidEventsMap, elem.tid)
+							ca.tidExpiredQueue.Pop()
 						}
-						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)
-						ca.tidExpiredQueue.Pop()
-					} else {
-						break
-					}
+					}()
 				}
 			}()
 		}

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"))