git.m6w6.name Git - awesomized/libmemcached/blob - clients/memslap.cc

   1 /*  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
   2  *
   3  *  Libmemcached library
   4  *
   5  *  Copyright (C) 2011 Data Differential, http://datadifferential.com/
   6  *  Copyright (C) 2006-2009 Brian Aker All rights reserved.
   7  *
   8  *  Redistribution and use in source and binary forms, with or without
   9  *  modification, are permitted provided that the following conditions are
  10  *  met:
  11  *
  12  *      * Redistributions of source code must retain the above copyright
  13  *  notice, this list of conditions and the following disclaimer.
  14  *
  15  *      * Redistributions in binary form must reproduce the above
  16  *  copyright notice, this list of conditions and the following disclaimer
  17  *  in the documentation and/or other materials provided with the
  18  *  distribution.
  19  *
  20  *      * The names of its contributors may not be used to endorse or
  21  *  promote products derived from this software without specific prior
  22  *  written permission.
  23  *
  24  *  THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  25  *  "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  26  *  LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  27  *  A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  28  *  OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  29  *  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  30  *  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  31  *  DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  32  *  THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  33  *  (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  34  *  OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  35  *
  36  */
  37
  38
  39 #include <config.h>
  40
  41 #include <cassert>
  42 #include <cstdio>
  43 #include <cstdlib>
  44 #include <cstring>
  45 #include <fcntl.h>
  46 #include <getopt.h>
  47 #include <memory>
  48 #include <pthread.h>
  49 #include <sys/mman.h>
  50 #include <sys/stat.h>
  51 #include <sys/time.h>
  52 #include <sys/types.h>
  53 #include <sys/types.h>
  54
  55 #include <libmemcached/memcached.h>
  56
  57 #include "client_options.h"
  58 #include "utilities.h"
  59 #include "generator.h"
  60 #include "execute.h"
  61
  62 #define DEFAULT_INITIAL_LOAD 10000
  63 #define DEFAULT_EXECUTE_NUMBER 10000
  64 #define DEFAULT_CONCURRENCY 1
  65
  66 #define PROGRAM_NAME "memslap"
  67 #define PROGRAM_DESCRIPTION "Generates a load against a memcached custer of servers."
  68
  69 /* Global Thread counter */
  70 volatile unsigned int master_wakeup;
  71 pthread_mutex_t sleeper_mutex;
  72 pthread_cond_t sleep_threshhold;
  73
  74 /* Types */
  75 enum test_t {
  76   SET_TEST,
  77   GET_TEST,
  78   MGET_TEST
  79 };
  80
  81 struct thread_context_st {
  82   unsigned int key_count;
  83   pairs_st *initial_pairs;
  84   unsigned int initial_number;
  85   pairs_st *execute_pairs;
  86   unsigned int execute_number;
  87   char **keys;
  88   size_t *key_lengths;
  89   test_t test;
  90   memcached_st *memc;
  91   const memcached_st* root;
  92
  93   thread_context_st(const memcached_st* memc_arg, test_t test_arg) :
  94     key_count(0),
  95     initial_pairs(NULL),
  96     initial_number(0),
  97     execute_pairs(NULL),
  98     execute_number(0),
  99     keys(0),
 100     key_lengths(NULL),
 101     test(test_arg),
 102     memc(NULL),
 103     root(memc_arg)
 104   {
 105   }
 106
 107   void init()
 108   {
 109     memc= memcached_clone(NULL, root);
 110   }
 111
 112   ~thread_context_st()
 113   {
 114     if (execute_pairs)
 115     {
 116       pairs_free(execute_pairs);
 117     }
 118     memcached_free(memc);
 119   }
 120 };
 121
 122 struct conclusions_st {
 123   long int load_time;
 124   long int read_time;
 125   unsigned int rows_loaded;
 126   unsigned int rows_read;
 127
 128   conclusions_st() :
 129     load_time(0),
 130     read_time(0),
 131     rows_loaded(0),
 132     rows_read()
 133   { }
 134 };
 135
 136 /* Prototypes */
 137 void options_parse(int argc, char *argv[]);
 138 void conclusions_print(conclusions_st *conclusion);
 139 void scheduler(memcached_server_st *servers, conclusions_st *conclusion);
 140 pairs_st *load_create_data(memcached_st *memc, unsigned int number_of,
 141                            unsigned int *actual_loaded);
 142 void flush_all(memcached_st *memc);
 143
 144 static int opt_binary= 0;
 145 static int opt_verbose= 0;
 146 static int opt_flush= 0;
 147 static int opt_non_blocking_io= 0;
 148 static int opt_tcp_nodelay= 0;
 149 static unsigned int opt_execute_number= 0;
 150 static unsigned int opt_createial_load= 0;
 151 static unsigned int opt_concurrency= 0;
 152 static int opt_displayflag= 0;
 153 static char *opt_servers= NULL;
 154 static int opt_udp_io= 0;
 155 test_t opt_test= SET_TEST;
 156
 157 extern "C" {
 158
 159 static void *run_task(void *p)
 160 {
 161   thread_context_st *context= (thread_context_st *)p;
 162
 163   context->init();
 164
 165   pthread_mutex_lock(&sleeper_mutex);
 166   while (master_wakeup)
 167   {
 168     pthread_cond_wait(&sleep_threshhold, &sleeper_mutex);
 169   }
 170   pthread_mutex_unlock(&sleeper_mutex);
 171
 172   /* Do Stuff */
 173   switch (context->test)
 174   {
 175   case SET_TEST:
 176     assert(context->execute_pairs);
 177     execute_set(context->memc, context->execute_pairs, context->execute_number);
 178     break;
 179
 180   case GET_TEST:
 181     execute_get(context->memc, context->initial_pairs, context->initial_number);
 182     break;
 183
 184   case MGET_TEST:
 185     execute_mget(context->memc, (const char*const*)context->keys, context->key_lengths, context->initial_number);
 186     break;
 187   }
 188
 189   delete context;
 190
 191   pthread_exit(0);
 192 }
 193
 194 }
 195
 196
 197 int main(int argc, char *argv[])
 198 {
 199   conclusions_st conclusion;
 200
 201   srandom((unsigned int)time(NULL));
 202   options_parse(argc, argv);
 203
 204   if (opt_servers == NULL)
 205   {
 206     char *temp;
 207
 208     if ((temp= getenv("MEMCACHED_SERVERS")))
 209     {
 210       opt_servers= strdup(temp);
 211     }
 212     else
 213     {
 214       fprintf(stderr, "No Servers provided\n");
 215       return EXIT_FAILURE;
 216     }
 217   }
 218
 219   memcached_server_st *servers= memcached_servers_parse(opt_servers);
 220
 221   pthread_mutex_init(&sleeper_mutex, NULL);
 222   pthread_cond_init(&sleep_threshhold, NULL);
 223
 224   scheduler(servers, &conclusion);
 225
 226   free(opt_servers);
 227
 228   (void)pthread_mutex_destroy(&sleeper_mutex);
 229   (void)pthread_cond_destroy(&sleep_threshhold);
 230   conclusions_print(&conclusion);
 231   memcached_server_list_free(servers);
 232
 233   return EXIT_SUCCESS;
 234 }
 235
 236 void scheduler(memcached_server_st *servers, conclusions_st *conclusion)
 237 {
 238   unsigned int actual_loaded= 0; /* Fix warning */
 239
 240   struct timeval start_time, end_time;
 241   pairs_st *pairs= NULL;
 242
 243   memcached_st *memc= memcached_create(NULL);
 244
 245   /* We need to set udp behavior before adding servers to the client */
 246   if (opt_udp_io)
 247   {
 248     memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_USE_UDP,
 249                            (uint64_t)opt_udp_io);
 250     for (uint32_t x= 0; x < memcached_server_list_count(servers); x++ )
 251     {
 252       servers[x].type= MEMCACHED_CONNECTION_UDP;
 253     }
 254   }
 255   memcached_server_push(memc, servers);
 256
 257   memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_BINARY_PROTOCOL,
 258                          (uint64_t)opt_binary);
 259
 260   if (opt_flush)
 261     flush_all(memc);
 262
 263   if (opt_createial_load)
 264     pairs= load_create_data(memc, opt_createial_load, &actual_loaded);
 265
 266   char **keys= static_cast<char **>(calloc(actual_loaded, sizeof(char*)));
 267   size_t *key_lengths= static_cast<size_t *>(calloc(actual_loaded, sizeof(size_t)));
 268
 269   if (keys == NULL or key_lengths == NULL)
 270   {
 271     free(keys);
 272     free(key_lengths);
 273     keys= NULL;
 274     key_lengths= NULL;
 275   }
 276   else
 277   {
 278     for (uint32_t x= 0; x < actual_loaded; ++x)
 279     {
 280       keys[x]= pairs[x].key;
 281       key_lengths[x]= pairs[x].key_length;
 282     }
 283   }
 284
 285   /* We set this after we have loaded */
 286   {
 287     if (opt_non_blocking_io)
 288       memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_NO_BLOCK, 1);
 289
 290     if (opt_tcp_nodelay)
 291       memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_TCP_NODELAY, 1);
 292   }
 293
 294   pthread_mutex_lock(&sleeper_mutex);
 295   master_wakeup= 1;
 296   pthread_mutex_unlock(&sleeper_mutex);
 297
 298   pthread_t *threads= new  (std::nothrow) pthread_t[opt_concurrency];
 299
 300   if (not threads)
 301   {
 302     exit(EXIT_FAILURE);
 303   }
 304
 305   for (uint32_t x= 0; x < opt_concurrency; x++)
 306   {
 307     thread_context_st *context= new thread_context_st(memc, opt_test);
 308     context->test= opt_test;
 309
 310     context->initial_pairs= pairs;
 311     context->initial_number= actual_loaded;
 312     context->keys= keys;
 313     context->key_lengths= key_lengths;
 314
 315     if (opt_test == SET_TEST)
 316     {
 317       context->execute_pairs= pairs_generate(opt_execute_number, 400);
 318       context->execute_number= opt_execute_number;
 319     }
 320
 321     /* now you create the thread */
 322     if (pthread_create(threads +x, NULL, run_task, (void *)context) != 0)
 323     {
 324       fprintf(stderr,"Could not create thread\n");
 325       exit(1);
 326     }
 327   }
 328
 329   pthread_mutex_lock(&sleeper_mutex);
 330   master_wakeup= 0;
 331   pthread_mutex_unlock(&sleeper_mutex);
 332   pthread_cond_broadcast(&sleep_threshhold);
 333   gettimeofday(&start_time, NULL);
 334
 335   for (uint32_t x= 0; x < opt_concurrency; x++)
 336   {
 337     void *retval;
 338     pthread_join(threads[x], &retval);
 339   }
 340   delete [] threads;
 341
 342   gettimeofday(&end_time, NULL);
 343
 344   conclusion->load_time= timedif(end_time, start_time);
 345   conclusion->read_time= timedif(end_time, start_time);
 346   free(keys);
 347   free(key_lengths);
 348   pairs_free(pairs);
 349   memcached_free(memc);
 350 }
 351
 352 void options_parse(int argc, char *argv[])
 353 {
 354   memcached_programs_help_st help_options[]=
 355   {
 356     {0},
 357   };
 358
 359   static struct option long_options[]=
 360     {
 361       {(OPTIONSTRING)"concurrency", required_argument, NULL, OPT_SLAP_CONCURRENCY},
 362       {(OPTIONSTRING)"debug", no_argument, &opt_verbose, OPT_DEBUG},
 363       {(OPTIONSTRING)"execute-number", required_argument, NULL, OPT_SLAP_EXECUTE_NUMBER},
 364       {(OPTIONSTRING)"flag", no_argument, &opt_displayflag, OPT_FLAG},
 365       {(OPTIONSTRING)"flush", no_argument, &opt_flush, OPT_FLUSH},
 366       {(OPTIONSTRING)"help", no_argument, NULL, OPT_HELP},
 367       {(OPTIONSTRING)"initial-load", required_argument, NULL, OPT_SLAP_INITIAL_LOAD}, /* Number to load initially */
 368       {(OPTIONSTRING)"non-blocking", no_argument, &opt_non_blocking_io, OPT_SLAP_NON_BLOCK},
 369       {(OPTIONSTRING)"servers", required_argument, NULL, OPT_SERVERS},
 370       {(OPTIONSTRING)"tcp-nodelay", no_argument, &opt_tcp_nodelay, OPT_SLAP_TCP_NODELAY},
 371       {(OPTIONSTRING)"test", required_argument, NULL, OPT_SLAP_TEST},
 372       {(OPTIONSTRING)"verbose", no_argument, &opt_verbose, OPT_VERBOSE},
 373       {(OPTIONSTRING)"version", no_argument, NULL, OPT_VERSION},
 374       {(OPTIONSTRING)"binary", no_argument, NULL, OPT_BINARY},
 375       {(OPTIONSTRING)"udp", no_argument, NULL, OPT_UDP},
 376       {0, 0, 0, 0},
 377     };
 378
 379   int option_index= 0;
 380   int option_rv;
 381
 382   while (1)
 383   {
 384     option_rv= getopt_long(argc, argv, "Vhvds:", long_options, &option_index);
 385     if (option_rv == -1) break;
 386     switch (option_rv)
 387     {
 388     case 0:
 389       break;
 390
 391     case OPT_UDP:
 392       if (opt_test == GET_TEST)
 393       {
 394         fprintf(stderr, "You can not run a get test in UDP mode. UDP mode "
 395                   "does not currently support get ops.\n");
 396         exit(1);
 397       }
 398       opt_udp_io= 1;
 399       break;
 400
 401     case OPT_BINARY:
 402       opt_binary = 1;
 403       break;
 404
 405     case OPT_VERBOSE: /* --verbose or -v */
 406       opt_verbose = OPT_VERBOSE;
 407       break;
 408
 409     case OPT_DEBUG: /* --debug or -d */
 410       opt_verbose = OPT_DEBUG;
 411       break;
 412
 413     case OPT_VERSION: /* --version or -V */
 414       version_command(PROGRAM_NAME);
 415       break;
 416
 417     case OPT_HELP: /* --help or -h */
 418       help_command(PROGRAM_NAME, PROGRAM_DESCRIPTION, long_options, help_options);
 419       break;
 420
 421     case OPT_SERVERS: /* --servers or -s */
 422       opt_servers= strdup(optarg);
 423       break;
 424
 425     case OPT_SLAP_TEST:
 426       if (!strcmp(optarg, "get"))
 427       {
 428         if (opt_udp_io == 1)
 429         {
 430           fprintf(stderr, "You can not run a get test in UDP mode. UDP mode "
 431                   "does not currently support get ops.\n");
 432           exit(1);
 433         }
 434         opt_test= GET_TEST ;
 435       }
 436       else if (!strcmp(optarg, "set"))
 437       {
 438         opt_test= SET_TEST;
 439       }
 440       else if (!strcmp(optarg, "mget"))
 441       {
 442         opt_test= MGET_TEST;
 443       }
 444       else
 445       {
 446         fprintf(stderr, "Your test, %s, is not a known test\n", optarg);
 447         exit(1);
 448       }
 449       break;
 450
 451     case OPT_SLAP_CONCURRENCY:
 452       opt_concurrency= (unsigned int)strtoul(optarg, (char **)NULL, 10);
 453       break;
 454
 455     case OPT_SLAP_EXECUTE_NUMBER:
 456       opt_execute_number= (unsigned int)strtoul(optarg, (char **)NULL, 10);
 457       break;
 458
 459     case OPT_SLAP_INITIAL_LOAD:
 460       opt_createial_load= (unsigned int)strtoul(optarg, (char **)NULL, 10);
 461       break;
 462
 463     case '?':
 464       /* getopt_long already printed an error message. */
 465       exit(1);
 466
 467     default:
 468       abort();
 469     }
 470   }
 471
 472   if ((opt_test == GET_TEST or opt_test == MGET_TEST) && opt_createial_load == 0)
 473     opt_createial_load= DEFAULT_INITIAL_LOAD;
 474
 475   if (opt_execute_number == 0)
 476     opt_execute_number= DEFAULT_EXECUTE_NUMBER;
 477
 478   if (opt_concurrency == 0)
 479     opt_concurrency= DEFAULT_CONCURRENCY;
 480 }
 481
 482 void conclusions_print(conclusions_st *conclusion)
 483 {
 484   printf("\tThreads connecting to servers %u\n", opt_concurrency);
 485 #ifdef NOT_FINISHED
 486   printf("\tLoaded %u rows\n", conclusion->rows_loaded);
 487   printf("\tRead %u rows\n", conclusion->rows_read);
 488 #endif
 489   if (opt_test == SET_TEST)
 490     printf("\tTook %ld.%03ld seconds to load data\n", conclusion->load_time / 1000,
 491            conclusion->load_time % 1000);
 492   else
 493     printf("\tTook %ld.%03ld seconds to read data\n", conclusion->read_time / 1000,
 494            conclusion->read_time % 1000);
 495 }
 496
 497 void flush_all(memcached_st *memc)
 498 {
 499   memcached_flush(memc, 0);
 500 }
 501
 502 pairs_st *load_create_data(memcached_st *memc, unsigned int number_of,
 503                            unsigned int *actual_loaded)
 504 {
 505   memcached_st *memc_clone= memcached_clone(NULL, memc);
 506   /* We always used non-blocking IO for load since it is faster */
 507   memcached_behavior_set(memc_clone, MEMCACHED_BEHAVIOR_NO_BLOCK, 0);
 508
 509   pairs_st *pairs= pairs_generate(number_of, 400);
 510   *actual_loaded= execute_set(memc_clone, pairs, number_of);
 511
 512   memcached_free(memc_clone);
 513
 514   return pairs;
 515 }