EwbEngine: Allow arbitrary statuses for Sequence mode

Expand the Sequence mode to allow an arbitrary sequence of status
codes to be injected - previously only the original engine status code
(bit cleared) or a single status code (bit set) could be injected.

This allows more advanced error injection to be performed by allowing
different injected codes for the same setup of ewouldblock_engine.

Additionally, for Sequence mode enhance EWOULDBLOCK injection - the
actual result of the IO operation can now be specified (as the status
code following EWOULDBLOCK in the sequence) instead of always
returning SUCCESS.

Change-Id: I61968fb186108d42130c91f43038e7dfc83a3134
Reviewed-on: http://review.couchbase.org/109910
Tested-by: Build Bot <[email protected]>
Reviewed-by: Ben Huddleston <[email protected]>

Author: daverigby

engines/ewouldblock_engine/ewouldblock_engine.cc

@@ -270,18 +270,23 @@ class EWB_Engine : public EngineIface, public DcpIface {
         }
 
         const bool inject = iter->second.second->should_inject_error(cmd, err);
-        const bool add_to_pending_io_ops = iter->second.second->add_to_pending_io_ops();
 
         if (inject) {
             LOG_DEBUG("EWB_Engine: injecting error:{} for cmd:{}",
                       err,
                       to_string(cmd));
 
-            if (err == ENGINE_EWOULDBLOCK && add_to_pending_io_ops) {
-                // The server expects that if EWOULDBLOCK is returned then the
-                // server should be notified in the future when the operation is
-                // ready - so add this op to the pending IO queue.
-                schedule_notification(iter->second.first);
+            if (err == ENGINE_EWOULDBLOCK) {
+                const auto add_to_pending_io_ops =
+                        iter->second.second->add_to_pending_io_ops();
+                if (add_to_pending_io_ops) {
+                    // The server expects that if EWOULDBLOCK is returned then
+                    // the server should be notified in the future when the
+                    // operation is ready - so add this op to the pending IO
+                    // queue.
+                    schedule_notification(iter->second.first,
+                                          *add_to_pending_io_ops);
+                }
             }
         }
 
@@ -582,9 +587,19 @@ class EWB_Engine : public EngineIface, public DcpIface {
                     new_mode = std::make_shared<ErrOnFirst>(injected_error);
                     break;
 
-                case EWBEngineMode::Sequence:
-                    new_mode = std::make_shared<ErrSequence>(injected_error, value);
+                case EWBEngineMode::Sequence: {
+                    std::vector<cb::engine_errc> decoded;
+                    for (unsigned int ii = 0;
+                         ii < key.size() / sizeof(cb::engine_errc);
+                         ii++) {
+                        auto status = *reinterpret_cast<const uint32_t*>(
+                                key.data() + (ii * sizeof(cb::engine_errc)));
+                        status = ntohl(status);
+                        decoded.emplace_back(cb::engine_errc(status));
+                    }
+                    new_mode = std::make_shared<ErrSequence>(decoded);
                     break;
+                }
 
                 case EWBEngineMode::No_Notify:
                     new_mode = std::make_shared<ErrOnNoNotify>(injected_error);
@@ -1023,7 +1038,11 @@ class EWB_Engine : public EngineIface, public DcpIface {
     // thread processing pending io ops.
     std::mutex mutex;
     std::condition_variable condvar;
-    std::queue<const void*> pending_io_ops;
+    struct PendingIO {
+        const void* cookie;
+        ENGINE_ERROR_CODE status;
+    };
+    std::queue<PendingIO> pending_io_ops;
 
     std::atomic<bool> stop_notification_thread;
 
@@ -1034,9 +1053,14 @@ class EWB_Engine : public EngineIface, public DcpIface {
         FaultInjectMode(ENGINE_ERROR_CODE injected_error_)
           : injected_error(injected_error_) {}
 
-        virtual bool add_to_pending_io_ops() {
-            return true;
+        // In the event of injecting an EWOULDBLOCK error, should the connection
+        // be added to the pending_io_ops (and subsequently notified)?
+        // @returns empty if shouldn't be added, otherwise contains the
+        // status code to notify with.
+        virtual boost::optional<ENGINE_ERROR_CODE> add_to_pending_io_ops() {
+            return ENGINE_SUCCESS;
         }
+
         virtual bool should_inject_error(Cmd cmd, ENGINE_ERROR_CODE& err) = 0;
 
         virtual std::string to_string() const = 0;
@@ -1130,36 +1154,88 @@ class EWB_Engine : public EngineIface, public DcpIface {
         uint32_t percentage_to_err;
     };
 
+    /**
+     * Injects a sequence of error codes for each call to should_inject_error().
+     * If the end of the given sequence is reached, then throws
+     * std::logic_error.
+     *
+     * cb::mcbp::Status::ReservedUserStart can be used to specify that the
+     * no error is injected (the original status code is returned unchanged).
+     */
     class ErrSequence : public FaultInjectMode {
     public:
+        /**
+         * Construct with a sequence of the specified error, or the 'normal'
+         * status code.
+         */
         ErrSequence(ENGINE_ERROR_CODE injected_error_, uint32_t sequence_)
-            : FaultInjectMode(injected_error_),
+            : FaultInjectMode(injected_error_) {
+            for (int ii = 0; ii < 32; ii++) {
+                if ((sequence_ & (1 << ii)) != 0) {
+                    sequence.push_back(cb::engine_errc(injected_error_));
+                } else {
+                    sequence.push_back(cb::engine_errc(-1));
+                }
+            }
+            pos = sequence.begin();
+        }
+
+        /**
+         * Construct with a specific sequence of (potentially different) status
+         * codes encoded as vector of cb::engine_errc elements in the
+         * request value.
+         */
+        ErrSequence(std::vector<cb::engine_errc> sequence_)
+            : FaultInjectMode(ENGINE_SUCCESS),
               sequence(sequence_),
-              pos(0) {}
+              pos(sequence.begin()) {
+        }
 
         bool should_inject_error(Cmd cmd, ENGINE_ERROR_CODE& err) {
-            bool inject = false;
-            if (pos < 32) {
-                inject = (sequence & (1 << pos)) != 0;
-                pos++;
+            if (pos == sequence.end()) {
+                throw std::logic_error(
+                        "ErrSequence::should_inject_error() Reached end of "
+                        "sequence");
             }
-            if (inject) {
-                err = injected_error;
+            bool inject = false;
+            if (*pos != cb::engine_errc(-1)) {
+                inject = true;
+                err = ENGINE_ERROR_CODE(*pos);
             }
+            pos++;
             return inject;
         }
 
+        virtual boost::optional<ENGINE_ERROR_CODE> add_to_pending_io_ops() {
+            // If this function has been called, should_inject_error() must
+            // have returned true. Return the next status code in the sequnce
+            // as the result of the pending IO.
+            if (pos == sequence.end()) {
+                throw std::logic_error(
+                        "ErrSequence::add_to_pending_io_ops() Reached end of "
+                        "sequence");
+            }
+
+            return ENGINE_ERROR_CODE(*pos++);
+        }
+
         std::string to_string() const {
             std::stringstream ss;
-            ss << "ErrSequence inject_error=" << injected_error
-               << " sequence=0x" << std::hex << sequence
-               << " pos=" << pos;
+            ss << "ErrSequence sequence=[";
+            for (const auto& err : sequence) {
+                if (err == cb::engine_errc(-1)) {
+                    ss << "'<passthrough>',";
+                } else {
+                    ss << "'" << err << "',";
+                }
+            }
+            ss << "] pos=" << pos - sequence.begin();
             return ss.str();
         }
 
     private:
-        uint32_t sequence;
-        uint32_t pos;
+        std::vector<cb::engine_errc> sequence;
+        std::vector<cb::engine_errc>::const_iterator pos;
     };
 
     class ErrOnNoNotify : public FaultInjectMode {
@@ -1168,7 +1244,10 @@ class EWB_Engine : public EngineIface, public DcpIface {
               : FaultInjectMode(injected_error_),
                 issued_return_error(false) {}
 
-            bool add_to_pending_io_ops() {return false;}
+            boost::optional<ENGINE_ERROR_CODE> add_to_pending_io_ops() {
+                return {};
+            }
+
             bool should_inject_error(Cmd cmd, ENGINE_ERROR_CODE& err) {
                 if (!issued_return_error) {
                     issued_return_error = true;
@@ -1282,8 +1361,7 @@ class EWB_Engine : public EngineIface, public DcpIface {
             suspended_map.erase(iter);
         }
 
-
-        schedule_notification(cookie);
+        schedule_notification(cookie, ENGINE_SUCCESS);
         return true;
     }
 
@@ -1304,10 +1382,10 @@ class EWB_Engine : public EngineIface, public DcpIface {
         return false;
     }
 
-    void schedule_notification(const void* cookie) {
+    void schedule_notification(const void* cookie, ENGINE_ERROR_CODE status) {
         {
             std::lock_guard<std::mutex> guard(mutex);
-            pending_io_ops.push(cookie);
+            pending_io_ops.push({cookie, status});
         }
         LOG_DEBUG("EWB_Engine: connection {} should be resumed for engine {}",
                   (void*)cookie,
@@ -1840,11 +1918,11 @@ void EWB_Engine::process_notifications() {
             return (pending_io_ops.size() > 0) || stop_notification_thread;
         });
         while (!pending_io_ops.empty()) {
-            const void* cookie = pending_io_ops.front();
+            const auto op = pending_io_ops.front();
             pending_io_ops.pop();
             lk.unlock();
-            LOG_DEBUG("EWB_Engine: notify {}", cookie);
-            server->cookie->notify_io_complete(cookie, ENGINE_SUCCESS);
+            LOG_DEBUG("EWB_Engine: notify {} status:{}", op.cookie, op.status);
+            server->cookie->notify_io_complete(op.cookie, op.status);
             lk.lock();
         }
     }

engines/ewouldblock_engine/ewouldblock_engine.h

@@ -43,8 +43,13 @@ enum class EWBEngineMode : uint32_t {
     // previous function was not this one.
     First = 2,
 
-    // Make the next N calls return a sequence of either their normal value or
-    // the injected error code. The sequence can be up to 32 elements long.
+    // Make the next N calls return the specified status code (or their normal
+    // value if -1 is specified).
+    // The sequence is encoded in the request key (given it is of variable
+    // length) as an array of cb::engine_errc codes (uint32_t).
+    // If cb::engine_errc::wouldblock is specified, then the following element
+    // in the array specifies the status code that notify_io_complete() should
+    // return.
     Sequence = 3,
 
     // Simulate CAS mismatch - make the next N store operations return

protocol/mcbp/CMakeLists.txt

@@ -10,6 +10,7 @@ add_library(mcbp STATIC
             ${Memcached_SOURCE_DIR}/include/mcbp/protocol/status.h
             datatype.cc
             dump.cc
+            ewb_encode.cc
             feature.cc
             gdb_dump_parser.cc
             lldb_dump_parser.cc

protocol/mcbp/ewb_encode.cc

@@ -0,0 +1,43 @@
+/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+/*
+ *     Copyright 2019 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.
+ */
+
+/*
+ * Helper functions to encode EWouldBlockEngine specific packets.
+ */
+
+#include "ewb_encode.h"
+
+#include <platform/socket.h>
+#include <array>
+
+std::string ewb::encodeSequence(std::vector<cb::engine_errc> sequence) {
+    // Encode vector to network-endian.
+    std::string encoded;
+    union Swap {
+        uint32_t raw;
+        std::array<uint8_t, 4> array;
+    };
+    for (const auto& err : sequence) {
+        Swap s;
+        s.raw = static_cast<uint32_t>(err);
+        s.raw = htonl(s.raw);
+        for (const auto& byte : s.array) {
+            encoded.push_back(byte);
+        }
+    }
+    return encoded;
+}

protocol/mcbp/ewb_encode.h

@@ -0,0 +1,40 @@
+/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+/*
+ *     Copyright 2019 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.
+ */
+
+/*
+ * Helper functions to encode EWouldBlockEngine specific packets.
+ */
+
+#pragma once
+
+#include <memcached/engine_error.h>
+#include <string>
+#include <vector>
+
+namespace ewb {
+/**
+ * Helper function for EWBEngineMode::Sequence, encodes a vector of
+ * status codes to inject into network order to be used as the key.
+ */
+std::string encodeSequence(std::vector<cb::engine_errc> sequence);
+
+/**
+ * Value to use for EWBEngineMode::Sequence when the original engine API error
+ *  code should just be passed through.
+ */
+static constexpr cb::engine_errc Passthrough = cb::engine_errc(-1);
+} // namespace ewb

tests/testapp/testapp.cc

@@ -24,6 +24,7 @@
 
 #include <include/memcached/protocol_binary.h>
 #include <nlohmann/json.hpp>
+#include <protocol/mcbp/ewb_encode.h>
 #include <utilities/json_utilities.h>
 #include <atomic>
 #include <chrono>
@@ -1267,6 +1268,14 @@ void TestappTest::ewouldblock_engine_configure(ENGINE_ERROR_CODE err_code,
             cb::mcbp::Status::Success);
 }
 
+void TestappTest::ewouldblock_engine_configure(
+        std::vector<cb::engine_errc> sequence) {
+    ewouldblock_engine_configure(ENGINE_SUCCESS,
+                                 EWBEngineMode::Sequence,
+                                 0,
+                                 ewb::encodeSequence(sequence));
+}
+
 void TestappTest::ewouldblock_engine_disable() {
     // Value for err_code doesn't matter...
     ewouldblock_engine_configure(ENGINE_EWOULDBLOCK, EWBEngineMode::Next_N, 0);

tests/testapp/testapp.h

@@ -217,6 +217,13 @@ class TestappTest : public ::testing::Test {
                                              uint32_t value,
                                              const std::string& key = "");
 
+    /**
+     * Configure the ewouldblock error-injecting engine for the given
+     * sequence of status codes (EWBEngineMode::Sequence2).
+     */
+    static void ewouldblock_engine_configure(
+            std::vector<cb::engine_errc> sequence);
+
     /* Disable the ewouldblock_engine. */
     static void ewouldblock_engine_disable();