client.cpp

/*
 * Copyright (C) 2011-2017 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

#ifdef HAVE_SYS_TYPES_H
#include<sys/types.h>
#endif
#ifdef HAVE_FCNTL_H
#include<fcntl.h>
#endif
#include<unistd.h>
#include<stdlib.h>
#include<string.h>
#include<errno.h>
#ifdef HAVE_SYS_SOCKET_H
#include<sys/socket.h>
#endif
#ifdef HAVE_NETINET_TCP_H
#include<netinet/tcp.h>
#endif
#ifdef HAVE_LIMITS_H
#include<limits.h>
#endif

#ifdef HAVE_ASSERT_H
#include<assert.h>
#endif

#include<math.h>
#include<algorithm>
#include<arpa/inet.h>

#include"client.h"
#include"cluster_client.h"


boolclient::setup_client(benchmark_config *config, abstract_protocol *protocol, object_generator *objgen)
{
 m_config = config;
assert(m_config != NULL);
unsignedlonglong total_num_of_clients = config->clients*config->threads;

// create main connection
 shard_connection* conn = newshard_connection(m_connections.size(), this, m_config, m_event_base, protocol);
 m_connections.push_back(conn);

 m_obj_gen = objgen->clone();
assert(m_obj_gen != NULL);

if (config->distinct_client_seed && config->randomize)
 m_obj_gen->set_random_seed(config->randomize + config->next_client_idx);
elseif (config->randomize)
 m_obj_gen->set_random_seed(config->randomize);
elseif (config->distinct_client_seed)
 m_obj_gen->set_random_seed(config->next_client_idx);

// Setup first arbitrary command
if (config->arbitrary_commands->is_defined())
advance_arbitrary_command_index();

// Parallel key-pattern determined according to the first command
if ((config->arbitrary_commands->is_defined() && config->arbitrary_commands->at(0).key_pattern == 'P') ||
 (config->key_pattern[key_pattern_set]=='P')) {
unsignedlonglong client_index = config->next_client_idx % total_num_of_clients;

unsignedlonglong range = (config->key_maximum - config->key_minimum)/total_num_of_clients + 1;
unsignedlonglong min = config->key_minimum + (range * client_index);
unsignedlonglong max = min + range - 1;

if (client_index == (total_num_of_clients - 1)) {
 max = config->key_maximum; //the last clients takes the leftover
 }

 m_obj_gen->set_key_range(min, max);
 }
 config->next_client_idx++;

 m_keylist = newkeylist(m_config->multi_key_get + 1);
assert(m_keylist != NULL);

returntrue;
}

client::client(client_group* group) :
 m_event_base(NULL), m_initialized(false), m_end_set(false), m_config(NULL),
 m_obj_gen(NULL), m_stats(group->get_config()), m_reqs_processed(0), m_reqs_generated(0),
 m_set_ratio_count(0), m_get_ratio_count(0),
 m_arbitrary_command_ratio_count(0), m_executed_command_index(0),
 m_tot_set_ops(0), m_tot_wait_ops(0)
{
 m_event_base = group->get_event_base();

if (!setup_client(group->get_config(), group->get_protocol(), group->get_obj_gen())) {
return;
 }

benchmark_debug_log("new client %p successfully set up.\n", this);
 m_initialized = true;
}

client::client(structevent_base *event_base, benchmark_config *config,
 abstract_protocol *protocol, object_generator *obj_gen) :
 m_event_base(NULL), m_initialized(false), m_end_set(false), m_config(NULL),
 m_obj_gen(NULL), m_stats(config), m_reqs_processed(0), m_reqs_generated(0),
 m_set_ratio_count(0), m_get_ratio_count(0),
 m_arbitrary_command_ratio_count(0), m_executed_command_index(0),
 m_tot_set_ops(0), m_tot_wait_ops(0), m_keylist(NULL)
{
 m_event_base = event_base;

if (!setup_client(config, protocol, obj_gen)) {
return;
 }

benchmark_debug_log("new client %p successfully set up.\n", this);
 m_initialized = true;
}

client::~client()
{
for (unsignedint i = 0; i < m_connections.size(); i++) {
 shard_connection* sc = m_connections[i];
delete sc;
 }
 m_connections.clear();

if (m_obj_gen != NULL) {
delete m_obj_gen;
 m_obj_gen = NULL;
 }

if (m_keylist != NULL) {
delete m_keylist;
 m_keylist = NULL;
 }
}

boolclient::initialized(void)
{
return m_initialized;
}

voidclient::disconnect(void)
{
 shard_connection* sc = MAIN_CONNECTION;
assert(sc != NULL);

 sc->disconnect();
}

intclient::connect(void)
{
structconnect_info addr;

// get primary connection
 shard_connection* sc = MAIN_CONNECTION;
assert(sc != NULL);

// get address information
if (m_config->unix_socket == NULL) {
if (m_config->server_addr->get_connect_info(&addr) != 0) {
benchmark_error_log("connect: resolve error: %s\n", m_config->server_addr->get_last_error());
return -1;
 }

// Just in case we got domain name and not ip, we convert it
char address[INET6_ADDRSTRLEN];
if (addr.ci_family == PF_INET) {
structsockaddr_in *ipv4 = (structsockaddr_in *)addr.ci_addr;
inet_ntop(AF_INET, &(ipv4->sin_addr), address, INET_ADDRSTRLEN);
 } else {
structsockaddr_in6 *ipv6 = (structsockaddr_in6 *)addr.ci_addr;
inet_ntop(AF_INET6, &(ipv6->sin6_addr), address, INET6_ADDRSTRLEN);
 }

char port_str[20];
snprintf(port_str, sizeof(port_str)-1, "%u", m_config->port);

// save address and port
 sc->set_address_port(address, port_str);
 }

// call connect
int ret = sc->connect(&addr);
if (ret)
return ret;

return0;
}

boolclient::finished(void)
{
if (m_config->requests > 0 && m_reqs_processed >= m_config->requests)
returntrue;
if (m_config->test_time > 0 && m_stats.get_duration() >= m_config->test_time)
returntrue;
returnfalse;
}

voidclient::set_start_time() {
structtimeval now;

gettimeofday(&now, NULL);
 m_stats.set_start_time(&now);
}

voidclient::set_end_time() {
// update only once
if (!m_end_set) {
benchmark_debug_log("nothing else to do, test is finished.\n");

 m_stats.set_end_time(NULL);
 m_end_set = true;
 }
}

boolclient::hold_pipeline(unsignedint conn_id) {
// don't exceed requests
if (m_config->requests) {
if (m_reqs_generated >= m_config->requests)
returntrue;
 }

// if we have reconnect_interval stop enlarging the pipeline on time
if (m_config->reconnect_interval) {
if ((m_reqs_processed % m_config->reconnect_interval) + (m_reqs_generated - m_reqs_processed) >= m_config->reconnect_interval)
returntrue;
 }

returnfalse;
}

get_key_response client::get_key_for_conn(unsignedint command_index, unsignedint conn_id, unsignedlonglong* key_index) {
int iter;
if (m_config->arbitrary_commands->is_defined())
 iter = arbitrary_obj_iter_type(command_index);
else
 iter = obj_iter_type(m_config, command_index);

 *key_index = m_obj_gen->get_key_index(iter);

if (!m_config->data_import || m_config->generate_keys) {
 m_obj_gen->generate_key(*key_index);
 } else {
/* For SET command we already read a completes item (see create_set_request()) */
if (command_index == GET_CMD_IDX)
dynamic_cast<import_object_generator*>(m_obj_gen)->read_next_key(*key_index);
 }

return available_for_conn;
}

boolclient::create_arbitrary_request(unsignedint command_index, structtimeval& timestamp, unsignedint conn_id) {
int cmd_size = 0;

const arbitrary_command& cmd = get_arbitrary_command(command_index);

benchmark_debug_log("%s: %s:\n", m_connections[conn_id]->get_readable_id(), cmd.command.c_str());

for (unsignedint i = 0; i < cmd.command_args.size(); i++) {
const command_arg* arg = &cmd.command_args[i];
if (arg->type == const_type) {
 cmd_size += m_connections[conn_id]->send_arbitrary_command(arg);
 } elseif (arg->type == key_type) {
unsignedlonglong key_index;
 get_key_response res = get_key_for_conn(command_index, conn_id, &key_index);
/* If key not available for this connection, we have a bug of sending partial request */
assert(res == available_for_conn);
 cmd_size += m_connections[conn_id]->send_arbitrary_command(arg, m_obj_gen->get_key(), m_obj_gen->get_key_len());
 } elseif (arg->type == data_type) {
unsignedint value_len;
constchar *value = m_obj_gen->get_value(0, &value_len);

assert(value != NULL);
assert(value_len > 0);

 cmd_size += m_connections[conn_id]->send_arbitrary_command(arg, value, value_len);
 }
 }

 m_connections[conn_id]->send_arbitrary_command_end(command_index, &timestamp, cmd_size);
returntrue;
}

boolclient::create_wait_request(structtimeval& timestamp, unsignedint conn_id) {
unsignedint num_slaves = m_obj_gen->random_range(m_config->num_slaves.min, m_config->num_slaves.max);
unsignedint timeout = m_obj_gen->normal_distribution(m_config->wait_timeout.min,
 m_config->wait_timeout.max, 0,
 ((m_config->wait_timeout.max - m_config->wait_timeout.min)/2.0) + m_config->wait_timeout.min);

 m_connections[conn_id]->send_wait_command(&timestamp, num_slaves, timeout);
returntrue;
}

boolclient::create_set_request(structtimeval& timestamp, unsignedint conn_id) {
unsignedlonglong key_index;
 get_key_response res = get_key_for_conn(SET_CMD_IDX, conn_id, &key_index);
if (res == not_available)
returnfalse;

if (res == available_for_conn) {
unsignedint value_len;
constchar *value = m_obj_gen->get_value(key_index, &value_len);

 m_connections[conn_id]->send_set_command(&timestamp, m_obj_gen->get_key(), m_obj_gen->get_key_len(),
 value, value_len, m_obj_gen->get_expiry(),
 m_config->data_offset);
 }

returntrue;
}

boolclient::create_get_request(structtimeval& timestamp, unsignedint conn_id) {
unsignedlonglong key_index;
 get_key_response res = get_key_for_conn(GET_CMD_IDX, conn_id, &key_index);
if (res == not_available)
returnfalse;

if (res == available_for_conn) {
 m_connections[conn_id]->send_get_command(&timestamp, m_obj_gen->get_key(), m_obj_gen->get_key_len(), m_config->data_offset);
 }

returntrue;
}

boolclient::create_mget_request(structtimeval& timestamp, unsignedint conn_id) {
unsignedlonglong key_index;
unsignedint keys_count = m_config->ratio.b - m_get_ratio_count;
if ((int)keys_count > m_config->multi_key_get)
 keys_count = m_config->multi_key_get;

 m_keylist->clear();
for (unsignedint i = 0; i < keys_count; i++) {
 get_key_response res = get_key_for_conn(GET_CMD_IDX, conn_id, &key_index);
/* Not supported in cluster mode */
assert(res == available_for_conn);

 m_keylist->add_key(m_obj_gen->get_key(), m_obj_gen->get_key_len());
 }

 m_connections[conn_id]->send_mget_command(&timestamp, m_keylist);
returntrue;
}

// This function could use some urgent TLC -- but we need to do it without altering the behavior
voidclient::create_request(structtimeval timestamp, unsignedint conn_id)
{
// are we using arbitrary command?
if (m_config->arbitrary_commands->is_defined()) {
if (create_arbitrary_request(m_executed_command_index, timestamp, conn_id)) {
advance_arbitrary_command_index();
 m_reqs_generated++;
 }
return;
 }

// If the Set:Wait ratio is not 0, start off with WAITs
if (m_config->wait_ratio.b &&
 (m_tot_wait_ops == 0 ||
 (m_tot_set_ops/m_tot_wait_ops > m_config->wait_ratio.a/m_config->wait_ratio.b))) {
if (!create_wait_request(timestamp, conn_id))
return;

 m_reqs_generated++;
 m_tot_wait_ops++;
 }

// are we set or get? this depends on the ratio
elseif (m_set_ratio_count < m_config->ratio.a) {
/* Before we can create a SET request, we need to read the next imported item */
if (m_config->data_import) {
dynamic_cast<import_object_generator*>(m_obj_gen)->read_next_item();
 }

if (!create_set_request(timestamp, conn_id))
return;

 m_set_ratio_count++;
 m_reqs_generated++;
 m_tot_set_ops++;
 } elseif (m_get_ratio_count < m_config->ratio.b) {
// GET command
if (!m_config->multi_key_get) {
if (!create_get_request(timestamp, conn_id))
return;

 m_get_ratio_count++;
 m_reqs_generated++;
return;
 }

// MGET command
if (!create_mget_request(timestamp, conn_id))
return;

 m_get_ratio_count += m_config->multi_key_get;
 m_reqs_generated++;
 } else {
// overlap counters
 m_get_ratio_count = m_set_ratio_count = 0;
 }
}

intclient::prepare(void)
{
if (MAIN_CONNECTION == NULL)
return -1;

int ret = this->connect();
if (ret < 0) {
benchmark_error_log("prepare: failed to connect, test aborted.\n");
return ret;
 }

return0;
}

voidclient::handle_response(unsignedint conn_id, structtimeval timestamp,
 request *request, protocol_response *response)
{
if (response->is_error()) {
benchmark_error_log("server %s handle error response: %s\n",
 m_connections[conn_id]->get_readable_id(),
 response->get_status());
 }
switch (request->m_type) {
case rt_get:
 m_stats.update_get_op(&timestamp,
 response->get_total_len(),
 request->m_size,
ts_diff(request->m_sent_time, timestamp),
 response->get_hits(),
 request->m_keys - response->get_hits());
break;
case rt_set:
 m_stats.update_set_op(&timestamp,
 response->get_total_len(),
 request->m_size,
ts_diff(request->m_sent_time, timestamp));
break;
case rt_wait:
 m_stats.update_wait_op(&timestamp,
ts_diff(request->m_sent_time, timestamp));
break;
case rt_arbitrary: {
 arbitrary_request *ar = static_cast<arbitrary_request *>(request);
 m_stats.update_arbitrary_op(&timestamp,
 response->get_total_len(),
 request->m_size,
ts_diff(request->m_sent_time, timestamp),
 ar->index);
break;
 }
default:
assert(0);
break;
 }
}

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

verify_client::verify_client(structevent_base *event_base,
 benchmark_config *config,
 abstract_protocol *protocol,
 object_generator *obj_gen) : client(event_base, config, protocol, obj_gen),
 m_finished(false), m_verified_keys(0), m_errors(0)
{
 MAIN_CONNECTION->get_protocol()->set_keep_value(true);
}

unsignedlonglongintverify_client::get_verified_keys(void)
{
return m_verified_keys;
}

unsignedlonglongintverify_client::get_errors(void)
{
return m_errors;
}

boolverify_client::create_wait_request(structtimeval& timestamp, unsignedint conn_id) {
// Nothing to do
returntrue;
}

boolverify_client::create_set_request(structtimeval& timestamp, unsignedint conn_id) {
unsignedlonglong key_index;
 get_key_response res = get_key_for_conn(SET_CMD_IDX, conn_id, &key_index);
if (res == not_available)
returnfalse;

if (res == available_for_conn) {
unsignedint value_len;
constchar *value = m_obj_gen->get_value(key_index, &value_len);

 m_connections[conn_id]->send_verify_get_command(&timestamp, m_obj_gen->get_key(), m_obj_gen->get_key_len(),
 value, value_len,
 m_config->data_offset);
 }

returntrue;
}

boolverify_client::create_get_request(structtimeval& timestamp, unsignedint conn_id) {
// Just Keep object generator synced
unsignedlonglong key_index;
get_key_for_conn(GET_CMD_IDX, conn_id, &key_index);

returntrue;
}

boolverify_client::create_mget_request(structtimeval& timestamp, unsignedint conn_id) {
// Just Keep object generator synced
unsignedlonglong key_index;
unsignedint keys_count = m_config->ratio.b - m_get_ratio_count;
if ((int)keys_count > m_config->multi_key_get)
 keys_count = m_config->multi_key_get;

 m_keylist->clear();
for (unsignedint i = 0; i < keys_count; i++) {

get_key_for_conn(GET_CMD_IDX, conn_id, &key_index);
 }

returntrue;
}

voidverify_client::handle_response(unsignedint conn_id, structtimeval timestamp,
 request *request, protocol_response *response)
{
unsignedint rvalue_len;
constchar *rvalue = response->get_value(&rvalue_len);
 verify_request *vr = static_cast<verify_request *>(request);

assert(vr->m_type == rt_get);
if (response->is_error()) {
benchmark_error_log("error: request for key [%.*s] failed: %s\n",
 vr->m_key_len, vr->m_key, response->get_status());
 m_errors++;
 } else {
if (!rvalue || rvalue_len != vr->m_value_len || memcmp(rvalue, vr->m_value, rvalue_len) != 0) {
benchmark_error_log("error: key [%.*s]: expected [%.*s], got [%.*s]\n",
 vr->m_key_len, vr->m_key,
 vr->m_value_len, vr->m_value,
 rvalue_len, rvalue);
 m_errors++;
 } else {
benchmark_debug_log("key: [%.*s] verified successfuly.\n",
 vr->m_key_len, vr->m_key);
 m_verified_keys++;
 }
 }
}

boolverify_client::finished(void)
{
if (m_finished)
returntrue;
if (m_config->requests > 0 && m_reqs_processed >= m_config->requests)
returntrue;
returnfalse;
}

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

client_group::client_group(benchmark_config* config, abstract_protocol *protocol, object_generator* obj_gen) :
 m_base(NULL), m_config(config), m_protocol(protocol), m_obj_gen(obj_gen)
{
 m_base = event_base_new();
assert(m_base != NULL);

assert(protocol != NULL);
assert(obj_gen != NULL);
}

client_group::~client_group(void)
{
for (std::vector<client*>::iterator i = m_clients.begin(); i != m_clients.end(); i++) {
 client* c = *i;
delete c;
 }
 m_clients.clear();

if (m_base != NULL)
event_base_free(m_base);
 m_base = NULL;
}

intclient_group::create_clients(int num)
{
for (int i = 0; i < num; i++) {
 client* c;

if (m_config->cluster_mode)
 c = newcluster_client(this);
else
 c = newclient(this);

assert(c != NULL);

if (!c->initialized()) {
delete c;
return i;
 }

 m_clients.push_back(c);
 }

return num;
}

intclient_group::prepare(void)
{
for (std::vector<client*>::iterator i = m_clients.begin(); i != m_clients.end(); i++) {
 client* c = *i;
int ret = c->prepare();

if (ret < 0) {
return ret;
 }
 }

return0;
}

voidclient_group::run(void)
{
event_base_dispatch(m_base);
}

unsignedlongintclient_group::get_total_bytes(void)
{
unsignedlongint total_bytes = 0;
for (std::vector<client*>::iterator i = m_clients.begin(); i != m_clients.end(); i++) {
 total_bytes += (*i)->get_stats()->get_total_bytes();
 }

return total_bytes;
}

unsignedlongintclient_group::get_total_ops(void)
{
unsignedlongint total_ops = 0;
for (std::vector<client*>::iterator i = m_clients.begin(); i != m_clients.end(); i++) {
 total_ops += (*i)->get_stats()->get_total_ops();
 }

return total_ops;
}

unsignedlongintclient_group::get_total_latency(void)
{
unsignedlongint total_latency = 0;
for (std::vector<client*>::iterator i = m_clients.begin(); i != m_clients.end(); i++) {
 total_latency += (*i)->get_stats()->get_total_latency();
 }

return total_latency;
}

unsignedlongintclient_group::get_duration_usec(void)
{
unsignedlongint duration = 0;
unsignedint thread_counter = 1;
for (std::vector<client*>::iterator i = m_clients.begin(); i != m_clients.end(); i++, thread_counter++) {
float factor = ((float)(thread_counter - 1) / thread_counter);
 duration = factor * duration + (float)(*i)->get_stats()->get_duration_usec() / thread_counter ;
 }

return duration;
}

voidclient_group::merge_run_stats(run_stats* target)
{
assert(target != NULL);
unsignedint iteration_counter = 1;
for (std::vector<client*>::iterator i = m_clients.begin(); i != m_clients.end(); i++) {
 target->merge(*(*i)->get_stats(), iteration_counter++);
 }
}

voidclient_group::write_client_stats(constchar *prefix)
{
unsignedint client_id = 0;

for (std::vector<client*>::iterator i = m_clients.begin(); i != m_clients.end(); i++) {
char filename[PATH_MAX];

snprintf(filename, sizeof(filename)-1, "%s-%u.csv", prefix, client_id++);
if (!(*i)->get_stats()->save_csv(filename, m_config)) {
fprintf(stderr, "error: %s: failed to write client stats.\n", filename);
 }
 }
}