2 +--------------------------------------------------------------------+
3 | libmemcached - C/C++ Client Library for memcached |
4 +--------------------------------------------------------------------+
5 | Redistribution and use in source and binary forms, with or without |
6 | modification, are permitted under the terms of the BSD license. |
7 | You should have received a copy of the license in a bundled file |
8 | named LICENSE; in case you did not receive a copy you can review |
9 | the terms online at: https://opensource.org/licenses/BSD-3-Clause |
10 +--------------------------------------------------------------------+
11 | Copyright (c) 2006-2014 Brian Aker https://datadifferential.com/ |
12 | Copyright (c) 2020 Michael Wallner <mike@php.net> |
13 +--------------------------------------------------------------------+
16 #include "mem_config.h"
30 #include <sys/types.h>
35 #include "libmemcached-1.0/memcached.h"
37 #include "client_options.h"
38 #include "utilities.h"
39 #include "generator.h"
42 #define DEFAULT_INITIAL_LOAD 10000
43 #define DEFAULT_EXECUTE_NUMBER 10000
44 #define DEFAULT_CONCURRENCY 1
46 #define VALUE_BYTES 4096
48 #define PROGRAM_NAME "memslap"
49 #define PROGRAM_DESCRIPTION "Generates a load against a memcached custer of servers."
51 /* Global Thread counter */
52 volatile unsigned int master_wakeup
;
53 pthread_mutex_t sleeper_mutex
;
54 pthread_cond_t sleep_threshhold
;
57 enum test_t
{ SET_TEST
, GET_TEST
, MGET_TEST
};
59 struct thread_context_st
{
60 unsigned int key_count
;
61 pairs_st
*initial_pairs
;
62 unsigned int initial_number
;
63 pairs_st
*execute_pairs
;
64 unsigned int execute_number
;
69 const memcached_st
*root
;
71 thread_context_st(const memcached_st
*memc_arg
, test_t test_arg
)
83 void init() { memc
= memcached_clone(NULL
, root
); }
85 ~thread_context_st() {
87 pairs_free(execute_pairs
);
93 struct conclusions_st
{
96 unsigned int rows_loaded
;
97 unsigned int rows_read
;
107 void options_parse(int argc
, char *argv
[]);
108 void conclusions_print(conclusions_st
*conclusion
);
109 void scheduler(memcached_server_st
*servers
, conclusions_st
*conclusion
);
110 pairs_st
*load_create_data(memcached_st
*memc
, unsigned int number_of
, unsigned int *actual_loaded
);
111 void flush_all(memcached_st
*memc
);
113 static bool opt_binary
= 0;
114 static int opt_verbose
= 0;
115 static int opt_flush
= 0;
116 static int opt_non_blocking_io
= 0;
117 static int opt_tcp_nodelay
= 0;
118 static unsigned int opt_execute_number
= 0;
119 static unsigned int opt_createial_load
= 0;
120 static unsigned int opt_concurrency
= 0;
121 static int opt_displayflag
= 0;
122 static char *opt_servers
= NULL
;
123 static bool opt_udp_io
= false;
124 test_t opt_test
= SET_TEST
;
128 static __attribute__((noreturn
)) void *run_task(void *p
) {
129 thread_context_st
*context
= (thread_context_st
*) p
;
133 pthread_mutex_lock(&sleeper_mutex
);
134 while (master_wakeup
) {
135 pthread_cond_wait(&sleep_threshhold
, &sleeper_mutex
);
137 pthread_mutex_unlock(&sleeper_mutex
);
140 switch (context
->test
) {
142 assert(context
->execute_pairs
);
143 execute_set(context
->memc
, context
->execute_pairs
, context
->execute_number
);
146 case GET_TEST
: execute_get(context
->memc
, context
->initial_pairs
, context
->initial_number
); break;
149 execute_mget(context
->memc
, (const char *const *) context
->keys
, context
->key_lengths
,
150 context
->initial_number
);
160 int main(int argc
, char *argv
[]) {
161 conclusions_st conclusion
;
163 srandom((unsigned int) time(NULL
));
164 options_parse(argc
, argv
);
166 if (opt_servers
== NULL
) {
169 if ((temp
= getenv("MEMCACHED_SERVERS"))) {
170 opt_servers
= strdup(temp
);
173 if (opt_servers
== NULL
) {
174 std::cerr
<< "No servers provided" << std::endl
;
179 memcached_server_st
*servers
= memcached_servers_parse(opt_servers
);
180 if (servers
== NULL
or memcached_server_list_count(servers
) == 0) {
181 std::cerr
<< "Invalid server list provided:" << opt_servers
<< std::endl
;
185 pthread_mutex_init(&sleeper_mutex
, NULL
);
186 pthread_cond_init(&sleep_threshhold
, NULL
);
188 int error_code
= EXIT_SUCCESS
;
190 scheduler(servers
, &conclusion
);
191 } catch (std::exception
&e
) {
192 std::cerr
<< "Died with exception: " << e
.what() << std::endl
;
193 error_code
= EXIT_FAILURE
;
198 (void) pthread_mutex_destroy(&sleeper_mutex
);
199 (void) pthread_cond_destroy(&sleep_threshhold
);
200 conclusions_print(&conclusion
);
201 memcached_server_list_free(servers
);
206 void scheduler(memcached_server_st
*servers
, conclusions_st
*conclusion
) {
207 unsigned int actual_loaded
= 0; /* Fix warning */
209 struct timeval start_time
, end_time
;
210 pairs_st
*pairs
= NULL
;
212 memcached_st
*memc
= memcached_create(NULL
);
214 memcached_server_push(memc
, servers
);
216 /* We need to set udp behavior before adding servers to the client */
218 if (memcached_failed(memcached_behavior_set(memc
, MEMCACHED_BEHAVIOR_USE_UDP
, opt_udp_io
))) {
219 std::cerr
<< "Failed to enable UDP." << std::endl
;
220 memcached_free(memc
);
225 memcached_behavior_set(memc
, MEMCACHED_BEHAVIOR_BINARY_PROTOCOL
, (uint64_t) opt_binary
);
231 if (opt_createial_load
) {
232 pairs
= load_create_data(memc
, opt_createial_load
, &actual_loaded
);
235 char **keys
= static_cast<char **>(calloc(actual_loaded
, sizeof(char *)));
236 size_t *key_lengths
= static_cast<size_t *>(calloc(actual_loaded
, sizeof(size_t)));
238 if (keys
== NULL
or key_lengths
== NULL
) {
244 for (uint32_t x
= 0; x
< actual_loaded
; ++x
) {
245 keys
[x
] = pairs
[x
].key
;
246 key_lengths
[x
] = pairs
[x
].key_length
;
250 /* We set this after we have loaded */
252 if (opt_non_blocking_io
)
253 memcached_behavior_set(memc
, MEMCACHED_BEHAVIOR_NO_BLOCK
, 1);
256 memcached_behavior_set(memc
, MEMCACHED_BEHAVIOR_TCP_NODELAY
, 1);
259 pthread_mutex_lock(&sleeper_mutex
);
261 pthread_mutex_unlock(&sleeper_mutex
);
263 pthread_t
*threads
= new (std::nothrow
) pthread_t
[opt_concurrency
];
265 if (threads
== NULL
) {
269 for (uint32_t x
= 0; x
< opt_concurrency
; x
++) {
270 thread_context_st
*context
= new thread_context_st(memc
, opt_test
);
271 context
->test
= opt_test
;
273 context
->initial_pairs
= pairs
;
274 context
->initial_number
= actual_loaded
;
275 context
->keys
= keys
;
276 context
->key_lengths
= key_lengths
;
278 if (opt_test
== SET_TEST
) {
279 context
->execute_pairs
= pairs_generate(opt_execute_number
, VALUE_BYTES
);
280 context
->execute_number
= opt_execute_number
;
283 /* now you create the thread */
284 if (pthread_create(threads
+ x
, NULL
, run_task
, (void *) context
) != 0) {
285 fprintf(stderr
, "Could not create thread\n");
290 pthread_mutex_lock(&sleeper_mutex
);
292 pthread_mutex_unlock(&sleeper_mutex
);
293 pthread_cond_broadcast(&sleep_threshhold
);
294 gettimeofday(&start_time
, NULL
);
296 for (uint32_t x
= 0; x
< opt_concurrency
; x
++) {
298 pthread_join(threads
[x
], &retval
);
302 gettimeofday(&end_time
, NULL
);
304 conclusion
->load_time
= timedif(end_time
, start_time
);
305 conclusion
->read_time
= timedif(end_time
, start_time
);
309 memcached_free(memc
);
312 void options_parse(int argc
, char *argv
[]) {
313 memcached_programs_help_st help_options
[] = {
317 static struct option long_options
[] = {
318 {(OPTIONSTRING
) "concurrency", required_argument
, NULL
, OPT_SLAP_CONCURRENCY
},
319 {(OPTIONSTRING
) "debug", no_argument
, &opt_verbose
, OPT_DEBUG
},
320 {(OPTIONSTRING
) "quiet", no_argument
, NULL
, OPT_QUIET
},
321 {(OPTIONSTRING
) "execute-number", required_argument
, NULL
, OPT_SLAP_EXECUTE_NUMBER
},
322 {(OPTIONSTRING
) "flag", no_argument
, &opt_displayflag
, OPT_FLAG
},
323 {(OPTIONSTRING
) "flush", no_argument
, &opt_flush
, OPT_FLUSH
},
324 {(OPTIONSTRING
) "help", no_argument
, NULL
, OPT_HELP
},
325 {(OPTIONSTRING
) "initial-load", required_argument
, NULL
,
326 OPT_SLAP_INITIAL_LOAD
}, /* Number to load initially */
327 {(OPTIONSTRING
) "non-blocking", no_argument
, &opt_non_blocking_io
, OPT_SLAP_NON_BLOCK
},
328 {(OPTIONSTRING
) "servers", required_argument
, NULL
, OPT_SERVERS
},
329 {(OPTIONSTRING
) "tcp-nodelay", no_argument
, &opt_tcp_nodelay
, OPT_SLAP_TCP_NODELAY
},
330 {(OPTIONSTRING
) "test", required_argument
, NULL
, OPT_SLAP_TEST
},
331 {(OPTIONSTRING
) "verbose", no_argument
, &opt_verbose
, OPT_VERBOSE
},
332 {(OPTIONSTRING
) "version", no_argument
, NULL
, OPT_VERSION
},
333 {(OPTIONSTRING
) "binary", no_argument
, NULL
, OPT_BINARY
},
334 {(OPTIONSTRING
) "udp", no_argument
, NULL
, OPT_UDP
},
338 bool opt_help
= false;
339 bool opt_version
= false;
340 int option_index
= 0;
342 int option_rv
= getopt_long(argc
, argv
, "Vhvds:", long_options
, &option_index
);
351 if (opt_test
== GET_TEST
) {
353 "You can not run a get test in UDP mode. UDP mode "
354 "does not currently support get ops.\n");
360 case OPT_BINARY
: opt_binary
= true; break;
362 case OPT_VERBOSE
: /* --verbose or -v */ opt_verbose
= OPT_VERBOSE
; break;
364 case OPT_DEBUG
: /* --debug or -d */ opt_verbose
= OPT_DEBUG
; break;
366 case OPT_VERSION
: /* --version or -V */ opt_version
= true; break;
368 case OPT_HELP
: /* --help or -h */ opt_help
= true; break;
370 case OPT_SERVERS
: /* --servers or -s */ opt_servers
= strdup(optarg
); break;
373 if (strcmp(optarg
, "get") == 0) {
374 if (opt_udp_io
== 1) {
376 "You can not run a get test in UDP mode. UDP mode "
377 "does not currently support get ops.\n");
381 } else if (strcmp(optarg
, "set") == 0) {
383 } else if (strcmp(optarg
, "mget") == 0) {
384 opt_test
= MGET_TEST
;
386 fprintf(stderr
, "Your test, %s, is not a known test\n", optarg
);
391 case OPT_SLAP_CONCURRENCY
:
393 opt_concurrency
= (unsigned int) strtoul(optarg
, (char **) NULL
, 10);
395 fprintf(stderr
, "Invalid value for concurrency: %s\n", optarg
);
400 case OPT_SLAP_EXECUTE_NUMBER
:
402 opt_execute_number
= (unsigned int) strtoul(optarg
, (char **) NULL
, 10);
404 fprintf(stderr
, "Invalid value for execute: %s\n", optarg
);
409 case OPT_SLAP_INITIAL_LOAD
:
411 opt_createial_load
= (unsigned int) strtoul(optarg
, (char **) NULL
, 10);
413 fprintf(stderr
, "Invalid value for initial load: %s\n", optarg
);
418 case OPT_QUIET
: close_stdio(); break;
421 /* getopt_long already printed an error message. */
429 version_command(PROGRAM_NAME
);
434 help_command(PROGRAM_NAME
, PROGRAM_DESCRIPTION
, long_options
, help_options
);
438 if ((opt_test
== GET_TEST
or opt_test
== MGET_TEST
) and opt_createial_load
== 0)
439 opt_createial_load
= DEFAULT_INITIAL_LOAD
;
441 if (opt_execute_number
== 0)
442 opt_execute_number
= DEFAULT_EXECUTE_NUMBER
;
444 if (opt_concurrency
== 0)
445 opt_concurrency
= DEFAULT_CONCURRENCY
;
448 void conclusions_print(conclusions_st
*conclusion
) {
449 printf("\tThreads connecting to servers %u\n", opt_concurrency
);
451 printf("\tLoaded %u rows\n", conclusion
->rows_loaded
);
452 printf("\tRead %u rows\n", conclusion
->rows_read
);
454 if (opt_test
== SET_TEST
)
455 printf("\tTook %ld.%03ld seconds to load data\n", conclusion
->load_time
/ 1000,
456 conclusion
->load_time
% 1000);
458 printf("\tTook %ld.%03ld seconds to read data\n", conclusion
->read_time
/ 1000,
459 conclusion
->read_time
% 1000);
462 void flush_all(memcached_st
*memc
) {
463 memcached_flush(memc
, 0);
466 pairs_st
*load_create_data(memcached_st
*memc
, unsigned int number_of
,
467 unsigned int *actual_loaded
) {
468 memcached_st
*memc_clone
= memcached_clone(NULL
, memc
);
469 /* We always used non-blocking IO for load since it is faster */
470 memcached_behavior_set(memc_clone
, MEMCACHED_BEHAVIOR_NO_BLOCK
, 0);
472 pairs_st
*pairs
= pairs_generate(number_of
, VALUE_BYTES
);
473 *actual_loaded
= execute_set(memc_clone
, pairs
, number_of
);
475 memcached_free(memc_clone
);