<commit_before>/* -*- 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.
 */

/*
 *                "ewouldblock_engine"
 *
 * The "ewouldblock_engine" allows one to test how memcached responds when the
 * engine returns EWOULDBLOCK instead of the correct response.
 *
 * Motivation:
 *
 * The EWOULDBLOCK response code can be returned from a number of engine
 * functions, and is used to indicate that the request could not be immediately
 * fulfilled, and it "would block" if it tried to. The correct way for
 * memcached to handle this (in general) is to suspend that request until it
 * is later notified by the engine (via notify_io_complete()).
 *
 * However, engines typically return the correct response to requests
 * immediately, only rarely (and from memcached's POV non-deterministically)
 * returning EWOULDBLOCK. This makes testing of the code-paths handling
 * EWOULDBLOCK tricky.
 *
 *
 * Operation:
 * This engine, when loaded by memcached proxies requests to a "real" engine.
 * Depending on how it is configured, it can simply pass the request on to the
 * real engine, or artificially return EWOULDBLOCK back to memcached.
 *
 * See the 'Modes' enum below for the possible modes for a connection. The mode
 * can be selected by sending a `request_ewouldblock_ctl` command
 *  (opcode PROTOCOL_BINARY_CMD_EWOULDBLOCK_CTL).
 *
 * DCP:
 *    There is a special DCP stream named "ewb_internal" which is an
 *    endless stream of items. You may also add a number at the end
 *    e.g. "ewb_internal:10" and it'll create a stream with 10 entries.
 *    It will always send the same K-V pair.
 *    Note that we don't register for disconnect events so you might
 *    experience weirdness if you first try to use the internal dcp
 *    stream, and then later on want to use the one provided by the
 *    engine. The workaround for that is to delete the bucket
 *    in between ;-) (put them in separate test suites and it'll all
 *    be handled for you.
 *
 *    Any other stream name results in proxying the dcp request to
 *    the underlying engine's DCP implementation.
 *
 */

#include "ewouldblock_engine.h"

#include <atomic>
#include <condition_variable>
#include <cstring>
#include <iostream>
#include <map>
#include <memory>
#include <mutex>
#include <queue>
#include <random>
#include <sstream>
#include <string>

#include <memcached/engine.h>
#include <memcached/extension.h>
#include <platform/cb_malloc.h>
#include <platform/dirutils.h>
#include <platform/thread.h>
#include <xattr/blob.h>

#include "utilities/engine_loader.h"

/* Public API declaration ****************************************************/

extern "C" {
    MEMCACHED_PUBLIC_API
    ENGINE_ERROR_CODE create_instance(uint64_t interface, GET_SERVER_API gsa,
                                      ENGINE_HANDLE **handle);

    MEMCACHED_PUBLIC_API
    void destroy_engine(void);
}


class EWB_Engine;

// Mapping from wrapped handle to EWB handles.
static std::map<ENGINE_HANDLE*, EWB_Engine*> engine_map;

class NotificationThread : public Couchbase::Thread {
public:
    NotificationThread(EWB_Engine& engine_)
        : Thread("ewb:pendingQ"),
          engine(engine_) {}

protected:
    virtual void run() override;

protected:
    EWB_Engine& engine;
};

/**
 * The BlockMonitorThread represents the thread that is
 * monitoring the "lock" file. Once the file is no longer
 * there it will resume the client specified with the given
 * id.
 */
class BlockMonitorThread : public Couchbase::Thread {
public:
    BlockMonitorThread(EWB_Engine& engine_,
                       uint32_t id_,
                       const std::string file_)
        : Thread("ewb:BlockMon"),
          engine(engine_),
          id(id_),
          file(file_) {}

    /**
     * Wait for the underlying thread to reach the zombie state
     * (== terminated, but not reaped)
     */
    ~BlockMonitorThread() {
        waitForState(Couchbase::ThreadState::Zombie);
    }

protected:
    virtual void run() override;

private:
    EWB_Engine& engine;
    const uint32_t id;
    const std::string file;
};

static void register_callback(ENGINE_HANDLE *, ENGINE_EVENT_TYPE,
                              EVENT_CALLBACK, const void *);

static SERVER_HANDLE_V1 wrapped_api;
static SERVER_HANDLE_V1 *real_api;
static void init_wrapped_api(GET_SERVER_API fn) {
    static bool init = false;
    if (init) {
        return;
    }

    init = true;
    real_api = fn();
    wrapped_api = *real_api;

    // Overrides
    static SERVER_CALLBACK_API callback = *wrapped_api.callback;
    callback.register_callback = register_callback;
    wrapped_api.callback = &callback;
}

static SERVER_HANDLE_V1 *get_wrapped_gsa() {
    return &wrapped_api;
}

/** ewouldblock_engine class */
class EWB_Engine : public ENGINE_HANDLE_V1 {

private:
    enum class Cmd { NONE, GET_INFO, ALLOCATE, REMOVE, GET, STORE,
                     CAS, ARITHMETIC, LOCK, UNLOCK,
                     FLUSH, GET_STATS, UNKNOWN_COMMAND };

    const char* to_string(Cmd cmd);

    uint64_t (*get_connection_id)(const void* cookie);

public:
    EWB_Engine(GET_SERVER_API gsa_);

    ~EWB_Engine();

    // Convert from a handle back to the read object.
    static EWB_Engine* to_engine(ENGINE_HANDLE* handle) {
        return reinterpret_cast<EWB_Engine*> (handle);
    }

    /* Returns true if the next command should have a fake error code injected.
     * @param func Address of the command function (get, store, etc).
     * @param cookie The cookie for the user's request.
     * @param[out] Error code to return.
     */
    bool should_inject_error(Cmd cmd, const void* cookie,
                             ENGINE_ERROR_CODE& err) {

        if (is_connection_suspended(cookie)) {
            err = ENGINE_EWOULDBLOCK;
            return true;
        }

        uint64_t id = get_connection_id(cookie);

        std::lock_guard<std::mutex> guard(cookie_map_mutex);

        auto iter = connection_map.find(id);
        if (iter == connection_map.end()) {
            return false;
        }

        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) {
            auto logger = gsa()->log->get_logger();
            logger->log(EXTENSION_LOG_DEBUG, NULL,
                        "EWB_Engine: injecting error:%d for cmd:%s",
                        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);
            }
        }

        return inject;
    }

    /* Implementation of all the engine functions. ***************************/

    static const engine_info* get_info(ENGINE_HANDLE* handle) {
        return &to_engine(handle)->info.eng_info;
    }

    static ENGINE_ERROR_CODE initialize(ENGINE_HANDLE* handle,
                                        const char* config_str) {
        EWB_Engine* ewb = to_engine(handle);
        auto logger = ewb->gsa()->log->get_logger();

        // Extract the name of the real engine we will be proxying; then
        // create and initialize it.
        std::string config(config_str);
        auto seperator = config.find(";");
        std::string real_engine_name(config.substr(0, seperator));
        std::string real_engine_config;
        if (seperator != std::string::npos) {
            real_engine_config = config.substr(seperator + 1);
        }

        if ((ewb->real_engine_ref = load_engine(real_engine_name.c_str(),
                                                NULL, NULL, logger)) == NULL) {
            logger->log(EXTENSION_LOG_WARNING, NULL,
                        "ERROR: EWB_Engine::initialize(): Failed to load real "
                        "engine '%s'", real_engine_name.c_str());
            abort();
        }

        if (!create_engine_instance(ewb->real_engine_ref, get_wrapped_gsa, NULL,
                                    &ewb->real_handle)) {
            logger->log(EXTENSION_LOG_WARNING, NULL,
                        "ERROR: EWB_Engine::initialize(): Failed create "
                        "engine instance '%s'", real_engine_name.c_str());
            abort();
        }

        if (ewb->real_handle->interface != 1) {
            logger->log(EXTENSION_LOG_WARNING, NULL,
                        "ERROR: EWB_Engine::initialize(): Only support engine "
                        "with interface v1 - got v%" PRIu64 ".",
                        ewb->real_engine->interface.interface);
            abort();
        }
        ewb->real_engine =
                reinterpret_cast<ENGINE_HANDLE_V1*>(ewb->real_handle);


        engine_map[ewb->real_handle] = ewb;
        ENGINE_ERROR_CODE res = ewb->real_engine->initialize(
                ewb->real_handle, real_engine_config.c_str());

        if (res == ENGINE_SUCCESS) {
            // For engine interface functions which cannot return EWOULDBLOCK,
            // and we otherwise don't want to interpose, we can simply use the
            // real_engine's functions directly.
            ewb->ENGINE_HANDLE_V1::item_set_cas = ewb->real_engine->item_set_cas;
            ewb->ENGINE_HANDLE_V1::set_item_info = ewb->real_engine->set_item_info;
        }
        return res;
    }

    static void destroy(ENGINE_HANDLE* handle, const bool force) {
        EWB_Engine* ewb = to_engine(handle);
        ewb->real_engine->destroy(ewb->real_handle, force);
        delete ewb;
    }

    static cb::EngineErrorItemPair allocate(ENGINE_HANDLE* handle,
                                            const void* cookie,
                                            const DocKey& key,
                                            const size_t nbytes,
                                            const int flags,
                                            const rel_time_t exptime,
                                            uint8_t datatype,
                                            uint16_t vbucket) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::ALLOCATE, cookie, err)) {
            return cb::makeEngineErrorItemPair(cb::engine_errc(err));
        } else {
            return ewb->real_engine->allocate(ewb->real_handle,
                                              cookie,
                                              key,
                                              nbytes,
                                              flags,
                                              exptime,
                                              datatype,
                                              vbucket);
        }
    }

    static std::pair<cb::unique_item_ptr, item_info> allocate_ex(ENGINE_HANDLE* handle,
                                                                 const void* cookie,
                                                                 const DocKey& key,
                                                                 const size_t nbytes,
                                                                 const size_t priv_nbytes,
                                                                 const int flags,
                                                                 const rel_time_t exptime,
                                                                 uint8_t datatype,
                                                                 uint16_t vbucket) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::ALLOCATE, cookie, err)) {
            throw cb::engine_error(cb::engine_errc(err), "ewb: injecting error");
        } else {
            return ewb->real_engine->allocate_ex(ewb->real_handle, cookie,
                                                 key, nbytes, priv_nbytes,
                                                 flags, exptime, datatype,
                                                 vbucket);
        }
    }

    static ENGINE_ERROR_CODE remove(ENGINE_HANDLE* handle, const void* cookie,
                                    const DocKey& key, uint64_t* cas,
                                    uint16_t vbucket,
                                    mutation_descr_t* mut_info) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::REMOVE, cookie, err)) {
            return err;
        } else {
            return ewb->real_engine->remove(ewb->real_handle, cookie, key, cas,
                                            vbucket, mut_info);
        }
    }

    static void release(ENGINE_HANDLE* handle, const void *cookie, item* item) {
        EWB_Engine* ewb = to_engine(handle);
        auto logger = ewb->gsa()->log->get_logger();
        logger->log(EXTENSION_LOG_DEBUG, nullptr, "EWB_Engine: release");

        if (item == &ewb->dcp_mutation_item) {
            // Ignore the DCP mutation, we own it (and don't track
            // refcounts on it).
        } else {
            return ewb->real_engine->release(ewb->real_handle, cookie, item);
        }
    }

    static cb::EngineErrorItemPair get(ENGINE_HANDLE* handle,
                                       const void* cookie,
                                       const DocKey& key,
                                       uint16_t vbucket,
                                       DocStateFilter documentStateFilter) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::GET, cookie, err)) {
            return std::make_pair(
                    cb::engine_errc(err),
                    cb::unique_item_ptr{nullptr, cb::ItemDeleter{handle}});
        } else {
            return ewb->real_engine->get(ewb->real_handle,
                                         cookie,
                                         key,
                                         vbucket,
                                         documentStateFilter);
        }
    }

    static cb::EngineErrorItemPair get_if(ENGINE_HANDLE* handle,
                                          const void* cookie,
                                          const DocKey& key,
                                          uint16_t vbucket,
                                          std::function<bool(
                                              const item_info&)> filter) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::GET, cookie, err)) {
            return cb::makeEngineErrorItemPair(cb::engine_errc::would_block);
        } else {
            return ewb->real_engine->get_if(
                    ewb->real_handle, cookie, key, vbucket, filter);
        }
    }

    static cb::EngineErrorItemPair get_and_touch(ENGINE_HANDLE* handle,
                                                 const void* cookie,
                                                 const DocKey& key,
                                                 uint16_t vbucket,
                                                 uint32_t exptime) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::GET, cookie, err)) {
            return cb::makeEngineErrorItemPair(cb::engine_errc::would_block);
        } else {
            return ewb->real_engine->get_and_touch(
                    ewb->real_handle, cookie, key, vbucket, exptime);
        }
    }

    static cb::EngineErrorItemPair get_locked(ENGINE_HANDLE* handle,
                                              const void* cookie,
                                              const DocKey& key,
                                              uint16_t vbucket,
                                              uint32_t lock_timeout) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::LOCK, cookie, err)) {
            return cb::makeEngineErrorItemPair(cb::engine_errc(err));
        } else {
            return ewb->real_engine->get_locked(
                    ewb->real_handle, cookie, key, vbucket, lock_timeout);
        }
    }

    static ENGINE_ERROR_CODE unlock(ENGINE_HANDLE* handle,
                                        const void* cookie,
                                        const DocKey& key,
                                        uint16_t vbucket,
                                        uint64_t cas) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::UNLOCK, cookie, err)) {
            return err;
        } else {
            return ewb->real_engine->unlock(ewb->real_handle, cookie, key,
                                            vbucket, cas);
        }
    }


    static ENGINE_ERROR_CODE store(ENGINE_HANDLE* handle, const void *cookie,
                                   item* item, uint64_t *cas,
                                   ENGINE_STORE_OPERATION operation,
                                   DocumentState document_state) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        Cmd opcode = (operation == OPERATION_CAS) ? Cmd::CAS : Cmd::STORE;
        if (ewb->should_inject_error(opcode, cookie, err)) {
            return err;
        } else {
            return ewb->real_engine->store(ewb->real_handle, cookie, item, cas,
                                           operation, document_state);
        }
    }

    static cb::EngineErrorCasPair store_if(ENGINE_HANDLE* handle,
                                           const void* cookie,
                                           item* item,
                                           uint64_t cas,
                                           ENGINE_STORE_OPERATION operation,
                                           cb::StoreIfPredicate predicate,
                                           DocumentState document_state) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        Cmd opcode = (operation == OPERATION_CAS) ? Cmd::CAS : Cmd::STORE;
        if (ewb->should_inject_error(opcode, cookie, err)) {
            return {cb::engine_errc(err), 0};
        } else {
            return ewb->real_engine->store_if(ewb->real_handle,
                                              cookie,
                                              item,
                                              cas,
                                              operation,
                                              predicate,
                                              document_state);
        }
    }

    static ENGINE_ERROR_CODE flush(ENGINE_HANDLE* handle, const void* cookie) {
        // Flush is a little different - it often returns EWOULDBLOCK, and
        // notify_io_complete() just tells the server it can issue it's *next*
        // command (i.e. no need to re-flush). Therefore just pass Flush
        // straight through for now.
        EWB_Engine* ewb = to_engine(handle);
        return ewb->real_engine->flush(ewb->real_handle, cookie);
    }

    static ENGINE_ERROR_CODE get_stats(ENGINE_HANDLE* handle,
                                       const void* cookie, const char* stat_key,
                                       int nkey, ADD_STAT add_stat) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::GET_STATS, cookie, err)) {
            return err;
        } else {
            return ewb->real_engine->get_stats(ewb->real_handle, cookie, stat_key,
                                               nkey, add_stat);
        }
    }

    static void reset_stats(ENGINE_HANDLE* handle, const void* cookie) {
        EWB_Engine* ewb = to_engine(handle);
        return ewb->real_engine->reset_stats(ewb->real_handle, cookie);
    }

    /* Handle 'unknown_command'. In additional to wrapping calls to the
     * underlying real engine, this is also used to configure
     * ewouldblock_engine itself using he CMD_EWOULDBLOCK_CTL opcode.
     */
    static ENGINE_ERROR_CODE unknown_command(ENGINE_HANDLE* handle,
                                             const void* cookie,
                                             protocol_binary_request_header *request,
                                             ADD_RESPONSE response,
                                             DocNamespace doc_namespace) {
        EWB_Engine* ewb = to_engine(handle);

        if (request->request.opcode == PROTOCOL_BINARY_CMD_EWOULDBLOCK_CTL) {
            auto logger = ewb->gsa()->log->get_logger();
            auto* req = reinterpret_cast<request_ewouldblock_ctl*>(request);
            const EWBEngineMode mode = static_cast<EWBEngineMode>(ntohl(req->message.body.mode));
            const uint32_t value = ntohl(req->message.body.value);
            const ENGINE_ERROR_CODE injected_error =
                    static_cast<ENGINE_ERROR_CODE>(ntohl(req->message.body.inject_error));
            const std::string key((char*)req->bytes + sizeof(req->bytes),
                                  ntohs(req->message.header.request.keylen));

            std::shared_ptr<FaultInjectMode> new_mode = nullptr;

            // Validate mode, and construct new fault injector.
            switch (mode) {
                case EWBEngineMode::Next_N:
                    new_mode = std::make_shared<ErrOnNextN>(injected_error, value);
                    break;

                case EWBEngineMode::Random:
                    new_mode = std::make_shared<ErrRandom>(injected_error, value);
                    break;

                case EWBEngineMode::First:
                    new_mode = std::make_shared<ErrOnFirst>(injected_error);
                    break;

                case EWBEngineMode::Sequence:
                    new_mode = std::make_shared<ErrSequence>(injected_error, value);
                    break;

                case EWBEngineMode::No_Notify:
                    new_mode = std::make_shared<ErrOnNoNotify>(injected_error);
                    break;

                case EWBEngineMode::CasMismatch:
                    new_mode = std::make_shared<CASMismatch>(value);
                    break;

                case EWBEngineMode::IncrementClusterMapRevno:
                    ewb->clustermap_revno++;
                    response(nullptr, 0, nullptr, 0, nullptr, 0,
                             PROTOCOL_BINARY_RAW_BYTES,
                             PROTOCOL_BINARY_RESPONSE_SUCCESS, 0, cookie);
                    return ENGINE_SUCCESS;

                case EWBEngineMode::BlockMonitorFile:
                    return ewb->handleBlockMonitorFile(cookie, value, key,
                                                       response);

                case EWBEngineMode::Suspend:
                    return ewb->handleSuspend(cookie, value, response);

                case EWBEngineMode::Resume:
                    return ewb->handleResume(cookie, value, response);

                case EWBEngineMode::SetItemCas:
                    return ewb->setItemCas(cookie, key, value, response);
            }

            if (new_mode == nullptr) {
                logger->log(EXTENSION_LOG_WARNING, NULL,
                            "EWB_Engine::unknown_command(): "
                            "Got unexpected mode=%d for EWOULDBLOCK_CTL, ",
                            mode);
                response(nullptr, 0, nullptr, 0, nullptr, 0,
                         PROTOCOL_BINARY_RAW_BYTES,
                         PROTOCOL_BINARY_RESPONSE_EINVAL, /*cas*/0, cookie);
                return ENGINE_FAILED;
            } else {
                try {
                    logger->log(EXTENSION_LOG_DEBUG, NULL,
                                "EWB_Engine::unknown_command(): Setting EWB mode to "
                                "%s for cookie %d", new_mode->to_string().c_str(),
                                cookie);

                    uint64_t id = ewb->get_connection_id(cookie);

                    {
                        std::lock_guard<std::mutex> guard(ewb->cookie_map_mutex);
                        ewb->connection_map.erase(id);
                        ewb->connection_map.emplace(id, std::make_pair(cookie, new_mode));
                    }

                    response(nullptr, 0, nullptr, 0, nullptr, 0,
                             PROTOCOL_BINARY_RAW_BYTES,
                             PROTOCOL_BINARY_RESPONSE_SUCCESS, /*cas*/0, cookie);
                    return ENGINE_SUCCESS;
                } catch (std::bad_alloc&) {
                    return ENGINE_ENOMEM;
                }
            }
        } else {
            ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
            if (ewb->should_inject_error(Cmd::UNKNOWN_COMMAND, cookie, err)) {
                return err;
            } else {
                return ewb->real_engine->unknown_command(ewb->real_handle, cookie,
                                                         request, response,
                                                         doc_namespace);
            }
        }
    }

    static void item_set_cas(ENGINE_HANDLE *handle, const void* cookie,
                             item* item, uint64_t cas) {
        // Should never be called as ENGINE_HANDLE_V1::item_set_cas is updated
        // to point to the real_engine once it is initialized. This function
        //only exists so there is a non-NULL value for
        // ENGINE_HANDLE_V1::item_set_cas initially to keep load_engine()
        // happy.
        abort();
    }

    static bool get_item_info(ENGINE_HANDLE *handle, const void *cookie,
                              const item* item, item_info *item_info) {
        EWB_Engine* ewb = to_engine(handle);
        auto logger = ewb->gsa()->log->get_logger();
        logger->log(EXTENSION_LOG_DEBUG, nullptr, "EWB_Engine: get_item_info");

        // This function cannot return EWOULDBLOCK - just chain to the real
        // engine's function, unless it is a request for our special DCP item.
        if (item == &ewb->dcp_mutation_item) {
            item_info->cas = 0;
            item_info->vbucket_uuid = 0;
            item_info->seqno = 0;
            item_info->exptime = 0;
            item_info->nbytes = ewb->dcp_mutation_item.value.size();
            item_info->flags = 0;
            item_info->datatype = PROTOCOL_BINARY_DATATYPE_XATTR;
            item_info->nkey = ewb->dcp_mutation_item.key.size();
            item_info->key = ewb->dcp_mutation_item.key.c_str();
            item_info->value[0].iov_base = &ewb->dcp_mutation_item.value[0];
            item_info->value[0].iov_len = item_info->nbytes;
            return true;
        } else {
            return ewb->real_engine->get_item_info(ewb->real_handle, cookie,
                                                   item, item_info);
        }
    }

    static bool set_item_info(ENGINE_HANDLE *handle, const void *cookie,
                              item* item, const item_info *item_info) {
        // Should never be called - set item_set_cas().
        abort();
    }

    static ENGINE_ERROR_CODE get_engine_vb_map(ENGINE_HANDLE* handle,
                                               const void * cookie,
                                               engine_get_vb_map_cb callback) {
        // Used to test NOT_MY_VBUCKET - just return a dummy config.
        EWB_Engine* ewb = to_engine(handle);
        auto logger = ewb->gsa()->log->get_logger();
        logger->log(EXTENSION_LOG_DEBUG, NULL, "EWB_Engine::get_engine_vb_map");

        std::string vbmap =
            "{\"rev\":" + std::to_string(ewb->clustermap_revno.load()) + "}";
        callback(cookie, vbmap.data(), vbmap.length());

        return ENGINE_SUCCESS;
    }


    GET_SERVER_API gsa;
    union {
        engine_info eng_info;
        char buffer[sizeof(engine_info) +
                    (sizeof(feature_info) * LAST_REGISTERED_ENGINE_FEATURE)];
    } info;

    // Actual engine we are proxying requests to.
    ENGINE_HANDLE* real_handle;
    ENGINE_HANDLE_V1* real_engine;
    engine_reference* real_engine_ref;

    std::atomic_int clustermap_revno;

    /**
     * The method responsible for pushing all of the notify_io_complete
     * to the frontend. It is run by notify_io_thread and not intended to
     * be called by anyone else!.
     */
    void process_notifications();<commit_msg>MB-25733: ewouldblock_engine: Remove disconnected connections from map<commit_after>/* -*- 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.
 */

/*
 *                "ewouldblock_engine"
 *
 * The "ewouldblock_engine" allows one to test how memcached responds when the
 * engine returns EWOULDBLOCK instead of the correct response.
 *
 * Motivation:
 *
 * The EWOULDBLOCK response code can be returned from a number of engine
 * functions, and is used to indicate that the request could not be immediately
 * fulfilled, and it "would block" if it tried to. The correct way for
 * memcached to handle this (in general) is to suspend that request until it
 * is later notified by the engine (via notify_io_complete()).
 *
 * However, engines typically return the correct response to requests
 * immediately, only rarely (and from memcached's POV non-deterministically)
 * returning EWOULDBLOCK. This makes testing of the code-paths handling
 * EWOULDBLOCK tricky.
 *
 *
 * Operation:
 * This engine, when loaded by memcached proxies requests to a "real" engine.
 * Depending on how it is configured, it can simply pass the request on to the
 * real engine, or artificially return EWOULDBLOCK back to memcached.
 *
 * See the 'Modes' enum below for the possible modes for a connection. The mode
 * can be selected by sending a `request_ewouldblock_ctl` command
 *  (opcode PROTOCOL_BINARY_CMD_EWOULDBLOCK_CTL).
 *
 * DCP:
 *    There is a special DCP stream named "ewb_internal" which is an
 *    endless stream of items. You may also add a number at the end
 *    e.g. "ewb_internal:10" and it'll create a stream with 10 entries.
 *    It will always send the same K-V pair.
 *    Note that we don't register for disconnect events so you might
 *    experience weirdness if you first try to use the internal dcp
 *    stream, and then later on want to use the one provided by the
 *    engine. The workaround for that is to delete the bucket
 *    in between ;-) (put them in separate test suites and it'll all
 *    be handled for you.
 *
 *    Any other stream name results in proxying the dcp request to
 *    the underlying engine's DCP implementation.
 *
 */

#include "ewouldblock_engine.h"

#include <atomic>
#include <condition_variable>
#include <cstring>
#include <iostream>
#include <map>
#include <memory>
#include <mutex>
#include <queue>
#include <random>
#include <sstream>
#include <string>

#include <memcached/engine.h>
#include <memcached/extension.h>
#include <platform/cb_malloc.h>
#include <platform/dirutils.h>
#include <platform/thread.h>
#include <xattr/blob.h>

#include "utilities/engine_loader.h"

/* Public API declaration ****************************************************/

extern "C" {
    MEMCACHED_PUBLIC_API
    ENGINE_ERROR_CODE create_instance(uint64_t interface, GET_SERVER_API gsa,
                                      ENGINE_HANDLE **handle);

    MEMCACHED_PUBLIC_API
    void destroy_engine(void);
}


class EWB_Engine;

// Mapping from wrapped handle to EWB handles.
static std::map<ENGINE_HANDLE*, EWB_Engine*> engine_map;

class NotificationThread : public Couchbase::Thread {
public:
    NotificationThread(EWB_Engine& engine_)
        : Thread("ewb:pendingQ"),
          engine(engine_) {}

protected:
    virtual void run() override;

protected:
    EWB_Engine& engine;
};

/**
 * The BlockMonitorThread represents the thread that is
 * monitoring the "lock" file. Once the file is no longer
 * there it will resume the client specified with the given
 * id.
 */
class BlockMonitorThread : public Couchbase::Thread {
public:
    BlockMonitorThread(EWB_Engine& engine_,
                       uint32_t id_,
                       const std::string file_)
        : Thread("ewb:BlockMon"),
          engine(engine_),
          id(id_),
          file(file_) {}

    /**
     * Wait for the underlying thread to reach the zombie state
     * (== terminated, but not reaped)
     */
    ~BlockMonitorThread() {
        waitForState(Couchbase::ThreadState::Zombie);
    }

protected:
    virtual void run() override;

private:
    EWB_Engine& engine;
    const uint32_t id;
    const std::string file;
};

static void register_callback(ENGINE_HANDLE *, ENGINE_EVENT_TYPE,
                              EVENT_CALLBACK, const void *);

static SERVER_HANDLE_V1 wrapped_api;
static SERVER_HANDLE_V1 *real_api;
static void init_wrapped_api(GET_SERVER_API fn) {
    static bool init = false;
    if (init) {
        return;
    }

    init = true;
    real_api = fn();
    wrapped_api = *real_api;

    // Overrides
    static SERVER_CALLBACK_API callback = *wrapped_api.callback;
    callback.register_callback = register_callback;
    wrapped_api.callback = &callback;
}

static SERVER_HANDLE_V1 *get_wrapped_gsa() {
    return &wrapped_api;
}

/** ewouldblock_engine class */
class EWB_Engine : public ENGINE_HANDLE_V1 {

private:
    enum class Cmd { NONE, GET_INFO, ALLOCATE, REMOVE, GET, STORE,
                     CAS, ARITHMETIC, LOCK, UNLOCK,
                     FLUSH, GET_STATS, UNKNOWN_COMMAND };

    const char* to_string(Cmd cmd);

    uint64_t (*get_connection_id)(const void* cookie);

public:
    EWB_Engine(GET_SERVER_API gsa_);

    ~EWB_Engine();

    // Convert from a handle back to the read object.
    static EWB_Engine* to_engine(ENGINE_HANDLE* handle) {
        return reinterpret_cast<EWB_Engine*> (handle);
    }

    /* Returns true if the next command should have a fake error code injected.
     * @param func Address of the command function (get, store, etc).
     * @param cookie The cookie for the user's request.
     * @param[out] Error code to return.
     */
    bool should_inject_error(Cmd cmd, const void* cookie,
                             ENGINE_ERROR_CODE& err) {

        if (is_connection_suspended(cookie)) {
            err = ENGINE_EWOULDBLOCK;
            return true;
        }

        uint64_t id = get_connection_id(cookie);

        std::lock_guard<std::mutex> guard(cookie_map_mutex);

        auto iter = connection_map.find(id);
        if (iter == connection_map.end()) {
            return false;
        }

        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) {
            auto logger = gsa()->log->get_logger();
            logger->log(EXTENSION_LOG_DEBUG, NULL,
                        "EWB_Engine: injecting error:%d for cmd:%s",
                        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);
            }
        }

        return inject;
    }

    /* Implementation of all the engine functions. ***************************/

    static const engine_info* get_info(ENGINE_HANDLE* handle) {
        return &to_engine(handle)->info.eng_info;
    }

    static ENGINE_ERROR_CODE initialize(ENGINE_HANDLE* handle,
                                        const char* config_str) {
        EWB_Engine* ewb = to_engine(handle);
        auto logger = ewb->gsa()->log->get_logger();

        // Extract the name of the real engine we will be proxying; then
        // create and initialize it.
        std::string config(config_str);
        auto seperator = config.find(";");
        std::string real_engine_name(config.substr(0, seperator));
        std::string real_engine_config;
        if (seperator != std::string::npos) {
            real_engine_config = config.substr(seperator + 1);
        }

        if ((ewb->real_engine_ref = load_engine(real_engine_name.c_str(),
                                                NULL, NULL, logger)) == NULL) {
            logger->log(EXTENSION_LOG_WARNING, NULL,
                        "ERROR: EWB_Engine::initialize(): Failed to load real "
                        "engine '%s'", real_engine_name.c_str());
            abort();
        }

        if (!create_engine_instance(ewb->real_engine_ref, get_wrapped_gsa, NULL,
                                    &ewb->real_handle)) {
            logger->log(EXTENSION_LOG_WARNING, NULL,
                        "ERROR: EWB_Engine::initialize(): Failed create "
                        "engine instance '%s'", real_engine_name.c_str());
            abort();
        }

        if (ewb->real_handle->interface != 1) {
            logger->log(EXTENSION_LOG_WARNING, NULL,
                        "ERROR: EWB_Engine::initialize(): Only support engine "
                        "with interface v1 - got v%" PRIu64 ".",
                        ewb->real_engine->interface.interface);
            abort();
        }
        ewb->real_engine =
                reinterpret_cast<ENGINE_HANDLE_V1*>(ewb->real_handle);


        engine_map[ewb->real_handle] = ewb;
        ENGINE_ERROR_CODE res = ewb->real_engine->initialize(
                ewb->real_handle, real_engine_config.c_str());

        if (res == ENGINE_SUCCESS) {
            // For engine interface functions which cannot return EWOULDBLOCK,
            // and we otherwise don't want to interpose, we can simply use the
            // real_engine's functions directly.
            ewb->ENGINE_HANDLE_V1::item_set_cas = ewb->real_engine->item_set_cas;
            ewb->ENGINE_HANDLE_V1::set_item_info = ewb->real_engine->set_item_info;
        }

        // Register a callback on DISCONNECT events, so we can delete
        // any stale elements from connection_map when a connection
        // DC's.
        real_api->callback->register_callback(
                reinterpret_cast<ENGINE_HANDLE*>(ewb),
                ON_DISCONNECT,
                handle_disconnect,
                reinterpret_cast<ENGINE_HANDLE*>(ewb));

        return res;
    }

    static void destroy(ENGINE_HANDLE* handle, const bool force) {
        EWB_Engine* ewb = to_engine(handle);
        ewb->real_engine->destroy(ewb->real_handle, force);
        delete ewb;
    }

    static cb::EngineErrorItemPair allocate(ENGINE_HANDLE* handle,
                                            const void* cookie,
                                            const DocKey& key,
                                            const size_t nbytes,
                                            const int flags,
                                            const rel_time_t exptime,
                                            uint8_t datatype,
                                            uint16_t vbucket) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::ALLOCATE, cookie, err)) {
            return cb::makeEngineErrorItemPair(cb::engine_errc(err));
        } else {
            return ewb->real_engine->allocate(ewb->real_handle,
                                              cookie,
                                              key,
                                              nbytes,
                                              flags,
                                              exptime,
                                              datatype,
                                              vbucket);
        }
    }

    static std::pair<cb::unique_item_ptr, item_info> allocate_ex(ENGINE_HANDLE* handle,
                                                                 const void* cookie,
                                                                 const DocKey& key,
                                                                 const size_t nbytes,
                                                                 const size_t priv_nbytes,
                                                                 const int flags,
                                                                 const rel_time_t exptime,
                                                                 uint8_t datatype,
                                                                 uint16_t vbucket) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::ALLOCATE, cookie, err)) {
            throw cb::engine_error(cb::engine_errc(err), "ewb: injecting error");
        } else {
            return ewb->real_engine->allocate_ex(ewb->real_handle, cookie,
                                                 key, nbytes, priv_nbytes,
                                                 flags, exptime, datatype,
                                                 vbucket);
        }
    }

    static ENGINE_ERROR_CODE remove(ENGINE_HANDLE* handle, const void* cookie,
                                    const DocKey& key, uint64_t* cas,
                                    uint16_t vbucket,
                                    mutation_descr_t* mut_info) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::REMOVE, cookie, err)) {
            return err;
        } else {
            return ewb->real_engine->remove(ewb->real_handle, cookie, key, cas,
                                            vbucket, mut_info);
        }
    }

    static void release(ENGINE_HANDLE* handle, const void *cookie, item* item) {
        EWB_Engine* ewb = to_engine(handle);
        auto logger = ewb->gsa()->log->get_logger();
        logger->log(EXTENSION_LOG_DEBUG, nullptr, "EWB_Engine: release");

        if (item == &ewb->dcp_mutation_item) {
            // Ignore the DCP mutation, we own it (and don't track
            // refcounts on it).
        } else {
            return ewb->real_engine->release(ewb->real_handle, cookie, item);
        }
    }

    static cb::EngineErrorItemPair get(ENGINE_HANDLE* handle,
                                       const void* cookie,
                                       const DocKey& key,
                                       uint16_t vbucket,
                                       DocStateFilter documentStateFilter) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::GET, cookie, err)) {
            return std::make_pair(
                    cb::engine_errc(err),
                    cb::unique_item_ptr{nullptr, cb::ItemDeleter{handle}});
        } else {
            return ewb->real_engine->get(ewb->real_handle,
                                         cookie,
                                         key,
                                         vbucket,
                                         documentStateFilter);
        }
    }

    static cb::EngineErrorItemPair get_if(ENGINE_HANDLE* handle,
                                          const void* cookie,
                                          const DocKey& key,
                                          uint16_t vbucket,
                                          std::function<bool(
                                              const item_info&)> filter) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::GET, cookie, err)) {
            return cb::makeEngineErrorItemPair(cb::engine_errc::would_block);
        } else {
            return ewb->real_engine->get_if(
                    ewb->real_handle, cookie, key, vbucket, filter);
        }
    }

    static cb::EngineErrorItemPair get_and_touch(ENGINE_HANDLE* handle,
                                                 const void* cookie,
                                                 const DocKey& key,
                                                 uint16_t vbucket,
                                                 uint32_t exptime) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::GET, cookie, err)) {
            return cb::makeEngineErrorItemPair(cb::engine_errc::would_block);
        } else {
            return ewb->real_engine->get_and_touch(
                    ewb->real_handle, cookie, key, vbucket, exptime);
        }
    }

    static cb::EngineErrorItemPair get_locked(ENGINE_HANDLE* handle,
                                              const void* cookie,
                                              const DocKey& key,
                                              uint16_t vbucket,
                                              uint32_t lock_timeout) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::LOCK, cookie, err)) {
            return cb::makeEngineErrorItemPair(cb::engine_errc(err));
        } else {
            return ewb->real_engine->get_locked(
                    ewb->real_handle, cookie, key, vbucket, lock_timeout);
        }
    }

    static ENGINE_ERROR_CODE unlock(ENGINE_HANDLE* handle,
                                        const void* cookie,
                                        const DocKey& key,
                                        uint16_t vbucket,
                                        uint64_t cas) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::UNLOCK, cookie, err)) {
            return err;
        } else {
            return ewb->real_engine->unlock(ewb->real_handle, cookie, key,
                                            vbucket, cas);
        }
    }


    static ENGINE_ERROR_CODE store(ENGINE_HANDLE* handle, const void *cookie,
                                   item* item, uint64_t *cas,
                                   ENGINE_STORE_OPERATION operation,
                                   DocumentState document_state) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        Cmd opcode = (operation == OPERATION_CAS) ? Cmd::CAS : Cmd::STORE;
        if (ewb->should_inject_error(opcode, cookie, err)) {
            return err;
        } else {
            return ewb->real_engine->store(ewb->real_handle, cookie, item, cas,
                                           operation, document_state);
        }
    }

    static cb::EngineErrorCasPair store_if(ENGINE_HANDLE* handle,
                                           const void* cookie,
                                           item* item,
                                           uint64_t cas,
                                           ENGINE_STORE_OPERATION operation,
                                           cb::StoreIfPredicate predicate,
                                           DocumentState document_state) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        Cmd opcode = (operation == OPERATION_CAS) ? Cmd::CAS : Cmd::STORE;
        if (ewb->should_inject_error(opcode, cookie, err)) {
            return {cb::engine_errc(err), 0};
        } else {
            return ewb->real_engine->store_if(ewb->real_handle,
                                              cookie,
                                              item,
                                              cas,
                                              operation,
                                              predicate,
                                              document_state);
        }
    }

    static ENGINE_ERROR_CODE flush(ENGINE_HANDLE* handle, const void* cookie) {
        // Flush is a little different - it often returns EWOULDBLOCK, and
        // notify_io_complete() just tells the server it can issue it's *next*
        // command (i.e. no need to re-flush). Therefore just pass Flush
        // straight through for now.
        EWB_Engine* ewb = to_engine(handle);
        return ewb->real_engine->flush(ewb->real_handle, cookie);
    }

    static ENGINE_ERROR_CODE get_stats(ENGINE_HANDLE* handle,
                                       const void* cookie, const char* stat_key,
                                       int nkey, ADD_STAT add_stat) {
        EWB_Engine* ewb = to_engine(handle);
        ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
        if (ewb->should_inject_error(Cmd::GET_STATS, cookie, err)) {
            return err;
        } else {
            return ewb->real_engine->get_stats(ewb->real_handle, cookie, stat_key,
                                               nkey, add_stat);
        }
    }

    static void reset_stats(ENGINE_HANDLE* handle, const void* cookie) {
        EWB_Engine* ewb = to_engine(handle);
        return ewb->real_engine->reset_stats(ewb->real_handle, cookie);
    }

    /* Handle 'unknown_command'. In additional to wrapping calls to the
     * underlying real engine, this is also used to configure
     * ewouldblock_engine itself using he CMD_EWOULDBLOCK_CTL opcode.
     */
    static ENGINE_ERROR_CODE unknown_command(ENGINE_HANDLE* handle,
                                             const void* cookie,
                                             protocol_binary_request_header *request,
                                             ADD_RESPONSE response,
                                             DocNamespace doc_namespace) {
        EWB_Engine* ewb = to_engine(handle);

        if (request->request.opcode == PROTOCOL_BINARY_CMD_EWOULDBLOCK_CTL) {
            auto logger = ewb->gsa()->log->get_logger();
            auto* req = reinterpret_cast<request_ewouldblock_ctl*>(request);
            const EWBEngineMode mode = static_cast<EWBEngineMode>(ntohl(req->message.body.mode));
            const uint32_t value = ntohl(req->message.body.value);
            const ENGINE_ERROR_CODE injected_error =
                    static_cast<ENGINE_ERROR_CODE>(ntohl(req->message.body.inject_error));
            const std::string key((char*)req->bytes + sizeof(req->bytes),
                                  ntohs(req->message.header.request.keylen));

            std::shared_ptr<FaultInjectMode> new_mode = nullptr;

            // Validate mode, and construct new fault injector.
            switch (mode) {
                case EWBEngineMode::Next_N:
                    new_mode = std::make_shared<ErrOnNextN>(injected_error, value);
                    break;

                case EWBEngineMode::Random:
                    new_mode = std::make_shared<ErrRandom>(injected_error, value);
                    break;

                case EWBEngineMode::First:
                    new_mode = std::make_shared<ErrOnFirst>(injected_error);
                    break;

                case EWBEngineMode::Sequence:
                    new_mode = std::make_shared<ErrSequence>(injected_error, value);
                    break;

                case EWBEngineMode::No_Notify:
                    new_mode = std::make_shared<ErrOnNoNotify>(injected_error);
                    break;

                case EWBEngineMode::CasMismatch:
                    new_mode = std::make_shared<CASMismatch>(value);
                    break;

                case EWBEngineMode::IncrementClusterMapRevno:
                    ewb->clustermap_revno++;
                    response(nullptr, 0, nullptr, 0, nullptr, 0,
                             PROTOCOL_BINARY_RAW_BYTES,
                             PROTOCOL_BINARY_RESPONSE_SUCCESS, 0, cookie);
                    return ENGINE_SUCCESS;

                case EWBEngineMode::BlockMonitorFile:
                    return ewb->handleBlockMonitorFile(cookie, value, key,
                                                       response);

                case EWBEngineMode::Suspend:
                    return ewb->handleSuspend(cookie, value, response);

                case EWBEngineMode::Resume:
                    return ewb->handleResume(cookie, value, response);

                case EWBEngineMode::SetItemCas:
                    return ewb->setItemCas(cookie, key, value, response);
            }

            if (new_mode == nullptr) {
                logger->log(EXTENSION_LOG_WARNING, NULL,
                            "EWB_Engine::unknown_command(): "
                            "Got unexpected mode=%d for EWOULDBLOCK_CTL, ",
                            mode);
                response(nullptr, 0, nullptr, 0, nullptr, 0,
                         PROTOCOL_BINARY_RAW_BYTES,
                         PROTOCOL_BINARY_RESPONSE_EINVAL, /*cas*/0, cookie);
                return ENGINE_FAILED;
            } else {
                try {
                    logger->log(EXTENSION_LOG_DEBUG, NULL,
                                "EWB_Engine::unknown_command(): Setting EWB mode to "
                                "%s for cookie %d", new_mode->to_string().c_str(),
                                cookie);

                    uint64_t id = ewb->get_connection_id(cookie);

                    {
                        std::lock_guard<std::mutex> guard(ewb->cookie_map_mutex);
                        ewb->connection_map.erase(id);
                        ewb->connection_map.emplace(id, std::make_pair(cookie, new_mode));
                    }

                    response(nullptr, 0, nullptr, 0, nullptr, 0,
                             PROTOCOL_BINARY_RAW_BYTES,
                             PROTOCOL_BINARY_RESPONSE_SUCCESS, /*cas*/0, cookie);
                    return ENGINE_SUCCESS;
                } catch (std::bad_alloc&) {
                    return ENGINE_ENOMEM;
                }
            }
        } else {
            ENGINE_ERROR_CODE err = ENGINE_SUCCESS;
            if (ewb->should_inject_error(Cmd::UNKNOWN_COMMAND, cookie, err)) {
                return err;
            } else {
                return ewb->real_engine->unknown_command(ewb->real_handle, cookie,
                                                         request, response,
                                                         doc_namespace);
            }
        }
    }

    static void item_set_cas(ENGINE_HANDLE *handle, const void* cookie,
                             item* item, uint64_t cas) {
        // Should never be called as ENGINE_HANDLE_V1::item_set_cas is updated
        // to point to the real_engine once it is initialized. This function
        //only exists so there is a non-NULL value for
        // ENGINE_HANDLE_V1::item_set_cas initially to keep load_engine()
        // happy.
        abort();
    }

    static bool get_item_info(ENGINE_HANDLE *handle, const void *cookie,
                              const item* item, item_info *item_info) {
        EWB_Engine* ewb = to_engine(handle);
        auto logger = ewb->gsa()->log->get_logger();
        logger->log(EXTENSION_LOG_DEBUG, nullptr, "EWB_Engine: get_item_info");

        // This function cannot return EWOULDBLOCK - just chain to the real
        // engine's function, unless it is a request for our special DCP item.
        if (item == &ewb->dcp_mutation_item) {
            item_info->cas = 0;
            item_info->vbucket_uuid = 0;
            item_info->seqno = 0;
            item_info->exptime = 0;
            item_info->nbytes = ewb->dcp_mutation_item.value.size();
            item_info->flags = 0;
            item_info->datatype = PROTOCOL_BINARY_DATATYPE_XATTR;
            item_info->nkey = ewb->dcp_mutation_item.key.size();
            item_info->key = ewb->dcp_mutation_item.key.c_str();
            item_info->value[0].iov_base = &ewb->dcp_mutation_item.value[0];
            item_info->value[0].iov_len = item_info->nbytes;
            return true;
        } else {
            return ewb->real_engine->get_item_info(ewb->real_handle, cookie,
                                                   item, item_info);
        }
    }

    static bool set_item_info(ENGINE_HANDLE *handle, const void *cookie,
                              item* item, const item_info *item_info) {
        // Should never be called - set item_set_cas().
        abort();
    }

    static ENGINE_ERROR_CODE get_engine_vb_map(ENGINE_HANDLE* handle,
                                               const void * cookie,
                                               engine_get_vb_map_cb callback) {
        // Used to test NOT_MY_VBUCKET - just return a dummy config.
        EWB_Engine* ewb = to_engine(handle);
        auto logger = ewb->gsa()->log->get_logger();
        logger->log(EXTENSION_LOG_DEBUG, NULL, "EWB_Engine::get_engine_vb_map");

        std::string vbmap =
            "{\"rev\":" + std::to_string(ewb->clustermap_revno.load()) + "}";
        callback(cookie, vbmap.data(), vbmap.length());

        return ENGINE_SUCCESS;
    }

    static void handle_disconnect(const void* cookie,
                                  ENGINE_EVENT_TYPE type,
                                  const void* event_data,
                                  const void* cb_data) {
        cb_assert(event_data == NULL);
        EWB_Engine* ewb =
                reinterpret_cast<EWB_Engine*>(const_cast<void*>(cb_data));
        auto logger = ewb->gsa()->log->get_logger();
        logger->log(EXTENSION_LOG_DEBUG, NULL, "EWB_Engine::handle_disconnect");

        uint64_t id = ewb->get_connection_id(cookie);
        {
            std::lock_guard<std::mutex> guard(ewb->cookie_map_mutex);
            ewb->connection_map.erase(id);
        }
    }

    GET_SERVER_API gsa;
    union {
        engine_info eng_info;
        char buffer[sizeof(engine_info) +
                    (sizeof(feature_info) * LAST_REGISTERED_ENGINE_FEATURE)];
    } info;

    // Actual engine we are proxying requests to.
    ENGINE_HANDLE* real_handle;
    ENGINE_HANDLE_V1* real_engine;
    engine_reference* real_engine_ref;

    std::atomic_int clustermap_revno;

    /**
     * The method responsible for pushing all of the notify_io_complete
     * to the frontend. It is run by notify_io_thread and not intended to
     * be called by anyone else!.
     */
    void process_notifications();
{'fraction_non_alphanumeric': 0.08667784733407502, 'fraction_numerical': 0.003337595270909785, 'mean_word_length': 1.5859418609644038, 'pattern_counts': {'":': 2, '<': 76, '<?xml version=': 0, '>': 385, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 323, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5318151', 'n_tokens_mistral': 16261, 'n_tokens_neox': 15392, 'n_words': 3887}