Fix data race on Flusher::taskId

Flusher::taskId is accessed from different threads without correct
synchronization - see ThreadSanitizer report:

  WARNING: ThreadSanitizer: data race (pid=13527)
    Write of size 8 at 0x7d4400009650 by thread T14 (mutexes: write M5523):
      #0 Flusher::step(GlobalTask*) couchbase/ep-engine/src/flusher.cc:200 (ep.so+0x0000000e78ae)
      couchbase#1 FlusherTask::run() couchbase/ep-engine/src/tasks.cc:61 (ep.so+0x0000001202e2)
      couchbase#2 ExecutorThread::run() couchbase/ep-engine/src/executorthread.cc:124 (ep.so+0x0000000e2e05)
      couchbase#3 launch_executor_thread(void*) couchbase/ep-engine/src/executorthread.cc:34 (ep.so+0x0000000e28b9)
      couchbase#4 platform_thread_wrap .ccache/tmp/cb_pthread.tmp.7e5bc917ff0e.45579.i:0 (libplatform.so.0.1.0+0x000000003891)

    Previous read of size 8 at 0x7d4400009650 by main thread:
      #0 Flusher::wait() couchbase/ep-engine/src/flusher.cc:41 (ep.so+0x0000000e66cf)
      couchbase#1 EventuallyPersistentStore::~EventuallyPersistentStore() couchbase/ep-engine/src/ep.cc:514 (ep.so+0x000000071386)
      couchbase#2 EventuallyPersistentEngine::~EventuallyPersistentEngine() couchbase/ep-engine/src/ep_engine.cc:6201 (ep.so+0x0000000bfeea)
      couchbase#3 EvpDestroy(engine_interface*, bool) couchbase/ep-engine/src/ep_engine.cc:141 (ep.so+0x0000000a0e9c)
      couchbase#4 mock_destroy(engine_interface*, bool) couchbase/memcached/programs/engine_testapp/engine_testapp.cc:98 (engine_testapp+0x0000000c4b87)
      couchbase#5 execute_test(test, char const*, char const*) couchbase/memcached/programs/engine_testapp/engine_testapp.cc:995 (engine_testapp+0x0000000c4076)
      #6 __libc_start_main /build/buildd/eglibc-2.15/csu/libc-start.c:226 (libc.so.6+0x00000002176c)

Given that taskId is a simple primitive type (size_t) fix by removing
the mutex (which wasn't acquired for all accesses) and replace taskId
with an atomic type.

Change-Id: Idc75278ed2882abd173297b77bdb72834cbe4163
Reviewed-on: http://review.couchbase.org/53220
Tested-by: buildbot <[email protected]>
Reviewed-by: Chiyoung Seo <[email protected]>

Author: daverigby

src/flusher.cc

@@ -143,14 +143,13 @@ void Flusher::start() {
     LockHolder lh(taskMutex);
     if (taskId) {
         LOG(EXTENSION_LOG_WARNING, "Double start in flusher task id %llu: %s",
-                taskId, stateName());
+                taskId.load(), stateName());
         return;
     }
     schedule_UNLOCKED();
 }
 
 void Flusher::wake(void) {
-    LockHolder lh(taskMutex);
     cb_assert(taskId > 0);
     ExecutorPool::get()->wake(taskId);
 }
@@ -196,7 +195,6 @@ bool Flusher::step(GlobalTask *task) {
             transition_state(stopped);
         case stopped:
             {
-                LockHolder lh(taskMutex);
                 taskId = 0;
                 return false;
             }

src/flusher.h

@@ -117,8 +117,11 @@ class Flusher {
 
     EventuallyPersistentStore   *store;
     AtomicValue<enum flusher_state> _state;
+
+    // Used for serializaling attempts to start the flusher from
+    // different threads.
     Mutex                        taskMutex;
-    size_t                       taskId;
+    AtomicValue<size_t>      taskId;
 
     double                   minSleepTime;
     uint16_t                 initCommitInterval;