protocol.cpp

/*
 * Copyright (C) 2011-2026 Redis Labs Ltd.
 *
 * This file is part of memtier_benchmark.
 *
 * memtier_benchmark is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, version 2.
 *
 * memtier_benchmark is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with memtier_benchmark.  If not, see <http://www.gnu.org/licenses/>.
 */

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include <stdlib.h>
#include <errno.h>
#include <limits.h>
#include <stdio.h>
#include <string.h>
#include <deque>
#ifdef HAVE_ASSERT_H
#include <assert.h>
#endif

#include "protocol.h"
#include "keylist_growth.h"
#include "memtier_benchmark.h"
#include "libmemcached_protocol/binary.h"

/////////////////////////////////////////////////////////////////////////

abstract_protocol::abstract_protocol() :
        m_read_buf(NULL), m_write_buf(NULL), m_keep_value(false), m_track_elem_misses(false)
{
}

abstract_protocol::~abstract_protocol() {}

void abstract_protocol::set_buffers(struct evbuffer *read_buf, struct evbuffer *write_buf)
{
    m_read_buf = read_buf;
    m_write_buf = write_buf;
}

void abstract_protocol::set_keep_value(bool flag)
{
    m_keep_value = flag;
}

void abstract_protocol::set_track_elem_misses(bool flag)
{
    m_track_elem_misses = flag;
}

void abstract_protocol::reset_state()
{
    // Default: just clear m_last_response so any allocated mbulk tree /
    // status / value strings are freed before the next connection reuses
    // this object. Subclasses with their own parser-cursor fields (see
    // redis_protocol) override to additionally reset those.
    m_last_response.clear();
}

/////////////////////////////////////////////////////////////////////////

protocol_response::protocol_response() :
        m_status(NULL),
        m_mbulk_value(NULL),
        m_value(NULL),
        m_value_len(0),
        m_hits(0),
        m_error(false),
        m_top_array_len(-1)
{
}

protocol_response::~protocol_response()
{
    clear();
}

void protocol_response::set_error()
{
    m_error = true;
}

bool protocol_response::is_error(void)
{
    return m_error;
}

void protocol_response::set_status(const char *status)
{
    if (m_status != NULL) free((void *) m_status);
    m_status = status;
}

const char *protocol_response::get_status(void)
{
    return m_status;
}

void protocol_response::set_value(const char *value, unsigned int value_len)
{
    if (m_value != NULL) free((void *) m_value);
    m_value = value;
    m_value_len = value_len;
}

const char *protocol_response::get_value(unsigned int *value_len)
{
    assert(value_len != NULL);

    *value_len = m_value_len;
    return m_value;
}

void protocol_response::set_total_len(unsigned int total_len)
{
    m_total_len = total_len;
}

unsigned int protocol_response::get_total_len(void)
{
    return m_total_len;
}

void protocol_response::incr_hits(void)
{
    m_hits++;
}

unsigned int protocol_response::get_hits(void)
{
    return m_hits;
}

void protocol_response::set_top_array_len(int len)
{
    m_top_array_len = len;
}

int protocol_response::get_top_array_len(void)
{
    return m_top_array_len;
}

void protocol_response::elem_hits_reserve(size_t n)
{
    m_elem_hits.reserve(n);
}

void protocol_response::elem_hit_push(bool hit)
{
    m_elem_hits.push_back(hit ? 1 : 0);
}

const std::vector<uint8_t> &protocol_response::get_elem_hits(void)
{
    return m_elem_hits;
}

void protocol_response::clear(void)
{
    if (m_status != NULL) {
        free((void *) m_status);
        m_status = NULL;
    }
    if (m_value != NULL) {
        free((void *) m_value);
        m_value = NULL;
    }
    if (m_mbulk_value != NULL) {
        delete m_mbulk_value;
        m_mbulk_value = NULL;
    }

    m_value_len = 0;
    m_total_len = 0;
    m_hits = 0;
    m_error = 0;
    m_top_array_len = -1;
    m_elem_hits.clear();
}

void protocol_response::set_mbulk_value(mbulk_size_el *element)
{
    m_mbulk_value = element;
}

mbulk_size_el *protocol_response::get_mbulk_value()
{
    return m_mbulk_value;
}

/////////////////////////////////////////////////////////////////////////

class redis_protocol : public abstract_protocol
{
protected:
    enum response_state
    {
        rs_initial,
        rs_read_bulk,
        rs_read_line,
        rs_end_bulk
    };
    response_state m_response_state;
    long m_bulk_len;
    size_t m_response_len;

    unsigned int m_total_bulks_count;
    mbulk_size_el *m_current_mbulk;
    bool m_resp3;
    bool m_attribute;
    // Drain mode for RESP3 server-pushed frames ('>'). Pushes are
    // out-of-band (pubsub, keyspace notifications, client tracking
    // invalidations) and must not be delivered as a reply to any
    // in-flight command. While true the parser still walks the frame
    // to advance the read buffer, but suppresses all writes into
    // m_last_response (status, mbulk tree, value, hits).
    // TODO: full out-of-band routing — surface push frames to a
    // listener instead of silently dropping them.
    bool m_push;
    // Per-response state for alloc-free ArrayPerElementNulls miss tracking
    // (m_track_elem_misses). m_tracking turns true once the reply's top-level
    // array header is seen. m_nested_remaining counts leaf/element slots still
    // to consume inside the current top-level element's subtree: while >0 we
    // are below the top level and skip recording; when it returns to 0 the next
    // element is again a direct child of the top array. This handles arbitrary
    // nesting depth without a frame stack.
    bool m_tracking;
    int m_nested_remaining;

    bool aggregate_type(char c);
    bool blob_type(char c);
    bool single_type(char c);
    bool response_ended();
    // Record one leaf element (bulk / single type) for alloc-free miss
    // tracking: pushes a hit/miss flag when the leaf is a direct child of the
    // top-level array, otherwise just consumes a slot of the current nested
    // subtree. No-op unless tracking is active for this reply.
    inline void track_leaf(bool hit)
    {
        if (!m_track_elem_misses || m_push || !m_tracking) return;
        if (m_nested_remaining == 0)
            m_last_response.elem_hit_push(hit);
        else
            m_nested_remaining--;
    }

public:
    redis_protocol() :
            m_response_state(rs_initial),
            m_bulk_len(0),
            m_response_len(0),
            m_total_bulks_count(0),
            m_current_mbulk(NULL),
            m_resp3(false),
            m_attribute(false),
            m_push(false),
            m_tracking(false),
            m_nested_remaining(0)
    {
    }
    virtual redis_protocol *clone(void) { return new redis_protocol(); }
    virtual void reset_state();
    virtual int select_db(int db);
    virtual int authenticate(const char *credentials);
    virtual int configure_protocol(enum PROTOCOL_TYPE type);
    virtual int write_command_cluster_slots();
    virtual int write_command_readonly();
    virtual int write_command_set(const char *key, int key_len, const char *value, int value_len, int expiry,
                                  unsigned int offset);
    virtual int write_command_get(const char *key, int key_len, unsigned int offset);
    virtual int write_command_multi_get(const keylist *keylist);
    virtual int write_command_wait(unsigned int num_slaves, unsigned int timeout);
    virtual int parse_response(void);

    // handle arbitrary command
    virtual bool format_arbitrary_command(arbitrary_command &cmd);
    int write_arbitrary_command(const command_arg *arg);
    int write_arbitrary_command(const char *val, int val_len);
};

void redis_protocol::reset_state()
{
    // Discard partial-parse state from a previous connection. Without this,
    // a TCP RST received mid-bulk under RESP3 (e.g. while draining a server
    // push frame) leaves m_response_state at rs_read_bulk / rs_end_bulk and
    // m_total_bulks_count / m_push / m_attribute mid-walk. The next
    // bufferevent attached to this object would then parse fresh bytes
    // under stale cursor state and either lose framing or silently drop a
    // reply.
    //
    // m_last_response.clear() also frees any partially-built mbulk tree
    // hanging off m_current_mbulk (the tree's root is owned by
    // m_last_response::m_mbulk_value), so we just null the cursor here.
    //
    // NB: m_resp3 is intentionally preserved -- it reflects the negotiated
    // protocol version from HELLO, not parser cursor state. A reconnect
    // re-runs HELLO via the setup ladder, but we should not roll back the
    // observed value here.
    m_last_response.clear();
    m_response_state = rs_initial;
    m_total_bulks_count = 0;
    m_attribute = false;
    m_push = false;
    m_bulk_len = 0;
    m_response_len = 0;
    m_current_mbulk = NULL;
    m_tracking = false;
    m_nested_remaining = 0;
}

int redis_protocol::select_db(int db)
{
    int size = 0;
    char db_str[20];

    snprintf(db_str, sizeof(db_str) - 1, "%d", db);
    size = evbuffer_add_printf(m_write_buf,
                               "*2\r\n"
                               "$6\r\n"
                               "SELECT\r\n"
                               "$%u\r\n"
                               "%s\r\n",
                               (unsigned int) strlen(db_str), db_str);
    return size;
}

int redis_protocol::authenticate(const char *credentials)
{
    int size = 0;
    assert(credentials != NULL);

    /* Credentials may be one of:
     * <PASSWORD>           For Redis <6.0 simple AUTH commands.
     * <USER>:<PASSWORD>    For Redis 6.0+ AUTH with both username and password.
     *
     * A :<PASSWORD> will be handled as a special case of quoting a password that
     * contains a colon.
     */

    const char *user = NULL;
    const char *password;

    if (credentials[0] == ':') {
        password = credentials + 1;
    } else {
        password = strchr(credentials, ':');
        if (!password) {
            password = credentials;
        } else {
            user = credentials;
            password++;
        }
    }

    if (!user) {
        size = evbuffer_add_printf(m_write_buf,
                                   "*2\r\n"
                                   "$4\r\n"
                                   "AUTH\r\n"
                                   "$%llu\r\n"
                                   "%s\r\n",
                                   (unsigned long long) strlen(password), password);
    } else {
        size_t user_len = password - user - 1;
        size = evbuffer_add_printf(m_write_buf,
                                   "*3\r\n"
                                   "$4\r\n"
                                   "AUTH\r\n"
                                   "$%llu\r\n"
                                   "%.*s\r\n"
                                   "$%llu\r\n"
                                   "%s\r\n",
                                   (unsigned long long) user_len, (int) user_len, user,
                                   (unsigned long long) strlen(password), password);
    }
    return size;
}

int redis_protocol::configure_protocol(enum PROTOCOL_TYPE type)
{
    int size = 0;
    if (type == PROTOCOL_RESP2 || type == PROTOCOL_RESP3) {
        m_resp3 = type == PROTOCOL_RESP3;
        size = evbuffer_add_printf(m_write_buf,
                                   "*2\r\n"
                                   "$5\r\n"
                                   "HELLO\r\n"
                                   "$1\r\n"
                                   "%d\r\n",
                                   type == PROTOCOL_RESP2 ? 2 : 3);
    }
    return size;
}

int redis_protocol::write_command_cluster_slots()
{
    int size = 0;

    size = evbuffer_add(m_write_buf,
                        "*2\r\n"
                        "$7\r\n"
                        "CLUSTER\r\n"
                        "$5\r\n"
                        "SLOTS\r\n",
                        28);

    return size;
}

int redis_protocol::write_command_readonly()
{
    int size = 0;

    size = evbuffer_add(m_write_buf,
                        "*1\r\n"
                        "$8\r\n"
                        "READONLY\r\n",
                        16);

    return size;
}

int redis_protocol::write_command_set(const char *key, int key_len, const char *value, int value_len, int expiry,
                                      unsigned int offset)
{
    assert(key != NULL);
    assert(key_len > 0);
    assert(value != NULL);
    assert(value_len > 0);
    int size = 0;

    if (!expiry && !offset) {
        size = evbuffer_add_printf(m_write_buf,
                                   "*3\r\n"
                                   "$3\r\n"
                                   "SET\r\n"
                                   "$%u\r\n",
                                   key_len);
        evbuffer_add(m_write_buf, key, key_len);
        size += key_len;
        size += evbuffer_add_printf(m_write_buf,
                                    "\r\n"
                                    "$%u\r\n",
                                    value_len);
    } else if (offset) {
        char offset_str[30];
        snprintf(offset_str, sizeof(offset_str) - 1, "%u", offset);

        size = evbuffer_add_printf(m_write_buf,
                                   "*4\r\n"
                                   "$8\r\n"
                                   "SETRANGE\r\n"
                                   "$%u\r\n",
                                   key_len);
        evbuffer_add(m_write_buf, key, key_len);
        size += key_len;
        size += evbuffer_add_printf(m_write_buf,
                                    "\r\n"
                                    "$%u\r\n"
                                    "%s\r\n"
                                    "$%u\r\n",
                                    (unsigned int) strlen(offset_str), offset_str, value_len);
    } else {
        char expiry_str[30];
        snprintf(expiry_str, sizeof(expiry_str) - 1, "%u", expiry);

        size = evbuffer_add_printf(m_write_buf,
                                   "*4\r\n"
                                   "$5\r\n"
                                   "SETEX\r\n"
                                   "$%u\r\n",
                                   key_len);
        evbuffer_add(m_write_buf, key, key_len);
        size += key_len;
        size += evbuffer_add_printf(m_write_buf,
                                    "\r\n"
                                    "$%u\r\n"
                                    "%s\r\n"
                                    "$%u\r\n",
                                    (unsigned int) strlen(expiry_str), expiry_str, value_len);
    }
    evbuffer_add(m_write_buf, value, value_len);
    evbuffer_add(m_write_buf, "\r\n", 2);
    size += value_len + 2;

    return size;
}

int redis_protocol::write_command_multi_get(const keylist *keylist)
{
    assert(keylist != NULL);
    assert(keylist->get_keys_count() > 0);

    int size = 0;
    unsigned int keys_count = keylist->get_keys_count();

    // *<1+keys_count>\r\n$4\r\nMGET\r\n
    size = evbuffer_add_printf(m_write_buf,
                               "*%u\r\n"
                               "$4\r\n"
                               "MGET\r\n",
                               1 + keys_count);

    for (unsigned int i = 0; i < keys_count; i++) {
        unsigned int key_len = 0;
        const char *key = keylist->get_key(i, &key_len);
        assert(key != NULL);

        size += evbuffer_add_printf(m_write_buf, "$%u\r\n", key_len);
        evbuffer_add(m_write_buf, key, key_len);
        evbuffer_add(m_write_buf, "\r\n", 2);
        size += key_len + 2;
    }

    return size;
}

int redis_protocol::write_command_get(const char *key, int key_len, unsigned int offset)
{
    assert(key != NULL);
    assert(key_len > 0);
    int size = 0;

    if (!offset) {
        size = evbuffer_add_printf(m_write_buf,
                                   "*2\r\n"
                                   "$3\r\n"
                                   "GET\r\n"
                                   "$%u\r\n",
                                   key_len);
        evbuffer_add(m_write_buf, key, key_len);
        evbuffer_add(m_write_buf, "\r\n", 2);
        size += key_len + 2;
    } else {
        char offset_str[30];
        snprintf(offset_str, sizeof(offset_str) - 1, "%u", offset);

        size = evbuffer_add_printf(m_write_buf,
                                   "*4\r\n"
                                   "$8\r\n"
                                   "GETRANGE\r\n"
                                   "$%u\r\n",
                                   key_len);
        evbuffer_add(m_write_buf, key, key_len);
        size += key_len;
        size += evbuffer_add_printf(m_write_buf,
                                    "\r\n"
                                    "$%u\r\n"
                                    "%s\r\n"
                                    "$2\r\n"
                                    "-1\r\n",
                                    (unsigned int) strlen(offset_str), offset_str);
    }

    return size;
}

/*
 * Utility function to get the number of digits in a number
 */
static int get_number_length(unsigned int num)
{
    if (num < 10) return 1;
    if (num < 100) return 2;
    if (num < 1000) return 3;
    if (num < 10000) return 4;
    if (num < 100000) return 5;
    if (num < 1000000) return 6;
    if (num < 10000000) return 7;
    if (num < 100000000) return 8;
    if (num < 1000000000) return 9;
    return 10;
}

int redis_protocol::write_command_wait(unsigned int num_slaves, unsigned int timeout)
{
    int size = 0;
    size = evbuffer_add_printf(m_write_buf,
                               "*3\r\n"
                               "$4\r\n"
                               "WAIT\r\n"
                               "$%u\r\n"
                               "%u\r\n"
                               "$%u\r\n"
                               "%u\r\n",
                               get_number_length(num_slaves), num_slaves, get_number_length(timeout), timeout);
    return size;
}

bool redis_protocol::aggregate_type(char c)
{
    if (c == '*') return true;

    // RESP3: map (%), set (~), attribute (|), and push (>) are all aggregate
    // types.  Push frames (">N\r\n" followed by N elements) are structurally
    // identical to arrays and can be emitted by Redis 7+ at any point on a
    // RESP3 connection (pubsub, keyspace notifications, client-tracking
    // invalidation).  We parse them as ordinary aggregates and discard the
    // content; the alternative -- returning -1 from parse_response -- tears
    // down the connection unnecessarily.
    // TODO: route drained push frames to an out-of-band callback so callers
    // can observe server-initiated events without disrupting the reply stream.
    if (m_resp3 && (c == '%' || c == '~' || c == '|' || c == '>')) return true;

    return false;
}

bool redis_protocol::blob_type(char c)
{
    if (c == '$') return true;

    if (m_resp3 && (c == '!' || c == '=')) return true;

    return false;
}

bool redis_protocol::single_type(char c)
{
    if (c == '+' || c == '-' || c == ':') return true;

    if (m_resp3 && (c == '_' || c == ',' || c == '#' || c == '(')) return true;

    return false;
}

bool redis_protocol::response_ended()
{
    if (m_total_bulks_count != 0) return false;

    bool was_push = m_push;
    bool was_attribute = m_attribute;

    if (was_attribute) m_attribute = false;
    // A RESP3 push frame just finished draining. The actual reply to
    // the in-flight command (if any) is still pending — keep reading
    // and do NOT signal a complete reply for the push.
    // Also handles the case where an attribute (|) is nested inside a
    // push frame (>) and both complete at the same call: clear BOTH
    // flags together so m_push does not remain set and eat the next reply.
    if (was_push) {
        m_push = false;
        m_response_len = 0; // reset: push bytes don't count toward the next reply
    }

    if (was_push || was_attribute) return false;
    return true;
}

int redis_protocol::parse_response(void)
{
    char *line;
    size_t res_len;

    while (true) {
        switch (m_response_state) {
        case rs_initial:
            // clear last response
            m_last_response.clear();
            m_response_len = 0;
            m_total_bulks_count = 0;
            m_attribute = 0;
            m_push = 0;
            m_tracking = false;
            m_nested_remaining = 0;
            m_response_state = rs_read_line;

            break;
        case rs_read_line:
            line = evbuffer_readln(m_read_buf, &res_len, EVBUFFER_EOL_CRLF_STRICT);

            // maybe we didn't get it yet?
            if (line == NULL) {
                return 0;
            }

            // count CRLF
            m_response_len += res_len + 2;

            if (aggregate_type(line[0])) {
                int count = strtol(line + 1, NULL, 10);

                // The reply's top-level aggregate is the one parsed while no
                // bulks are still outstanding. Capture its declared length so
                // miss-tracking can tell an empty collection from a populated
                // one without materializing values (see EmptyCollection in
                // client::handle_response). Out-of-band frames are excluded:
                // attribute (|) is a prefix to the real reply, push (>) is not
                // a reply at all.
                bool top_level_aggregate = (m_total_bulks_count == 0);

                // in case of nested mbulk, the mbulk is one of the total bulks
                if (m_total_bulks_count > 0) {
                    m_total_bulks_count--;
                }

                // from bulks counter perspective every count < 0 is equal to 0, because it's not followed by bulks.
                if (count < 0) {
                    count = 0;
                }

                if (line[0] == '|') {
                    // Nested attribute?
                    assert(!m_attribute);

                    m_attribute = true;
                }

                if (line[0] == '>') {
                    // Top-level push frame: enter drain mode. Nested
                    // pushes (push inside push) are not defined by RESP3
                    // and not expected; only set the flag once.
                    m_push = true;
                }

                // Map or Attribute contain key-value pair
                if (line[0] == '%' || line[0] == '|') {
                    count *= 2;
                }

                // Record the reply body's declared element count (count is now
                // null-clamped to >=0 and map-doubled). '|' is the attribute
                // prefix, not the reply body; '>' (m_push) is out-of-band.
                if (top_level_aggregate && !m_push && line[0] != '|') {
                    m_last_response.set_top_array_len(count);
                }

                // Alloc-free per-element miss tracking (m_track_elem_misses).
                // The top-level array container starts tracking; a nested
                // aggregate is itself an element -- a hit iff it is non-empty
                // (count>0). See the m_tracking/m_nested_remaining contract.
                if (m_track_elem_misses && !m_push && line[0] != '|') {
                    if (top_level_aggregate) {
                        m_tracking = true;
                        m_nested_remaining = 0;
                        m_last_response.elem_hits_reserve(count);
                    } else if (m_tracking) {
                        if (m_nested_remaining == 0) {
                            m_last_response.elem_hit_push(count > 0);
                            m_nested_remaining = count; // descend into this element's subtree
                        } else {
                            m_nested_remaining += count - 1; // consume this slot, add its children
                        }
                    }
                }

                // Suppress mbulk allocation while draining a push frame:
                // the contents are out-of-band and must not be delivered
                // as the reply to any in-flight command.
                if (m_keep_value && !m_push) {
                    mbulk_size_el *new_mbulk_size = new mbulk_size_el();
                    new_mbulk_size->bulks_count = count;
                    new_mbulk_size->upper_level = m_current_mbulk;

                    // update first mbulk as the response mbulk, or insert it to current mbulk
                    if (m_last_response.get_mbulk_value() == NULL) {
                        m_last_response.set_mbulk_value(new_mbulk_size);
                    } else {
                        m_current_mbulk->add_new_element(new_mbulk_size);
                    }

                    // update current mbulk
                    m_current_mbulk = new_mbulk_size->get_next_mbulk();
                }

                if (!m_push) {
                    m_last_response.set_status(line);
                } else {
                    free(line);
                }
                m_total_bulks_count += count;

                if (response_ended()) {
                    m_last_response.set_total_len(m_response_len);
                    m_response_state = rs_initial;
                    return 1;
                }
            } else if (blob_type(line[0])) {
                // if it's single bulk (not part of mbulk), we count it here
                if (m_total_bulks_count == 0) {
                    m_total_bulks_count++;
                }

                m_bulk_len = strtol(line + 1, NULL, 10);
                // A bulk leaf is a hit iff it carries a value ($N, N>0). A null
                // bulk ($-1) and an empty bulk ($0) are misses -- matching the
                // pre-existing value!=NULL convention of the materialized walk.
                track_leaf(m_bulk_len > 0);
                // Suppress reply mutation while draining a push frame.
                if (!m_push) {
                    m_last_response.set_status(line);

                    if (line[0] == '!') m_last_response.set_error();
                } else {
                    free(line);
                }

                /*
                 * only on negative bulk, the data ends right after the first CRLF ($-1\r\n), so
                 * we skip on rs_read_bulk and jump into rs_end_bulk
                 */
                if (m_bulk_len < 0) {
                    m_response_state = rs_end_bulk;
                } else {
                    m_response_state = rs_read_bulk;
                }
            } else if (single_type(line[0])) {
                // if it's single bulk (not part of mbulk), we count it here
                if (m_total_bulks_count == 0) {
                    m_total_bulks_count++;
                }

                // A single-type leaf carries a value (+status, :int, ,double,
                // #bool, (bignum, -error) and counts as a hit; only the RESP3
                // null '_' is a miss. Mirrors the materialized walk, which kept
                // a value for every single type except '_' (is_resp3_null).
                track_leaf(line[0] != '_');

                // if we are not inside mbulk, the status will be kept in m_status anyway
                if (m_keep_value && m_current_mbulk && !m_push) {
                    char *bulk_value = strdup(line);
                    assert(bulk_value != NULL);

                    bulk_el *new_bulk = new bulk_el();
                    new_bulk->value = bulk_value;
                    new_bulk->value_len = strlen(bulk_value);
                    // Mark RESP3 null elements explicitly so the walker can
                    // distinguish them from a legitimate bulk string whose
                    // content happens to be the single character '_' (which
                    // would arrive via blob_type / rs_read_bulk, not here).
                    new_bulk->is_resp3_null = (line[0] == '_');

                    // insert it to current mbulk
                    m_current_mbulk->add_new_element(new_bulk);
                    m_current_mbulk = m_current_mbulk->get_next_mbulk();
                }

                if (!m_push) {
                    if (line[0] == '-') m_last_response.set_error();

                    m_last_response.set_status(line);
                } else {
                    free(line);
                }
                m_total_bulks_count--;

                if (response_ended()) {
                    m_last_response.set_total_len(m_response_len);
                    m_response_state = rs_initial;
                    return 1;
                }
            } else {
                benchmark_debug_log("unsupported response: '%s'.\n", line);
                free(line);
                return -1;
            }
            break;
        case rs_read_bulk:
            if (evbuffer_get_length(m_read_buf) >= (unsigned long) (m_bulk_len + 2)) {
                m_response_len += m_bulk_len + 2;

                /*
                 * KNOWN ISSUE:
                 * in case of key with zero size (SET X "") that will return $0,
                 * we are not counting it as "hit", and the report will be wrong.
                 * currently this is a limitation because GETRANGE returns $0 for
                 * such key as well as non existing key or existing key without data
                 * in the requested range
                 */
                // Suppress hit accounting while draining a push frame; the
                // bulk belongs to an out-of-band message, not the reply.
                if (m_bulk_len > 0 && !m_push) {
                    m_last_response.incr_hits();
                }

                m_response_state = rs_end_bulk;
            } else {
                return 0;
            }
            break;
        case rs_end_bulk:
            // Push-frame drain: read+discard the bulk bytes without
            // touching m_last_response so the reply remains clean.
            if (m_push) {
                if (m_bulk_len >= 0) {
                    int ret = evbuffer_drain(m_read_buf, m_bulk_len + 2);
                    assert(ret != -1);
                }
            } else if (m_keep_value) {
                /*
                 * keep bulk value - in case we need to save bulk value it depends
                 * if it's inside a mbulk or not.
                 * in case of receiving just bulk as a response, we save it directly to the m_last_response,
                 * otherwise we insert it to the current mbulk element
                 */
                char *bulk_value = NULL;
                int ret;
                if (m_bulk_len > 0) {
                    bulk_value = (char *) malloc(m_bulk_len);
                    assert(bulk_value != NULL);

                    ret = evbuffer_remove(m_read_buf, bulk_value, m_bulk_len);
                    assert(ret != -1);
                }

                // drain last CRLF, zero bulk also includes it ($0\r\n\r\n)
                if (m_bulk_len >= 0) {
                    ret = evbuffer_drain(m_read_buf, 2);
                    assert(ret != -1);
                }

                // in case we are inside mbulk
                if (m_current_mbulk) {
                    bulk_el *new_bulk = new bulk_el();
                    new_bulk->value = bulk_value;
                    // negative bulk len counted as empty bulk
                    new_bulk->value_len = m_bulk_len > 0 ? m_bulk_len : 0;

                    // insert it to current mbulk
                    m_current_mbulk->add_new_element(new_bulk);
                    m_current_mbulk = m_current_mbulk->get_next_mbulk();
                } else {
                    // negative bulk len counted as empty bulk
                    m_last_response.set_value(bulk_value, m_bulk_len > 0 ? m_bulk_len : 0);
                }
            } else {
                // just drain the buffer, include the CRLF
                if (m_bulk_len >= 0) {
                    int ret = evbuffer_drain(m_read_buf, m_bulk_len + 2);
                    assert(ret != -1);
                }
            }

            m_total_bulks_count--;

            if (response_ended()) {
                m_last_response.set_total_len(m_response_len);
                m_response_state = rs_initial;
                return 1;
            } else {
                m_response_state = rs_read_line;
            }
            break;
        default:
            return -1;
        }
    }

    return -1;
}

int redis_protocol::write_arbitrary_command(const command_arg *arg)
{
    evbuffer_add(m_write_buf, arg->data.c_str(), arg->data.length());

    return arg->data.length();
}

int redis_protocol::write_arbitrary_command(const char *rand_val, int rand_val_len)
{
    int size = 0;

    size = evbuffer_add_printf(m_write_buf, "$%d\r\n", rand_val_len);
    evbuffer_add(m_write_buf, rand_val, rand_val_len);
    size += rand_val_len;
    evbuffer_add(m_write_buf, "\r\n", 2);
    size += 2;

    return size;
}

bool redis_protocol::format_arbitrary_command(arbitrary_command &cmd)
{
    for (unsigned int i = 0; i < cmd.command_args.size(); i++) {
        command_arg *current_arg = &cmd.command_args[i];
        current_arg->type = const_type;

        // check arg type
        if (current_arg->scan_cursor) {
            current_arg->type = scan_cursor_type;
        } else if (current_arg->resolved) {
            // Already filled in by client::resolve_monitor_placeholders():
            // frame the generated key as-is, without scanning it again.
        } else if (current_arg->data.find(KEY_PLACEHOLDER) != std::string::npos) {
            if (current_arg->data.length() != strlen(KEY_PLACEHOLDER)) {
                current_arg->has_key_affixes = true;
                current_arg->data_prefix = current_arg->data.substr(0, current_arg->data.find(KEY_PLACEHOLDER));
                current_arg->data_suffix =
                    current_arg->data.substr(current_arg->data.find(KEY_PLACEHOLDER) + strlen(KEY_PLACEHOLDER));
            }
            cmd.keys_count++;
            current_arg->type = key_type;
        } else if (current_arg->data.find(DATA_PLACEHOLDER) != std::string::npos) {
            if (current_arg->data.length() != strlen(DATA_PLACEHOLDER)) {
                benchmark_error_log("error: data placeholder can't combined with other data\n");
                return false;
            }

            current_arg->type = data_type;
        }

        // we expect that first arg is the COMMAND name
        assert(i != 0 || (i == 0 && current_arg->type == const_type && "first arg is not command name?"));

        if (current_arg->type == const_type) {
            char buffer[20];
            int buffer_len;

            // if it's first arg we add also the mbulk size
            if (i == 0) {
                buffer_len =
                    snprintf(buffer, 20, "*%zd\r\n$%zd\r\n", cmd.command_args.size(), current_arg->data.length());
            } else {
                buffer_len = snprintf(buffer, 20, "$%zd\r\n", current_arg->data.length());
            }

            current_arg->data.insert(0, buffer, buffer_len);
            current_arg->data += "\r\n";
        }
    }

    return true;
}

/////////////////////////////////////////////////////////////////////////

class memcache_text_protocol : public abstract_protocol
{
protected:
    enum response_state
    {
        rs_initial,
        rs_read_section,
        rs_read_value,
        rs_read_end
    };
    response_state m_response_state;
    unsigned int m_value_len;
    size_t m_response_len;

public:
    memcache_text_protocol() : m_response_state(rs_initial), m_value_len(0), m_response_len(0) {}
    virtual memcache_text_protocol *clone(void) { return new memcache_text_protocol(); }
    virtual int select_db(int db);
    virtual int authenticate(const char *credentials);
    virtual int configure_protocol(enum PROTOCOL_TYPE type);
    virtual int write_command_cluster_slots();
    virtual int write_command_readonly();
    virtual int write_command_set(const char *key, int key_len, const char *value, int value_len, int expiry,
                                  unsigned int offset);
    virtual int write_command_get(const char *key, int key_len, unsigned int offset);
    virtual int write_command_multi_get(const keylist *keylist);
    virtual int write_command_wait(unsigned int num_slaves, unsigned int timeout);
    virtual int parse_response(void);

    // handle arbitrary command
    virtual bool format_arbitrary_command(arbitrary_command &cmd);
    virtual int write_arbitrary_command(const command_arg *arg);
    virtual int write_arbitrary_command(const char *val, int val_len);
};

int memcache_text_protocol::select_db(int db)
{
    assert(0);
    return -1;
}

int memcache_text_protocol::authenticate(const char *credentials)
{
    assert(0);
    return -1;
}

int memcache_text_protocol::configure_protocol(enum PROTOCOL_TYPE type)
{
    assert(0);
    return -1;
}

int memcache_text_protocol::write_command_cluster_slots()
{
    assert(0);
    return -1;
}

int memcache_text_protocol::write_command_readonly()
{
    assert(0);
    return -1;
}

int memcache_text_protocol::write_command_set(const char *key, int key_len, const char *value, int value_len,
                                              int expiry, unsigned int offset)
{
    assert(key != NULL);
    assert(key_len > 0);
    assert(value != NULL);
    assert(value_len > 0);
    int size = 0;

    size = evbuffer_add_printf(m_write_buf, "set %.*s 0 %u %u\r\n", key_len, key, expiry, value_len);
    evbuffer_add(m_write_buf, value, value_len);
    evbuffer_add(m_write_buf, "\r\n", 2);
    size += value_len + 2;

    return size;
}

int memcache_text_protocol::write_command_get(const char *key, int key_len, unsigned int offset)
{
    assert(key != NULL);
    assert(key_len > 0);
    int size = 0;

    size = evbuffer_add_printf(m_write_buf, "get %.*s\r\n", key_len, key);
    return size;
}

int memcache_text_protocol::write_command_multi_get(const keylist *keylist)
{
    assert(keylist != NULL);
    assert(keylist->get_keys_count() > 0);

    int n = 0;
    int size = 0;

    n = evbuffer_add(m_write_buf, "get", 3);
    assert(n != -1);
    size = 3;

    for (unsigned int i = 0; i < keylist->get_keys_count(); i++) {
        const char *key;
        unsigned int key_len = 0;

        n = evbuffer_add(m_write_buf, " ", 1);
        assert(n != -1);
        size++;

        key = keylist->get_key(i, &key_len);
        assert(key != NULL);

        n = evbuffer_add(m_write_buf, key, key_len);
        assert(n != -1);
        size += key_len;
    }

    n = evbuffer_add(m_write_buf, "\r\n", 2);
    assert(n != -1);
    size += 2;

    return size;
}

int memcache_text_protocol::write_command_wait(unsigned int num_slaves, unsigned int timeout)
{
    fprintf(stderr, "error: WAIT command not implemented for memcache!\n");
    assert(0);
    return -1;
}

int memcache_text_protocol::parse_response(void)
{
    char *line;
    size_t tmplen;

    while (true) {
        switch (m_response_state) {
        case rs_initial:
            m_last_response.clear();
            m_response_state = rs_read_section;
            m_response_len = 0;
            break;

        case rs_read_section:
            line = evbuffer_readln(m_read_buf, &tmplen, EVBUFFER_EOL_CRLF_STRICT);
            if (!line) return 0;

            m_response_len += tmplen + 2; // For CRLF
            if (m_last_response.get_status() == NULL) {
                m_last_response.set_status(line);
            }
            m_last_response.set_total_len((unsigned int) m_response_len); // for now...

            if (memcmp(line, "VALUE", 5) == 0) {
                char prefix[50];
                char key[256];
                unsigned int flags;
                unsigned int cas;

                // Width specifiers cap %s at sizeof(buf)-1; a malicious/buggy
                // memcached server sending "VALUE <very-long-key> ..." cannot
                // overflow `key[256]` or `prefix[50]` and corrupt the worker
                // stack. Same class of bug as the VLA in arbitrary_command
                // (PR #405).
                int res = sscanf(line, "%49s %255s %u %u %u", prefix, key, &flags, &m_value_len, &cas);
                if (res < 4 || res > 5) {
                    benchmark_debug_log("unexpected VALUE response: %s\n", line);
                    if (m_last_response.get_status() != line) free(line);
                    return -1;
                }

                m_response_state = rs_read_value;
                continue;
            } else if (memcmp(line, "END", 3) == 0 || memcmp(line, "STORED", 6) == 0) {
                if (m_last_response.get_status() != line) free(line);
                m_response_state = rs_read_end;
                break;
            } else {
                m_last_response.set_error();
                benchmark_debug_log("unknown response: %s\n", line);
                return -1;
            }
            break;

        case rs_read_value:
            if (evbuffer_get_length(m_read_buf) >= m_value_len + 2) {
                if (m_keep_value) {
                    char *value = (char *) malloc(m_value_len);
                    assert(value != NULL);

                    int ret = evbuffer_remove(m_read_buf, value, m_value_len);
                    assert(ret != -1); // evbuffer_remove returns bytes read on success, not 0

                    m_last_response.set_value(value, m_value_len);
                } else {
                    int ret = evbuffer_drain(m_read_buf, m_value_len);
                    assert((unsigned int) ret == 0);
                }

                int ret = evbuffer_drain(m_read_buf, 2);
                assert((unsigned int) ret == 0);

                m_last_response.incr_hits();
                m_response_len += m_value_len + 2;
                m_response_state = rs_read_section;
            } else {
                return 0;
            }
            break;
        case rs_read_end:
            m_response_state = rs_initial;
            return 1;

        default:
            benchmark_debug_log("unknown response state %d.\n", m_response_state);
            return -1;
        }
    }

    return -1;
}

bool memcache_text_protocol::format_arbitrary_command(arbitrary_command &cmd)
{
    assert(0);
    return false;
}

int memcache_text_protocol::write_arbitrary_command(const command_arg *arg)
{
    assert(0);
    return -1;
}

int memcache_text_protocol::write_arbitrary_command(const char *val, int val_len)
{
    assert(0);
    return -1;
}

/////////////////////////////////////////////////////////////////////////
//
// Memcache Meta Protocol (memcached 1.6.10+).
//
// The meta protocol is a compact, flag-based ASCII protocol that supersedes
// the classic text protocol. We implement just enough to drive a representative
// SET / GET / multi-GET benchmark:
//
//   ms <key> <datalen> T<exp>\r\n<data>\r\n   ->  HD | NS | EX | NF
//   mg <key> v\r\n                            ->  VA <size>\r\n<data>\r\n | EN
//   <pipelined mg ...> mn\r\n                 ->  ...VA/EN..., MN  (multi-get)
//
// Inheriting from memcache_text_protocol lets us share the read/write buffers
// and the methods that are also unsupported on the text path
// (select_db / authenticate / configure_protocol / write_command_cluster_slots /
// write_command_wait / *arbitrary_command -- all assert(0)). Meta overrides
// its command writers, parser and reconnect reset.
//
// IMPORTANT design choice -- every command is terminated on the wire by `mn`,
// the meta protocol's no-op:
//   ms <k> <l> T<e>\r\n<data>\r\nmn\r\n   ->  HD\r\nMN\r\n
//   mg <k> v\r\nmn\r\n                  ->  VA <n>\r\n<data>\r\nMN\r\n  (or EN\r\nMN\r\n)
//   <pipelined mg ...> mn\r\n            ->  ...VA/EN..., MN  (mn is the natural terminator)
//
// The ordered expectation queue validates reply kinds and exact section counts;
// MN then marks the end of each logical request, including errors and multi-get.
// Keeping this boundary uniform is a deliberate wire-format choice, not a
// requirement of the queue. Single-key operations could omit mn in another mode.
// Each logical operation includes one extra server command and 4 request +
// 4 response bytes; its processing overhead depends on the workload.
class memcache_meta_protocol : public memcache_text_protocol
{
    struct expected_reply
    {
        bool get;
        unsigned int remaining;
    };
    // Validate before returning a completed response to shard_connection.
    // Raw serialized retries must restore this metadata through the replay hook.
    std::deque<expected_reply> m_expected_replies;

public:
    memcache_meta_protocol() {}
    virtual memcache_meta_protocol *clone(void) { return new memcache_meta_protocol(); }
    virtual void prepare_request_replay(bool is_get, unsigned int keys)
    {
        m_expected_replies.push_back({is_get, is_get ? keys : 1});
    }
    virtual void reset_state()
    {
        abstract_protocol::reset_state();
        m_response_state = rs_initial;
        m_value_len = 0;
        m_response_len = 0;
        m_expected_replies.clear();
    }

    virtual int write_command_set(const char *key, int key_len, const char *value, int value_len, int expiry,
                                  unsigned int offset);
    virtual int write_command_get(const char *key, int key_len, unsigned int offset);
    virtual int write_command_multi_get(const keylist *keylist);
    virtual int parse_response(void);
};

// Header / trailer literals factored out so that:
//   (a) `mg ... v\r\n` is built with three raw `evbuffer_add` calls instead of
//       a `vsnprintf`-driven `evbuffer_add_printf`. libevent appends each piece
//       to the tail chain as a single memcpy, which matches the existing
//       memcache_text_protocol multi-get write path.
//   (b) Trailer length is shared between get / multi-get so the math is
//       impossible to drift.
static const char k_mg_prefix[] = "mg ";
static const char k_mg_suffix[] = " v\r\n";
static const char k_mn_term[] = "mn\r\n";
static const size_t k_mg_prefix_len = sizeof(k_mg_prefix) - 1; // "mg "
static const size_t k_mg_suffix_len = sizeof(k_mg_suffix) - 1; // " v\r\n"
static const size_t k_mn_term_len = sizeof(k_mn_term) - 1;     // "mn\r\n"

int memcache_meta_protocol::write_command_set(const char *key, int key_len, const char *value, int value_len,
                                              int expiry, unsigned int offset)
{
    assert(key != NULL);
    assert(key_len > 0);
    assert(value != NULL);
    assert(value_len > 0);

    // `ms <key> <datalen> T<exp>\r\n<value>\r\nmn\r\n`. The trailing mn makes
    // every command's response self-terminating at MN. The expectation queue
    // also validates that this section is a storage reply.
    int size = evbuffer_add_printf(m_write_buf, "ms %.*s %u T%u\r\n", key_len, key, (unsigned int) value_len, expiry);
    evbuffer_add(m_write_buf, value, value_len);
    evbuffer_add(m_write_buf, "\r\n", 2);
    evbuffer_add(m_write_buf, k_mn_term, k_mn_term_len);
    m_expected_replies.push_back({false, 1});
    return size + value_len + 2 + (int) k_mn_term_len;
}

int memcache_meta_protocol::write_command_get(const char *key, int key_len, unsigned int offset)
{
    assert(key != NULL);
    assert(key_len > 0);
    evbuffer_add(m_write_buf, k_mg_prefix, k_mg_prefix_len);
    evbuffer_add(m_write_buf, key, key_len);
    evbuffer_add(m_write_buf, k_mg_suffix, k_mg_suffix_len);
    evbuffer_add(m_write_buf, k_mn_term, k_mn_term_len);
    m_expected_replies.push_back({true, 1});
    return (int) (k_mg_prefix_len + key_len + k_mg_suffix_len + k_mn_term_len);
}

int memcache_meta_protocol::write_command_multi_get(const keylist *keylist)
{
    assert(keylist != NULL);
    assert(keylist->get_keys_count() > 0);

    int size = 0;
    for (unsigned int i = 0; i < keylist->get_keys_count(); i++) {
        unsigned int key_len;
        const char *key = keylist->get_key(i, &key_len);
        assert(key != NULL);
        evbuffer_add(m_write_buf, k_mg_prefix, k_mg_prefix_len);
        evbuffer_add(m_write_buf, key, key_len);
        evbuffer_add(m_write_buf, k_mg_suffix, k_mg_suffix_len);
        size += (int) (k_mg_prefix_len + key_len + k_mg_suffix_len);
    }
    // Single mn at the end of the pipeline -- one terminator covers all keys
    // in this multi-get logical request, mirroring the per-command `mn`
    // suffix used by ms / mg above.
    evbuffer_add(m_write_buf, k_mn_term, k_mn_term_len);
    m_expected_replies.push_back({true, keylist->get_keys_count()});
    return size + (int) k_mn_term_len;
}

int memcache_meta_protocol::parse_response(void)
{
    char *line;
    size_t tmplen;

    while (true) {
        switch (m_response_state) {
        case rs_initial:
            m_last_response.clear();
            m_response_state = rs_read_section;
            m_response_len = 0;
            break;

        case rs_read_section: {
            line = evbuffer_readln(m_read_buf, &tmplen, EVBUFFER_EOL_CRLF_STRICT);
            if (!line) return 0;

            m_response_len += tmplen + 2; // For CRLF
            // Adopt the first status; a later defined error replaces it so
            // error accounting/retry sees the error rather than a prior hit.
            if (m_last_response.get_status() == NULL) {
                m_last_response.set_status(line);
            }
            m_last_response.set_total_len((unsigned int) m_response_len);

            // Decode the section header. Meta has very few shapes:
            //   VA <size> [<flags>...]   value follows on the next line
            //   MN                       end of a multi-op pipeline
            //   HD | EN | NS | EX | NF   2-letter status (stored / miss / not-stored / exists / not-found)
            bool need_value = false;
            bool done = false;
            bool unknown = false;
            bool server_error = false;

            if (memchr(line, '\0', tmplen) != NULL) {
                unknown = true;
            } else if ((tmplen == 5 && memcmp(line, "ERROR", 5) == 0) ||
                       (tmplen >= 12 && (tmplen == 12 || line[12] == ' ') &&
                        (memcmp(line, "CLIENT_ERROR", 12) == 0 || memcmp(line, "SERVER_ERROR", 12) == 0))) {
                // Every command may return a defined error instead of its
                // normal section. Retain the first error even if an earlier
                // MGET section supplied a valid status, and still drain to MN.
                server_error = true;
                if (!m_last_response.is_error() && m_last_response.get_status() != line)
                    m_last_response.set_status(line);
                m_last_response.set_error();
            } else if (tmplen >= 3 && memcmp(line, "VA ", 3) == 0) {
                // Parse the value size with strtoul rather than sscanf -- on
                // the GET hot path this avoids the full stdio scanner machinery
                // (locale, format-string re-parse) for what is just a single
                // unsigned integer immediately following a fixed prefix.
                char *endptr = NULL;
                errno = 0;
                unsigned long v = strtoul(line + 3, &endptr, 10);
                // evbuffer_remove returns int; reject sizes that cannot be
                // represented, signs, missing digits and partial number tokens.
                if (line[3] < '0' || line[3] > '9' || errno == ERANGE || v > INT_MAX ||
                    (*endptr != '\0' && *endptr != ' ')) {
                    unknown = true;
                } else {
                    m_value_len = (unsigned int) v;
                    need_value = true;
                }
            } else if (tmplen == 2 && memcmp(line, "MN", 2) == 0) {
                // MN is the universal response terminator. Every command we
                // write ends in `mn\r\n`, so every response ends with `MN`.
                done = m_last_response.get_status() != line;
                unknown = !done; // A terminator alone is not a command reply.
            } else if (tmplen >= 2 && (tmplen == 2 || line[2] == ' ') &&
                       (memcmp(line, "HD", 2) == 0 || memcmp(line, "EN", 2) == 0 || memcmp(line, "NS", 2) == 0 ||
                        memcmp(line, "EX", 2) == 0 || memcmp(line, "NF", 2) == 0)) {
                // 2-letter status: stored / miss / not-stored / exists / not-found.
                // Always intermediate -- we keep reading until MN.
            } else {
                unknown = true;
            }

            // Non-quiet mg returns exactly one VA/EN per key, while ms
            // returns one storage status. Keep expectations in issue order:
            // a single mutable mode cannot describe mixed pipelined requests.
            if (!unknown) {
                if (m_expected_replies.empty()) {
                    unknown = true;
                } else if (done) {
                    unknown = m_expected_replies.front().remaining != 0;
                } else {
                    expected_reply &expected = m_expected_replies.front();
                    bool get_reply = need_value || memcmp(line, "EN", 2) == 0;
                    unknown = expected.remaining == 0 || (!server_error && expected.get != get_reply);
                    if (!unknown) --expected.remaining;
                }
            }

            if (unknown) benchmark_debug_log("invalid meta response: %s\n", line);
            // Free `line` if it wasn't adopted as status.
            if (m_last_response.get_status() != line) free(line);

            if (unknown) {
                m_last_response.set_error();
                // Reset state so the connection (reused by shard_connection)
                // starts cleanly on the next call.
                m_response_state = rs_initial;
                return -1;
            }
            if (need_value) {
                m_response_state = rs_read_value;
                continue;
            }
            if (done) {
                m_response_state = rs_read_end;
            }
            // else: 2-letter status -- keep reading sections until MN.
            break;
        }

        case rs_read_value:
            if (evbuffer_get_length(m_read_buf) < (size_t) m_value_len + 2) return 0;
            if (m_keep_value) {
                char *value = (char *) malloc(m_value_len ? m_value_len : 1);
                if (value == NULL) {
                    m_last_response.set_error();
                    m_response_state = rs_initial;
                    return -1;
                }
                int ret = evbuffer_remove(m_read_buf, value, m_value_len);
                assert(ret == (int) m_value_len);
                m_last_response.set_value(value, m_value_len);
            } else {
                int ret = evbuffer_drain(m_read_buf, m_value_len);
                assert(ret == 0);
            }
            {
                char ending[2];
                int ret = evbuffer_remove(m_read_buf, ending, sizeof(ending));
                if (ret != 2 || ending[0] != '\r' || ending[1] != '\n') {
                    m_last_response.set_error();
                    m_response_state = rs_initial;
                    return -1;
                }
            }
            m_last_response.incr_hits();
            m_response_len += m_value_len + 2;
            // Always go back to read more sections; the response only ends at MN.
            m_response_state = rs_read_section;
            break;

        case rs_read_end:
            m_expected_replies.pop_front();
            m_response_state = rs_initial;
            return 1;

        default:
            benchmark_debug_log("unknown response state %d.\n", m_response_state);
            m_response_state = rs_initial;
            return -1;
        }
    }

    return -1;
}

/////////////////////////////////////////////////////////////////////////

class memcache_binary_protocol : public abstract_protocol
{
protected:
    enum response_state
    {
        rs_initial,
        rs_read_body
    };
    response_state m_response_state;
    protocol_binary_response_no_extras m_response_hdr;
    size_t m_response_len;

    const char *status_text(void);

public:
    memcache_binary_protocol() : m_response_state(rs_initial), m_response_len(0) {}
    virtual memcache_binary_protocol *clone(void) { return new memcache_binary_protocol(); }
    virtual int select_db(int db);
    virtual int authenticate(const char *credentials);
    virtual int configure_protocol(enum PROTOCOL_TYPE type);
    virtual int write_command_cluster_slots();
    virtual int write_command_readonly();
    virtual int write_command_set(const char *key, int key_len, const char *value, int value_len, int expiry,
                                  unsigned int offset);
    virtual int write_command_get(const char *key, int key_len, unsigned int offset);
    virtual int write_command_multi_get(const keylist *keylist);
    virtual int write_command_wait(unsigned int num_slaves, unsigned int timeout);
    virtual int parse_response(void);

    // handle arbitrary command
    virtual bool format_arbitrary_command(arbitrary_command &cmd);
    virtual int write_arbitrary_command(const command_arg *arg);
    virtual int write_arbitrary_command(const char *val, int val_len);
};

int memcache_binary_protocol::select_db(int db)
{
    assert(0);
    return -1;
}

int memcache_binary_protocol::authenticate(const char *credentials)
{
    protocol_binary_request_no_extras req;
    char nullbyte = '\0';
    const char mechanism[] = "PLAIN";
    int mechanism_len = sizeof(mechanism) - 1;
    const char *colon;
    const char *user;
    int user_len;
    const char *passwd;
    int passwd_len;

    assert(credentials != NULL);
    colon = strchr(credentials, ':');
    assert(colon != NULL);

    user = credentials;
    user_len = colon - user;
    passwd = colon + 1;
    passwd_len = strlen(passwd);

    memset(&req, 0, sizeof(req));
    req.message.header.request.magic = PROTOCOL_BINARY_REQ;
    req.message.header.request.opcode = PROTOCOL_BINARY_CMD_SASL_AUTH;
    req.message.header.request.keylen = htons(mechanism_len);
    req.message.header.request.datatype = PROTOCOL_BINARY_RAW_BYTES;
    req.message.header.request.bodylen = htonl(mechanism_len + user_len + passwd_len + 2);
    evbuffer_add(m_write_buf, &req, sizeof(req));
    evbuffer_add(m_write_buf, mechanism, mechanism_len);
    evbuffer_add(m_write_buf, &nullbyte, 1);
    evbuffer_add(m_write_buf, user, user_len);
    evbuffer_add(m_write_buf, &nullbyte, 1);
    evbuffer_add(m_write_buf, passwd, passwd_len);

    return sizeof(req) + user_len + passwd_len + 2 + sizeof(mechanism) - 1;
}

int memcache_binary_protocol::configure_protocol(enum PROTOCOL_TYPE type)
{
    assert(0);
    return -1;
}

int memcache_binary_protocol::write_command_cluster_slots()
{
    assert(0);
    return -1;
}

int memcache_binary_protocol::write_command_readonly()
{
    assert(0);
    return -1;
}

int memcache_binary_protocol::write_command_set(const char *key, int key_len, const char *value, int value_len,
                                                int expiry, unsigned int offset)
{
    assert(key != NULL);
    assert(key_len > 0);
    assert(value != NULL);
    assert(value_len > 0);

    protocol_binary_request_set req;

    memset(&req, 0, sizeof(req));
    req.message.header.request.magic = PROTOCOL_BINARY_REQ;
    req.message.header.request.opcode = PROTOCOL_BINARY_CMD_SET;
    req.message.header.request.keylen = htons(key_len);
    req.message.header.request.datatype = PROTOCOL_BINARY_RAW_BYTES;
    req.message.header.request.bodylen = htonl(sizeof(req.message.body) + value_len + key_len);
    req.message.header.request.extlen = sizeof(req.message.body);
    req.message.body.expiration = htonl(expiry);

    evbuffer_add(m_write_buf, &req, sizeof(req));
    evbuffer_add(m_write_buf, key, key_len);
    evbuffer_add(m_write_buf, value, value_len);

    return sizeof(req) + key_len + value_len;
}

int memcache_binary_protocol::write_command_get(const char *key, int key_len, unsigned int offset)
{
    assert(key != NULL);
    assert(key_len > 0);

    protocol_binary_request_get req;

    memset(&req, 0, sizeof(req));
    req.message.header.request.magic = PROTOCOL_BINARY_REQ;
    req.message.header.request.opcode = PROTOCOL_BINARY_CMD_GET;
    req.message.header.request.keylen = htons(key_len);
    req.message.header.request.datatype = PROTOCOL_BINARY_RAW_BYTES;
    req.message.header.request.bodylen = htonl(key_len);
    req.message.header.request.extlen = 0;

    evbuffer_add(m_write_buf, &req, sizeof(req));
    evbuffer_add(m_write_buf, key, key_len);

    return sizeof(req) + key_len;
}

int memcache_binary_protocol::write_command_multi_get(const keylist *keylist)
{
    fprintf(stderr, "error: multi get not implemented for binary memcache yet!\n");
    assert(0);
    return -1;
}

const char *memcache_binary_protocol::status_text(void)
{
    int status;
    static const char *status_str_00[] = {
        "PROTOCOL_BINARY_RESPONSE_SUCCESS",      "PROTOCOL_BINARY_RESPONSE_KEY_ENOENT",
        "PROTOCOL_BINARY_RESPONSE_KEY_EEXISTS",  "PROTOCOL_BINARY_RESPONSE_E2BIG",
        "PROTOCOL_BINARY_RESPONSE_EINVAL",       "PROTOCOL_BINARY_RESPONSE_NOT_STORED",
        "PROTOCOL_BINARY_RESPONSE_DELTA_BADVAL", "PROTOCOL_BINARY_RESPONSE_NOT_MY_VBUCKET",
    };
    static const char *status_str_20[] = {"PROTOCOL_BINARY_RESPONSE_AUTH_ERROR",
                                          "PROTOCOL_BINARY_RESPONSE_AUTH_CONTINUE"};
    static const char *status_str_80[] = {NULL,
                                          "PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND",
                                          "PROTOCOL_BINARY_RESPONSE_ENOMEM",
                                          "PROTOCOL_BINARY_RESPONSE_NOT_SUPPORTED",
                                          "PROTOCOL_BINARY_RESPONSE_EINTERNAL",
                                          "PROTOCOL_BINARY_RESPONSE_EBUSY",
                                          "PROTOCOL_BINARY_RESPONSE_ETMPFAIL"};

    status = ntohs(m_response_hdr.message.header.response.status);
    if (status <= 0x07) {
        return status_str_00[status];
    } else if (status >= 0x20 && status <= 0x21) {
        return status_str_20[status - 0x20];
    } else if (status >= 0x80 && status <= 0x86) {
        return status_str_80[status - 0x80];
    } else {
        return NULL;
    }
}

int memcache_binary_protocol::write_command_wait(unsigned int num_slaves, unsigned int timeout)
{
    fprintf(stderr, "error: WAIT command not implemented for memcache!\n");
    assert(0);
    return -1;
}

int memcache_binary_protocol::parse_response(void)
{
    while (true) {
        int ret;
        int status;

        switch (m_response_state) {
        case rs_initial:
            if (evbuffer_get_length(m_read_buf) < sizeof(m_response_hdr)) return 0; // no header yet?

            ret = evbuffer_remove(m_read_buf, (void *) &m_response_hdr, sizeof(m_response_hdr));
            assert(ret == sizeof(m_response_hdr));

            if (m_response_hdr.message.header.response.magic != PROTOCOL_BINARY_RES) {
                benchmark_error_log("error: invalid memcache response header magic.\n");
                return -1;
            }

            m_response_len = sizeof(m_response_hdr);
            m_last_response.clear();
            if (status_text()) {
                m_last_response.set_status(strdup(status_text()));
            }

            status = ntohs(m_response_hdr.message.header.response.status);
            if (status == PROTOCOL_BINARY_RESPONSE_AUTH_ERROR || status == PROTOCOL_BINARY_RESPONSE_AUTH_CONTINUE ||
                status == PROTOCOL_BINARY_RESPONSE_NOT_SUPPORTED ||
                status == PROTOCOL_BINARY_RESPONSE_UNKNOWN_COMMAND ||
                status == PROTOCOL_BINARY_RESPONSE_NOT_SUPPORTED || status == PROTOCOL_BINARY_RESPONSE_EBUSY) {
                m_last_response.set_error();
            }

            if (ntohl(m_response_hdr.message.header.response.bodylen) > 0) {
                m_response_hdr.message.header.response.bodylen = ntohl(m_response_hdr.message.header.response.bodylen);
                m_response_hdr.message.header.response.keylen = ntohs(m_response_hdr.message.header.response.keylen);

                m_response_state = rs_read_body;
                continue;
            }

            return 1;
            break;
        case rs_read_body:
            if (evbuffer_get_length(m_read_buf) >= m_response_hdr.message.header.response.bodylen) {
                // get rid of extras and key, we don't care about them
                ret = evbuffer_drain(m_read_buf, m_response_hdr.message.header.response.extlen +
                                                     m_response_hdr.message.header.response.keylen);
                assert((unsigned int) ret == 0);

                int actual_body_len = m_response_hdr.message.header.response.bodylen -
                                      m_response_hdr.message.header.response.extlen -
                                      m_response_hdr.message.header.response.keylen;
                if (m_keep_value) {
                    char *value = (char *) malloc(actual_body_len);
                    assert(value != NULL);
                    ret = evbuffer_remove(m_read_buf, value, actual_body_len);
                    m_last_response.set_value(value, actual_body_len);
                } else {
                    int ret = evbuffer_drain(m_read_buf, actual_body_len);
                    assert((unsigned int) ret == 0);
                }

                if (m_response_hdr.message.header.response.status == PROTOCOL_BINARY_RESPONSE_SUCCESS)
                    m_last_response.incr_hits();

                m_response_len += m_response_hdr.message.header.response.bodylen;
                m_response_state = rs_initial;

                return 1;
            } else {
                return 0;
            }
            break;
        default:
            benchmark_debug_log("unknown response state.\n");
            return -1;
        }
    }

    return -1;
}

bool memcache_binary_protocol::format_arbitrary_command(arbitrary_command &cmd)
{
    assert(0);
    return false;
}

int memcache_binary_protocol::write_arbitrary_command(const command_arg *arg)
{
    assert(0);
    return -1;
}

int memcache_binary_protocol::write_arbitrary_command(const char *val, int val_len)
{
    assert(0);
    return -1;
}

/////////////////////////////////////////////////////////////////////////

class abstract_protocol *protocol_factory(enum PROTOCOL_TYPE type)
{
    if (is_redis_protocol(type)) {
        return new redis_protocol();
    } else if (type == PROTOCOL_MEMCACHE_TEXT) {
        return new memcache_text_protocol();
    } else if (type == PROTOCOL_MEMCACHE_BINARY) {
        return new memcache_binary_protocol();
    } else if (type == PROTOCOL_MEMCACHE_META) {
        return new memcache_meta_protocol();
    } else {
        benchmark_error_log("Error: unknown protocol type: %d.\n", type);
        return NULL;
    }
}

/////////////////////////////////////////////////////////////////////////

// The MGET key list can need GBs with a large --multi-key-get and long keys:
// fail with a clear message rather than an assert.
static void keylist_out_of_memory(void)
{
    fprintf(stderr, "error: out of memory building MGET key list (reduce --multi-key-get or the key size).\n");
    exit(1);
}

keylist::keylist(unsigned int max_keys) :
        m_buffer(NULL), m_buffer_ptr(NULL), m_buffer_size(0), m_keys(NULL), m_keys_size(0), m_keys_count(0)
{
    m_keys_size = max_keys;
    m_keys = (key_entry *) malloc(m_keys_size * sizeof(key_entry));
    if (m_keys == NULL) keylist_out_of_memory();
    memset(m_keys, 0, m_keys_size * sizeof(key_entry));

    /* allocate buffer for actual keys */
    m_buffer_size = (size_t) 256 * m_keys_size;
    m_buffer = (char *) malloc(m_buffer_size);
    if (m_buffer == NULL) keylist_out_of_memory();
    memset(m_buffer, 0, m_buffer_size);

    m_buffer_ptr = m_buffer;
}

keylist::~keylist()
{
    if (m_buffer != NULL) {
        free(m_buffer);
        m_buffer = NULL;
    }
    if (m_keys != NULL) {
        free(m_keys);
        m_keys = NULL;
    }
}

bool keylist::add_key(const char *key, unsigned int key_len)
{
    // have room?
    if (m_keys_count >= m_keys_size) return false;

    // have buffer? (+1 for the NUL terminator written after the key bytes)
    size_t offset = (size_t) (m_buffer_ptr - m_buffer);
    size_t needed = offset + (size_t) key_len + 1;
    if (needed > m_buffer_size) {
        // size_t growth: an unsigned int size wrapped to 0 past 2 GiB and the
        // doubling loop never ended.
        m_buffer_size = keylist_grow_size(m_buffer_size, needed);
        m_buffer = (char *) realloc(m_buffer, m_buffer_size);
        if (m_buffer == NULL) keylist_out_of_memory();
        // Re-base all key_ptr entries into the new allocation. Keys are packed
        // contiguously (key bytes + NUL terminator), so we can recompute each
        // pointer from m_keys[k].key_len without touching the old allocation.
        char *p = m_buffer;
        for (unsigned int k = 0; k < m_keys_count; k++) {
            m_keys[k].key_ptr = p;
            p += m_keys[k].key_len + 1;
        }
        m_buffer_ptr = m_buffer + offset;
    }

    // copy key
    memcpy(m_buffer_ptr, key, key_len);
    m_buffer_ptr[key_len] = '\0';
    m_keys[m_keys_count].key_ptr = m_buffer_ptr;
    m_keys[m_keys_count].key_len = key_len;

    m_buffer_ptr += key_len + 1;
    m_keys_count++;

    return true;
}

unsigned int keylist::get_keys_count(void) const
{
    return m_keys_count;
}

const char *keylist::get_key(unsigned int index, unsigned int *key_len) const
{
    if (index < 0 || index >= m_keys_count) return NULL;
    if (key_len != NULL) *key_len = m_keys[index].key_len;
    return m_keys[index].key_ptr;
}

void keylist::clear(void)
{
    m_keys_count = 0;
    m_buffer_ptr = m_buffer;
}