MB-19226: Address potential data races in the warmup code

Context: warmupState, estimatedWarmupCount,
warmup, metadata

As reported by ThreadSanitizer:

WARNING: ThreadSanitizer: data race (pid=7023)
  Write of size 8 at 0x7d240000d5e0 by thread T7:
    #0 Warmup::checkForAccessLog() /home/daver/repos/couchbase/server/ep-engine/src/warmup.cc:590 (ep.so+0x0000002d1ffc)
    #1 WarmupCheckforAccessLog::run() /home/daver/repos/couchbase/server/ep-engine/src/warmup.h:303 (ep.so+0x0000002e2bfb)
    #2 ExecutorThread::run() /home/daver/repos/couchbase/server/ep-engine/src/executorthread.cc:106 (ep.so+0x0000001e34f9)
    #3 launch_executor_thread(void*) /home/daver/repos/couchbase/server/ep-engine/src/executorthread.cc:34 (ep.so+0x0000001e2b7a)
    #4 platform_thread_wrap /home/daver/repos/couchbase/server/platform/src/cb_pthreads.c:19 (libplatform.so.0.1.0+0x0000000035dc)

Previous read of size 8 at 0x7d240000d5e0 by main thread (mutexes: write M699180732592992152):
  #0 Warmup::addStats(void (*)(char const*, unsigned short, char const*, unsigned int, void const*), void const*) const /home/daver/repos/couchbase/server/ep-engine/src/warmup.cc:893 (ep.so+0x0000002d6086)
  #1 EventuallyPersistentEngine::getStats(void const*, char const*, int, void (*)(char const*, unsigned short, char const*, unsigned int, void const*)) /home/daver/repos/couchbase/server/ep-engine/src/ep_engine.cc:4422 (ep.so+0x000000151813)
  #2 EvpGetStats(engine_interface*, void const*, char const*, int, void (*)(char const*, unsigned short, char const*, unsigned int, void const*)) /home/daver/repos/couchbase/server/ep-engine/src/ep_engine.cc:214 (ep.so+0x0000001367b2)
  #3 mock_get_stats /home/daver/repos/couchbase/server/memcached/programs/engine_testapp/engine_testapp.c:194 (engine_testapp+0x0000004bde63)
  #4 wait_for_warmup_complete(engine_interface*, engine_interface_v1*) /home/daver/repos/couchbase/server/ep-engine/tests/ep_test_apis.cc:898 (ep_testsuite.so+0x0000000ead95)
  #5 test_setup(engine_interface*, engine_interface_v1*) /home/daver/repos/couchbase/server/ep-engine/tests/ep_testsuite.cc:168 (ep_testsuite.so+0x0000000237d3)
  #6 execute_test /home/daver/repos/couchbase/server/memcached/programs/engine_testapp/engine_testapp.c:1037 (engine_testapp+0x0000004ba82b)
  #7 main /home/daver/repos/couchbase/server/memcached/programs/engine_testapp/engine_testapp.c:1296 (engine_testapp+0x0000004b8861)

Change-Id: If96933b3b8b0aa1ed75073a0d8d629f138da081f
Reviewed-on: http://review.couchbase.org/62916
Well-Formed: buildbot <build@couchbase.com>
Reviewed-by: Will Gardner <will.gardner@couchbase.com>
Tested-by: buildbot <build@couchbase.com>

diff --git a/src/warmup.cc b/src/warmup.cc
index 3502e6a..beb2655 100644 (file)
--- a/src/warmup.cc
+++ b/src/warmup.cc
@@ -132,7 +132,7 @@ const int WarmupState::LoadingData = 7;
 const int WarmupState::Done = 8;
 
 const char *WarmupState::toString(void) const {
-    return getStateDescription(state);
+    return getStateDescription(state.load());
 }
 
 const char *WarmupState::getStateDescription(int st) const {
@@ -164,10 +164,10 @@ void WarmupState::transition(int to, bool allowAnystate) {
     if (allowAnystate || legalTransition(to)) {
         std::stringstream ss;
         ss << "Warmup transition from state \""
-           << getStateDescription(state) << "\" to \""
+           << getStateDescription(state.load()) << "\" to \""
            << getStateDescription(to) << "\"";
         LOG(EXTENSION_LOG_DEBUG, "%s", ss.str().c_str());
-        state = to;
+        state.store(to);
     } else {
         // Throw an exception to make it possible to test the logic ;)
         std::stringstream ss;
@@ -177,7 +177,7 @@ void WarmupState::transition(int to, bool allowAnystate) {
 }
 
 bool WarmupState::legalTransition(int to) const {
-    switch (state) {
+    switch (state.load()) {
     case Initialize:
         return (to == CreateVBuckets);
     case CreateVBuckets:
@@ -384,12 +384,12 @@ Warmup::~Warmup() {
 
 void Warmup::setEstimatedWarmupCount(size_t to)
 {
-    estimatedWarmupCount = to;
+    estimatedWarmupCount.store(to);
 }
 
 size_t Warmup::getEstimatedItemCount()
 {
-    return estimatedItemCount;
+    return estimatedItemCount.load();
 }
 
 void Warmup::start(void)
@@ -587,9 +587,9 @@ void Warmup::scheduleCheckForAccessLog()
 
 void Warmup::checkForAccessLog()
 {
-    metadata = gethrtime() - startTime;
+    metadata.store(gethrtime() - startTime);
     LOG(EXTENSION_LOG_WARNING, "metadata loaded in %s",
-        hrtime2text(metadata).c_str());
+        hrtime2text(metadata.load()).c_str());
 
     if (store->maybeEnableTraffic()) {
         transition(WarmupState::Done);
@@ -629,7 +629,6 @@ void Warmup::scheduleLoadingAccessLog()
 
 void Warmup::loadingAccessLog(uint16_t shardId)
 {
-
     LoadStorageKVPairCallback *load_cb =
         new LoadStorageKVPairCallback(store, true, state.getState());
     bool success = false;
@@ -799,7 +798,7 @@ void Warmup::done()
         setWarmupTime();
         store->warmupCompleted();
         LOG(EXTENSION_LOG_WARNING, "warmup completed in %s",
-                                   hrtime2text(warmup).c_str());
+                                   hrtime2text(warmup.load()).c_str());
     }
 }
 
@@ -890,32 +889,36 @@ void Warmup::addStats(ADD_STAT add_stat, const void *c) const
         addStat("min_item_threshold",
                 stats.warmupNumReadCap * 100.0, add_stat, c);
 
-        if (metadata > 0) {
-            addStat("keys_time", metadata / 1000, add_stat, c);
+        hrtime_t md_time = metadata.load();
+        if (md_time > 0) {
+            addStat("keys_time", md_time / 1000, add_stat, c);
         }
 
-        if (warmup > 0) {
-            addStat("time", warmup / 1000, add_stat, c);
+        hrtime_t w_time = warmup.load();
+        if (w_time > 0) {
+            addStat("time", w_time / 1000, add_stat, c);
         }
 
-        if (estimatedItemCount == std::numeric_limits<size_t>::max()) {
+        size_t itemCount = estimatedItemCount.load();
+        if (itemCount == std::numeric_limits<size_t>::max()) {
             addStat("estimated_key_count", "unknown", add_stat, c);
         } else {
-            if (estimateTime != 0) {
-                addStat("estimate_time", estimateTime / 1000, add_stat, c);
+            hrtime_t e_time = estimateTime.load();
+            if (e_time != 0) {
+                addStat("estimate_time", e_time / 1000, add_stat, c);
             }
-            addStat("estimated_key_count", estimatedItemCount, add_stat, c);
+            addStat("estimated_key_count", itemCount, add_stat, c);
         }
 
         if (corruptAccessLog) {
             addStat("access_log", "corrupt", add_stat, c);
         }
 
-        if (estimatedWarmupCount ==  std::numeric_limits<size_t>::max()) {
+        size_t warmupCount = estimatedWarmupCount.load();
+        if (warmupCount ==  std::numeric_limits<size_t>::max()) {
             addStat("estimated_value_count", "unknown", add_stat, c);
         } else {
-            addStat("estimated_value_count", estimatedWarmupCount,
-            add_stat, c);
+            addStat("estimated_value_count", warmupCount, add_stat, c);
         }
    } else {
         addStat(NULL, "disabled", add_stat, c);

diff --git a/src/warmup.h b/src/warmup.h
index 6ef7b93..1982f98 100644 (file)
--- a/src/warmup.h
+++ b/src/warmup.h
@@ -47,7 +47,7 @@ public:
     int getState(void) const { return state; }
 
 private:
-    int state;
+    AtomicValue<int> state;
     const char *getStateDescription(int val) const;
     bool legalTransition(int to) const;
     friend std::ostream& operator<< (std::ostream& out,
@@ -119,8 +119,6 @@ public:
     void start(void);
     void stop(void);
 
-    const WarmupState &getState(void) const { return state; }
-
     void setEstimatedWarmupCount(size_t num);
 
     size_t getEstimatedItemCount();
@@ -130,7 +128,7 @@ public:
     hrtime_t getTime(void) { return warmup; }
 
     void setWarmupTime(void) {
-        warmup = gethrtime() - startTime;
+        warmup.store(gethrtime() - startTime);
     }
 
     size_t doWarmup(MutationLog &lf, const std::map<uint16_t,
@@ -174,11 +172,12 @@ private:
     void transition(int to, bool force=false);
 
     WarmupState state;
+
     EventuallyPersistentStore *store;
     size_t taskId;
     hrtime_t startTime;
-    hrtime_t metadata;
-    hrtime_t warmup;
+    AtomicValue<hrtime_t> metadata;
+    AtomicValue<hrtime_t> warmup;
 
     std::vector<vbucket_state *> allVbStates;
     std::map<uint16_t, vbucket_state> *shardVbStates;
@@ -191,7 +190,7 @@ private:
     bool cleanShutdown;
     bool corruptAccessLog;
     AtomicValue<bool> warmupComplete;
-    size_t estimatedWarmupCount;
+    AtomicValue<size_t> estimatedWarmupCount;
 
     DISALLOW_COPY_AND_ASSIGN(Warmup);
 };