MB-16357: Interlock expiry and vbucket state changes

Expiry should only occur whilst the vbucket is active.
Background tasks performing expiry deletion must stop
driving deletions when the vb changes status to !active.

Using a reader/writer lock the core deleteExpiredItem
function which is used by both compaction driven expiry
and the item pager are now interlocked with vbucket::setState()

Change-Id: I19d30c3d7855778613ccb4534a042c0daf627b8c
Reviewed-on: http://review.couchbase.org/55646
Reviewed-by: Chiyoung Seo <chiyoung@couchbase.com>
Tested-by: buildbot <build@couchbase.com>

Author: jimwwalker

src/ep.cc

@@ -484,34 +484,39 @@ EventuallyPersistentStore::deleteExpiredItem(uint16_t vbid, std::string &key,
                                              uint64_t revSeqno) {
     RCPtr<VBucket> vb = getVBucket(vbid);
     if (vb) {
-        int bucket_num(0);
-        incExpirationStat(vb);
-        LockHolder lh = vb->ht.getLockedBucket(key, &bucket_num);
-        StoredValue *v = vb->ht.unlocked_find(key, bucket_num, true, false);
-        if (v) {
-            if (v->isTempNonExistentItem() || v->isTempDeletedItem()) {
-                // This is a temporary item whose background fetch for metadata
-                // has completed.
-                bool deleted = vb->ht.unlocked_del(key, bucket_num);
-                cb_assert(deleted);
-            } else if (v->isExpired(startTime) && !v->isDeleted()) {
-                vb->ht.unlocked_softDelete(v, 0, getItemEvictionPolicy());
-                queueDirty(vb, v, &lh, false);
-            }
-        } else {
-            if (eviction_policy == FULL_EVICTION) {
-                // Create a temp item and delete and push it
-                // into the checkpoint queue.
-                add_type_t rv = vb->ht.unlocked_addTempItem(bucket_num, key,
-                                                            eviction_policy);
-                if (rv == ADD_NOMEM) {
-                    return;
+        // Obtain reader access to the VB state change lock so that
+        // the VB can't switch state whilst we're processing
+        ReaderLockHolder(vb->getStateLock());
+        if (vb->getState() == vbucket_state_active) {
+            int bucket_num(0);
+            incExpirationStat(vb);
+            LockHolder lh = vb->ht.getLockedBucket(key, &bucket_num);
+            StoredValue *v = vb->ht.unlocked_find(key, bucket_num, true, false);
+            if (v) {
+                if (v->isTempNonExistentItem() || v->isTempDeletedItem()) {
+                    // This is a temporary item whose background fetch for metadata
+                    // has completed.
+                    bool deleted = vb->ht.unlocked_del(key, bucket_num);
+                    cb_assert(deleted);
+                } else if (v->isExpired(startTime) && !v->isDeleted()) {
+                    vb->ht.unlocked_softDelete(v, 0, getItemEvictionPolicy());
+                    queueDirty(vb, v, &lh, false);
+                }
+            } else {
+                if (eviction_policy == FULL_EVICTION) {
+                    // Create a temp item and delete and push it
+                    // into the checkpoint queue.
+                    add_type_t rv = vb->ht.unlocked_addTempItem(bucket_num, key,
+                                                                eviction_policy);
+                    if (rv == ADD_NOMEM) {
+                        return;
+                    }
+                    v = vb->ht.unlocked_find(key, bucket_num, true, false);
+                    v->setStoredValueState(StoredValue::state_deleted_key);
+                    v->setRevSeqno(revSeqno);
+                    vb->ht.unlocked_softDelete(v, 0, eviction_policy);
+                    queueDirty(vb, v, &lh, false);
                 }
-                v = vb->ht.unlocked_find(key, bucket_num, true, false);
-                v->setStoredValueState(StoredValue::state_deleted_key);
-                v->setRevSeqno(revSeqno);
-                vb->ht.unlocked_softDelete(v, 0, eviction_policy);
-                queueDirty(vb, v, &lh, false);
             }
         }
     }

src/locks.h

@@ -28,6 +28,7 @@
 #include "common.h"
 #include "mutex.h"
 #include "syncobject.h"
+#include "rwlock.h"
 
 /**
  * RAII lock holder to guarantee release of the lock.
@@ -158,4 +159,64 @@ class MultiLockHolder {
     DISALLOW_COPY_AND_ASSIGN(MultiLockHolder);
 };
 
+// RAII Reader lock
+class ReaderLockHolder {
+public:
+    ReaderLockHolder(RWLock& lock)
+      :  rwLock(lock) {
+        int locked = rwLock.readerLock();
+        if (locked != 0) {
+            char exceptionMsg[64];
+            snprintf(exceptionMsg, sizeof(exceptionMsg),
+                     "%d returned by readerLock()", locked);
+            throw std::runtime_error(exceptionMsg);
+        }
+    }
+
+    ~ReaderLockHolder() {
+        int unlocked = rwLock.readerUnlock();
+        if (unlocked != 0) {
+            char exceptionMsg[64];
+            snprintf(exceptionMsg, sizeof(exceptionMsg),
+                     "%d returned by readerUnlock()", unlocked);
+            throw std::runtime_error(exceptionMsg);
+        }
+    }
+
+private:
+    RWLock& rwLock;
+
+    DISALLOW_COPY_AND_ASSIGN(ReaderLockHolder);
+};
+
+// RAII Writer lock
+class WriterLockHolder {
+public:
+    WriterLockHolder(RWLock& lock)
+      :  rwLock(lock) {
+        int locked = rwLock.writerLock();
+        if (locked != 0) {
+            char exceptionMsg[64];
+            snprintf(exceptionMsg, sizeof(exceptionMsg),
+                     "%d returned by writerLock()", locked);
+            throw std::runtime_error(exceptionMsg);
+        }
+    }
+
+    ~WriterLockHolder() {
+        int unlocked = rwLock.writerUnlock();
+        if (unlocked != 0) {
+            char exceptionMsg[64];
+            snprintf(exceptionMsg, sizeof(exceptionMsg),
+                     "%d returned by writerUnlock()", unlocked);
+            throw std::runtime_error(exceptionMsg);
+        }
+    }
+
+private:
+    RWLock& rwLock;
+
+    DISALLOW_COPY_AND_ASSIGN(WriterLockHolder);
+};
+
 #endif  // SRC_LOCKS_H_

src/rwlock.h

@@ -0,0 +1,57 @@
+/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+/*
+ *     Copyright 2015 Couchbase, Inc
+ *
+ *   Licensed under the Apache License, Version 2.0 (the "License");
+ *   you may not use this file except in compliance with the License.
+ *   You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *   Unless required by applicable law or agreed to in writing, software
+ *   distributed under the License is distributed on an "AS IS" BASIS,
+ *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *   See the License for the specific language governing permissions and
+ *   limitations under the License.
+ */
+#pragma once
+
+#include "platform/platform.h"
+#include "common.h"
+
+/**
+ *   Reader/Write lock abstraction for platform provided cb_rw_lock
+ *
+ *   The lock allows many readers but mutual exclusion with a writer.
+ */
+class RWLock {
+public:
+    RWLock() {
+        cb_rw_lock_initialize(&lock);
+    }
+
+    ~RWLock() {
+        cb_rw_lock_destroy(&lock);
+    }
+
+    int readerLock() {
+        return cb_rw_reader_enter(&lock);
+    }
+
+    int readerUnlock() {
+        return cb_rw_reader_exit(&lock);
+    }
+
+    int writerLock() {
+        return cb_rw_writer_enter(&lock);
+    }
+
+    int writerUnlock() {
+        return cb_rw_writer_exit(&lock);
+    }
+
+private:
+    cb_rwlock_t lock;
+
+    DISALLOW_COPY_AND_ASSIGN(RWLock);
+};

src/vbucket.cc

@@ -183,6 +183,7 @@ void VBucket::fireAllOps(EventuallyPersistentEngine &engine) {
 
 void VBucket::setState(vbucket_state_t to, SERVER_HANDLE_V1 *sapi) {
     cb_assert(sapi);
+    WriterLockHolder wlh(stateLock);
     vbucket_state_t oldstate(state);
 
     if (to == vbucket_state_active &&

src/vbucket.h

@@ -242,6 +242,7 @@ class VBucket : public RCValue {
     int getId(void) const { return id; }
     vbucket_state_t getState(void) const { return state; }
     void setState(vbucket_state_t to, SERVER_HANDLE_V1 *sapi);
+    RWLock& getStateLock() {return stateLock;}
 
     vbucket_state_t getInitialState(void) { return initialState; }
     void setInitialState(vbucket_state_t initState) {
@@ -436,6 +437,7 @@ class VBucket : public RCValue {
 
     int                      id;
     AtomicValue<vbucket_state_t>  state;
+    RWLock                   stateLock;
     vbucket_state_t          initialState;
     Mutex                    pendingOpLock;
     std::vector<const void*> pendingOps;

tests/ep_testsuite.cc

@@ -11326,6 +11326,109 @@ static enum test_result test_failover_log_dcp(ENGINE_HANDLE *h,
     return SUCCESS;
 }
 
+struct mb16357_ctx {
+    mb16357_ctx(ENGINE_HANDLE *_h, ENGINE_HANDLE_V1 *_h1, int _items) :
+        h(_h), h1(_h1), items(_items) { }
+
+    ENGINE_HANDLE *h;
+    ENGINE_HANDLE_V1 *h1;
+    int items;
+};
+
+extern "C" {
+    static void dcp_thread_func(void *args) {
+        struct mb16357_ctx *ctx = static_cast<mb16357_ctx *>(args);
+
+        const void *cookie = testHarness.create_cookie();
+        uint32_t opaque = 0xFFFF0000;
+        uint32_t flags = 0;
+        std::string name = "unittest";
+
+        while (get_int_stat(ctx->h, ctx->h1, "ep_pending_compactions") == 0);
+
+        // Switch to replica
+        check(set_vbucket_state(ctx->h, ctx->h1, 0, vbucket_state_replica),
+                "Failed to set vbucket state.");
+
+        // Open consumer connection
+        checkeq(ctx->h1->dcp.open(ctx->h, cookie, opaque, 0, flags,
+                                  (void*)name.c_str(), name.length()),
+                ENGINE_SUCCESS,
+                "Failed dcp Consumer open connection.");
+
+        add_stream_for_consumer(ctx->h, ctx->h1, cookie, opaque++, 0, 0,
+                PROTOCOL_BINARY_RESPONSE_SUCCESS);
+
+
+        uint32_t stream_opaque = get_int_stat(ctx->h, ctx->h1,
+                                              "eq_dcpq:unittest:stream_0_opaque",
+                                              "dcp");
+
+
+        for (int i = 1; i <= ctx->items; i++) {
+            std::stringstream ss;
+            ss << "kamakeey-" << i;
+
+            // send mutations in single mutation snapshots to race more with compaction
+            checkeq(ctx->h1->dcp.snapshot_marker(ctx->h, cookie,
+                                                 stream_opaque, 0/*vbid*/,
+                                                 ctx->items, ctx->items + i, 2),
+                    ENGINE_SUCCESS,
+                    "Failed to send snapshot marker");
+            checkeq(ctx->h1->dcp.mutation(ctx->h, cookie, stream_opaque,
+                                          ss.str().c_str(), ss.str().length(),
+                                          "value", 5, i * 3, 0, 0, 0,
+                                          i + ctx->items, i + ctx->items,
+                                          0, 0, "", 0, INITIAL_NRU_VALUE),
+                    ENGINE_SUCCESS,
+                    "Failed to send dcp mutation");
+        }
+
+        testHarness.destroy_cookie(cookie);
+    }
+
+    static void compact_thread_func(void *args) {
+        struct mb16357_ctx *ctx = static_cast<mb16357_ctx *>(args);
+        compact_db(ctx->h, ctx->h1, 0, 99, ctx->items, 1);
+    }
+}
+
+static enum test_result test_mb16357(ENGINE_HANDLE *h,
+                                     ENGINE_HANDLE_V1 *h1) {
+
+    // Load up vb0 with n items, expire in 1 second
+    int num_items = 1000;
+
+    for (int j = 0; j < num_items; ++j) {
+        item *i = NULL;
+        std::stringstream ss;
+        ss << "key-" << j;
+        check(store(h, h1, NULL, OPERATION_SET,
+                    ss.str().c_str(), "data", &i, 0, 0, 1/*expire*/, 0)
+                    == ENGINE_SUCCESS, "Failed to store a value"); //expire in 1 second
+
+        h1->release(h, NULL, i);
+    }
+
+    wait_for_flusher_to_settle(h, h1);
+    testHarness.time_travel(3617); // force expiry pushing time forward.
+
+    struct mb16357_ctx ctx(h, h1, num_items);
+    cb_thread_t cp_thread, dcp_thread;
+
+    cb_assert(cb_create_thread(&cp_thread,
+                               compact_thread_func,
+                               &ctx, 0) == 0);
+    cb_assert(cb_create_thread(&dcp_thread,
+                               dcp_thread_func,
+                               &ctx, 0) == 0);
+
+    cb_assert(cb_join_thread(cp_thread) == 0);
+    cb_assert(cb_join_thread(dcp_thread) == 0);
+
+    return SUCCESS;
+}
+
 static enum test_result prepare(engine_test_t *test) {
 #ifdef __sun
         // Some of the tests doesn't work on Solaris.. Don't know why yet..
@@ -12251,7 +12354,9 @@ engine_test_t* get_tests(void) {
 
         TestCase("test failover log behavior", test_failover_log_behavior,
                  test_setup, teardown, NULL, prepare, cleanup),
-
+        TestCase("test MB-16357", test_mb16357,
+                 test_setup, teardown, "compaction_exp_mem_threshold=85",
+                 prepare, cleanup),
         TestCase(NULL, NULL, NULL, NULL, NULL, prepare, cleanup)
     };