Introduce LockedVBucketPtr; fix data race on VBucket::purge_seqno

As reported by TSan there is a race in one of the dcp_tests on
VBucket::purge_seqno (see below). This is due to it being modified in
dcp_test without first acquiring the VBucket lock.

We don't want to directly expose the VBucket mutexes to outside
clients, so instead introduce a RAII wrapper for a VBucket pointer and
a lock on the appropriate vb_mutex - LockedVBucketPtr.

Make use of that to fix the race.

WARNING: ThreadSanitizer: data race (pid=9519)
  Read of size 8 at 0x7d6c0000f8f0 by thread T20 (mutexes: write M344682):
    #0 Monotonic<unsigned long, DefaultOrderReversedPolicy, std::greater_equal>::operator unsigned long() const kv_engine/engines/ep/src/monotonic.h:100:16 (ep-engine_ep_unit_tests+0x0000009c7f3c)
    couchbase#1 VBucket::getPurgeSeqno() const kv_engine/engines/ep/src/vbucket.h:164 (ep-engine_ep_unit_tests+0x0000009c7f3c)
    couchbase#2 VBucket::getVBucketState() const kv_engine/engines/ep/src/vbucket.cc:325 (ep-engine_ep_unit_tests+0x0000009c7f3c)
    couchbase#3 KVBucket::flushVBucket(unsigned short) kv_engine/engines/ep/src/kv_bucket.cc:1866:32 (ep-engine_ep_unit_tests+0x00000099386a)
    couchbase#4 Flusher::flushVB() kv_engine/engines/ep/src/flusher.cc:281:20 (ep-engine_ep_unit_tests+0x00000096e17a)
    couchbase#5 Flusher::step(GlobalTask*) kv_engine/engines/ep/src/flusher.cc:190:9 (ep-engine_ep_unit_tests+0x00000096d0c9)

  Previous write of size 8 at 0x7d6c0000f8f0 by main thread:
    #0 Monotonic<unsigned long, DefaultOrderReversedPolicy, std::greater_equal>::operator=(unsigned long const&) kv_engine/engines/ep/src/monotonic.h:92:17 (ep-engine_ep_unit_tests+0x00000060211d)
    couchbase#1 VBucket::setPurgeSeqno(unsigned long) kv_engine/engines/ep/src/vbucket.h:168 (ep-engine_ep_unit_tests+0x00000060211d)
    couchbase#2 StreamTest_RollbackDueToPurge_Test::TestBody() kv_engine/engines/ep/tests/module_tests/dcp_test.cc:832 (ep-engine_ep_unit_tests+0x00000060211d)

Change-Id: If9c87d0d9fa828e274084b8ec967b1b8690bf665
Reviewed-on: http://review.couchbase.org/82733
Reviewed-by: Manu Dhundi <manu@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Trond Norbye <trond.norbye@gmail.com>

Author: daverigby

engines/ep/src/kv_bucket.h

@@ -328,6 +328,18 @@ class KVBucket : public KVBucketIface {
         return vbMap.getBucket(vbid);
     }
 
+    /**
+     * Return a pointer to the given VBucket, acquiring the appropriate VB
+     * mutex lock at the same time.
+     * @param vbid VBucket ID to get.
+     * @return A RAII-style handle which owns the correct VBucket mutex,
+     *         alongside a shared pointer to the requested VBucket.
+     */
+    LockedVBucketPtr getLockedVBucket(uint16_t vbid) {
+        std::unique_lock<std::mutex> lock(vb_mutexes[vbid]);
+        return {vbMap.getBucket(vbid), std::move(lock)};
+    }
+
     std::pair<uint64_t, bool> getLastPersistedCheckpointId(uint16_t vb) {
         // No persistence at the KVBucket class level.
         return {0, false};

engines/ep/src/vbucket.h

@@ -1680,3 +1680,24 @@ class VBucket : public std::enable_shared_from_this<VBucket> {
 };
 
 using VBucketPtr = std::shared_ptr<VBucket>;
+
+/**
+ * Represents a locked VBucket that provides RAII semantics for the lock.
+ *
+ * Behaves like the underlying shared_ptr - i.e. `operator->` is overloaded
+ * to return the underlying VBucket.
+ */
+class LockedVBucketPtr {
+public:
+    LockedVBucketPtr(VBucketPtr vb, std::unique_lock<std::mutex>&& lock)
+        : vb(std::move(vb)), lock(std::move(lock)) {
+    }
+
+    VBucket* operator->() const {
+        return vb.get();
+    }
+
+private:
+    VBucketPtr vb;
+    std::unique_lock<std::mutex> lock;
+};

engines/ep/tests/module_tests/dcp_test.cc

@@ -829,7 +829,7 @@ TEST_P(StreamTest, RollbackDueToPurge) {
               producer->closeStream(/*opaque*/ 0, vb0->getId()));
 
     /* Set a purge_seqno > start_seqno */
-    vb0->setPurgeSeqno(numItems - 1);
+    engine->getKVBucket()->getLockedVBucket(vbid)->setPurgeSeqno(numItems - 1);
 
     /* Now we expect a rollback to 0 */
     EXPECT_EQ(ENGINE_ROLLBACK,