From efa43483b209e79eb6a743e958c7c573ce56eebf Mon Sep 17 00:00:00 2001
From: Kevin Wan <wanjunfeng@gmail.com>
Date: Sun, 3 Jan 2021 20:56:17 +0800
Subject: [PATCH] fix potential data race in PeriodicalExecutor (#344)

* fix potential data race in PeriodicalExecutor

* add comment
---
 core/executors/periodicalexecutor.go | 21 ++++++++++++++-------
 1 file changed, 14 insertions(+), 7 deletions(-)

diff --git a/core/executors/periodicalexecutor.go b/core/executors/periodicalexecutor.go
index 2a2990c8..f797dfc2 100644
--- a/core/executors/periodicalexecutor.go
+++ b/core/executors/periodicalexecutor.go
@@ -134,9 +134,6 @@ func (pe *PeriodicalExecutor) backgroundFlush() {
 				} else if pe.Flush() {
 					last = timex.Now()
 				} else if pe.shallQuit(last) {
-					pe.lock.Lock()
-					pe.guarded = false
-					pe.lock.Unlock()
 					return
 				}
 			}
@@ -180,8 +177,18 @@ func (pe *PeriodicalExecutor) hasTasks(tasks interface{}) bool {
 	}
 }
 
-func (pe *PeriodicalExecutor) shallQuit(last time.Duration) bool {
-	idleEnough := timex.Since(last) > pe.interval*idleRound
-	noPending := atomic.LoadInt32(&pe.inflight) == 0
-	return idleEnough && noPending
+func (pe *PeriodicalExecutor) shallQuit(last time.Duration) (stop bool) {
+	if timex.Since(last) <= pe.interval*idleRound {
+		return
+	}
+
+	// checking pe.inflight and setting pe.guarded should be locked together
+	pe.lock.Lock()
+	if atomic.LoadInt32(&pe.inflight) == 0 {
+		pe.guarded = false
+		stop = true
+	}
+	pe.lock.Unlock()
+
+	return
 }