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-2012 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 <mem_config.h>
  40
  41 #include <cassert>
  42 #include <cerrno>
  43 #include <cstdio>
  44 #include <cstdlib>
  45 #include <cstring>
  46 #include <fcntl.h>
  47 #include <getopt.h>
  48 #include <memory>
  49 #include <pthread.h>
  50 #include <sys/mman.h>
  51 #include <sys/stat.h>
  52 #include <sys/time.h>
  53 #include <sys/types.h>
  54 #include <unistd.h>
  55
  56 #include <iostream>
  57
  58 #include <libmemcached-1.0/memcached.h>
  59
  60 #include "client_options.h"
  61 #include "utilities.h"
  62 #include "generator.h"
  63 #include "execute.h"
  64
  65 #define DEFAULT_INITIAL_LOAD 10000
  66 #define DEFAULT_EXECUTE_NUMBER 10000
  67 #define DEFAULT_CONCURRENCY 1
  68
  69 #define PROGRAM_NAME "memslap"
  70 #define PROGRAM_DESCRIPTION "Generates a load against a memcached custer of servers."
  71
  72 /* Global Thread counter */
  73 volatile unsigned int master_wakeup;
  74 pthread_mutex_t sleeper_mutex;
  75 pthread_cond_t sleep_threshhold;
  76
  77 /* Types */
  78 enum test_t {
  79   SET_TEST,
  80   GET_TEST,
  81   MGET_TEST
  82 };
  83
  84 struct thread_context_st {
  85   unsigned int key_count;
  86   pairs_st *initial_pairs;
  87   unsigned int initial_number;
  88   pairs_st *execute_pairs;
  89   unsigned int execute_number;
  90   char **keys;
  91   size_t *key_lengths;
  92   test_t test;
  93   memcached_st *memc;
  94   const memcached_st* root;
  95
  96   thread_context_st(const memcached_st* memc_arg, test_t test_arg) :
  97     key_count(0),
  98     initial_pairs(NULL),
  99     initial_number(0),
 100     execute_pairs(NULL),
 101     execute_number(0),
 102     keys(0),
 103     key_lengths(NULL),
 104     test(test_arg),
 105     memc(NULL),
 106     root(memc_arg)
 107   {
 108   }
 109
 110   void init()
 111   {
 112     memc= memcached_clone(NULL, root);
 113   }
 114
 115   ~thread_context_st()
 116   {
 117     if (execute_pairs)
 118     {
 119       pairs_free(execute_pairs);
 120     }
 121     memcached_free(memc);
 122   }
 123 };
 124
 125 struct conclusions_st {
 126   long int load_time;
 127   long int read_time;
 128   unsigned int rows_loaded;
 129   unsigned int rows_read;
 130
 131   conclusions_st() :
 132     load_time(0),
 133     read_time(0),
 134     rows_loaded(0),
 135     rows_read()
 136   { }
 137 };
 138
 139 /* Prototypes */
 140 void options_parse(int argc, char *argv[]);
 141 void conclusions_print(conclusions_st *conclusion);
 142 void scheduler(memcached_server_st *servers, conclusions_st *conclusion);
 143 pairs_st *load_create_data(memcached_st *memc, unsigned int number_of,
 144                            unsigned int *actual_loaded);
 145 void flush_all(memcached_st *memc);
 146
 147 static bool opt_binary= 0;
 148 static int opt_verbose= 0;
 149 static int opt_flush= 0;
 150 static int opt_non_blocking_io= 0;
 151 static int opt_tcp_nodelay= 0;
 152 static unsigned int opt_execute_number= 0;
 153 static unsigned int opt_createial_load= 0;
 154 static unsigned int opt_concurrency= 0;
 155 static int opt_displayflag= 0;
 156 static char *opt_servers= NULL;
 157 static bool opt_udp_io= false;
 158 test_t opt_test= SET_TEST;
 159
 160 extern "C" {
 161
 162 static __attribute__((noreturn)) void *run_task(void *p)
 163 {
 164   thread_context_st *context= (thread_context_st *)p;
 165
 166   context->init();
 167
 168   pthread_mutex_lock(&sleeper_mutex);
 169   while (master_wakeup)
 170   {
 171     pthread_cond_wait(&sleep_threshhold, &sleeper_mutex);
 172   }
 173   pthread_mutex_unlock(&sleeper_mutex);
 174
 175   /* Do Stuff */
 176   switch (context->test)
 177   {
 178   case SET_TEST:
 179     assert(context->execute_pairs);
 180     execute_set(context->memc, context->execute_pairs, context->execute_number);
 181     break;
 182
 183   case GET_TEST:
 184     execute_get(context->memc, context->initial_pairs, context->initial_number);
 185     break;
 186
 187   case MGET_TEST:
 188     execute_mget(context->memc, (const char*const*)context->keys, context->key_lengths, context->initial_number);
 189     break;
 190   }
 191
 192   delete context;
 193
 194   pthread_exit(0);
 195 }
 196
 197 }
 198
 199
 200 int main(int argc, char *argv[])
 201 {
 202   conclusions_st conclusion;
 203
 204   srandom((unsigned int)time(NULL));
 205   options_parse(argc, argv);
 206
 207   if (opt_servers == NULL)
 208   {
 209     char *temp;
 210
 211     if ((temp= getenv("MEMCACHED_SERVERS")))
 212     {
 213       opt_servers= strdup(temp);
 214     }
 215
 216     if (opt_servers == NULL)
 217     {
 218       std::cerr << "No Servers provided" << std::endl;
 219       exit(EXIT_FAILURE);
 220     }
 221   }
 222
 223   memcached_server_st *servers= memcached_servers_parse(opt_servers);
 224   if (servers == NULL or memcached_server_list_count(servers) == 0)
 225   {
 226     std::cerr << "Invalid server list provided:" << opt_servers << std::endl;
 227     return EXIT_FAILURE;
 228   }
 229
 230   pthread_mutex_init(&sleeper_mutex, NULL);
 231   pthread_cond_init(&sleep_threshhold, NULL);
 232
 233   int error_code= EXIT_SUCCESS;
 234   try {
 235     scheduler(servers, &conclusion);
 236   }
 237   catch(std::exception& e)
 238   {
 239     std::cerr << "Died with exception: " << e.what() << std::endl;
 240     error_code= EXIT_FAILURE;
 241   }
 242
 243   free(opt_servers);
 244
 245   (void)pthread_mutex_destroy(&sleeper_mutex);
 246   (void)pthread_cond_destroy(&sleep_threshhold);
 247   conclusions_print(&conclusion);
 248   memcached_server_list_free(servers);
 249
 250   return error_code;
 251 }
 252
 253 void scheduler(memcached_server_st *servers, conclusions_st *conclusion)
 254 {
 255   unsigned int actual_loaded= 0; /* Fix warning */
 256
 257   struct timeval start_time, end_time;
 258   pairs_st *pairs= NULL;
 259
 260   memcached_st *memc= memcached_create(NULL);
 261
 262   memcached_server_push(memc, servers);
 263
 264   /* We need to set udp behavior before adding servers to the client */
 265   if (opt_udp_io)
 266   {
 267     if (memcached_failed(memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_USE_UDP, opt_udp_io)))
 268     {
 269       std::cerr << "Failed to enable UDP." << std::endl;
 270       memcached_free(memc);
 271       exit(EXIT_FAILURE);
 272     }
 273   }
 274
 275   memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_BINARY_PROTOCOL,
 276                          (uint64_t)opt_binary);
 277
 278   if (opt_flush)
 279   {
 280     flush_all(memc);
 281   }
 282
 283   if (opt_createial_load)
 284   {
 285     pairs= load_create_data(memc, opt_createial_load, &actual_loaded);
 286   }
 287
 288   char **keys= static_cast<char **>(calloc(actual_loaded, sizeof(char*)));
 289   size_t *key_lengths= static_cast<size_t *>(calloc(actual_loaded, sizeof(size_t)));
 290
 291   if (keys == NULL or key_lengths == NULL)
 292   {
 293     free(keys);
 294     free(key_lengths);
 295     keys= NULL;
 296     key_lengths= NULL;
 297   }
 298   else
 299   {
 300     for (uint32_t x= 0; x < actual_loaded; ++x)
 301     {
 302       keys[x]= pairs[x].key;
 303       key_lengths[x]= pairs[x].key_length;
 304     }
 305   }
 306
 307   /* We set this after we have loaded */
 308   {
 309     if (opt_non_blocking_io)
 310       memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_NO_BLOCK, 1);
 311
 312     if (opt_tcp_nodelay)
 313       memcached_behavior_set(memc, MEMCACHED_BEHAVIOR_TCP_NODELAY, 1);
 314   }
 315
 316   pthread_mutex_lock(&sleeper_mutex);
 317   master_wakeup= 1;
 318   pthread_mutex_unlock(&sleeper_mutex);
 319
 320   pthread_t *threads= new  (std::nothrow) pthread_t[opt_concurrency];
 321
 322   if (threads == NULL)
 323   {
 324     exit(EXIT_FAILURE);
 325   }
 326
 327   for (uint32_t x= 0; x < opt_concurrency; x++)
 328   {
 329     thread_context_st *context= new thread_context_st(memc, opt_test);
 330     context->test= opt_test;
 331
 332     context->initial_pairs= pairs;
 333     context->initial_number= actual_loaded;
 334     context->keys= keys;
 335     context->key_lengths= key_lengths;
 336
 337     if (opt_test == SET_TEST)
 338     {
 339       context->execute_pairs= pairs_generate(opt_execute_number, 400);
 340       context->execute_number= opt_execute_number;
 341     }
 342
 343     /* now you create the thread */
 344     if (pthread_create(threads +x, NULL, run_task, (void *)context) != 0)
 345     {
 346       fprintf(stderr,"Could not create thread\n");
 347       exit(1);
 348     }
 349   }
 350
 351   pthread_mutex_lock(&sleeper_mutex);
 352   master_wakeup= 0;
 353   pthread_mutex_unlock(&sleeper_mutex);
 354   pthread_cond_broadcast(&sleep_threshhold);
 355   gettimeofday(&start_time, NULL);
 356
 357   for (uint32_t x= 0; x < opt_concurrency; x++)
 358   {
 359     void *retval;
 360     pthread_join(threads[x], &retval);
 361   }
 362   delete [] threads;
 363
 364   gettimeofday(&end_time, NULL);
 365
 366   conclusion->load_time= timedif(end_time, start_time);
 367   conclusion->read_time= timedif(end_time, start_time);
 368   free(keys);
 369   free(key_lengths);
 370   pairs_free(pairs);
 371   memcached_free(memc);
 372 }
 373
 374 void options_parse(int argc, char *argv[])
 375 {
 376   memcached_programs_help_st help_options[]=
 377   {
 378     {0},
 379   };
 380
 381   static struct option long_options[]=
 382     {
 383       {(OPTIONSTRING)"concurrency", required_argument, NULL, OPT_SLAP_CONCURRENCY},
 384       {(OPTIONSTRING)"debug", no_argument, &opt_verbose, OPT_DEBUG},
 385       {(OPTIONSTRING)"quiet", no_argument, NULL, OPT_QUIET},
 386       {(OPTIONSTRING)"execute-number", required_argument, NULL, OPT_SLAP_EXECUTE_NUMBER},
 387       {(OPTIONSTRING)"flag", no_argument, &opt_displayflag, OPT_FLAG},
 388       {(OPTIONSTRING)"flush", no_argument, &opt_flush, OPT_FLUSH},
 389       {(OPTIONSTRING)"help", no_argument, NULL, OPT_HELP},
 390       {(OPTIONSTRING)"initial-load", required_argument, NULL, OPT_SLAP_INITIAL_LOAD}, /* Number to load initially */
 391       {(OPTIONSTRING)"non-blocking", no_argument, &opt_non_blocking_io, OPT_SLAP_NON_BLOCK},
 392       {(OPTIONSTRING)"servers", required_argument, NULL, OPT_SERVERS},
 393       {(OPTIONSTRING)"tcp-nodelay", no_argument, &opt_tcp_nodelay, OPT_SLAP_TCP_NODELAY},
 394       {(OPTIONSTRING)"test", required_argument, NULL, OPT_SLAP_TEST},
 395       {(OPTIONSTRING)"verbose", no_argument, &opt_verbose, OPT_VERBOSE},
 396       {(OPTIONSTRING)"version", no_argument, NULL, OPT_VERSION},
 397       {(OPTIONSTRING)"binary", no_argument, NULL, OPT_BINARY},
 398       {(OPTIONSTRING)"udp", no_argument, NULL, OPT_UDP},
 399       {0, 0, 0, 0},
 400     };
 401
 402   bool opt_help= false;
 403   bool opt_version= false;
 404   int option_index= 0;
 405   while (1)
 406   {
 407     int option_rv= getopt_long(argc, argv, "Vhvds:", long_options, &option_index);
 408
 409     if (option_rv == -1) break;
 410
 411     switch (option_rv)
 412     {
 413     case 0:
 414       break;
 415
 416     case OPT_UDP:
 417       if (opt_test == GET_TEST)
 418       {
 419         fprintf(stderr, "You can not run a get test in UDP mode. UDP mode "
 420                   "does not currently support get ops.\n");
 421         exit(1);
 422       }
 423       opt_udp_io= true;
 424       break;
 425
 426     case OPT_BINARY:
 427       opt_binary= true;
 428       break;
 429
 430     case OPT_VERBOSE: /* --verbose or -v */
 431       opt_verbose= OPT_VERBOSE;
 432       break;
 433
 434     case OPT_DEBUG: /* --debug or -d */
 435       opt_verbose = OPT_DEBUG;
 436       break;
 437
 438     case OPT_VERSION: /* --version or -V */
 439       opt_version= true;
 440       break;
 441
 442     case OPT_HELP: /* --help or -h */
 443       opt_help= true;
 444       break;
 445
 446     case OPT_SERVERS: /* --servers or -s */
 447       opt_servers= strdup(optarg);
 448       break;
 449
 450     case OPT_SLAP_TEST:
 451       if (strcmp(optarg, "get") == 0)
 452       {
 453         if (opt_udp_io == 1)
 454         {
 455           fprintf(stderr, "You can not run a get test in UDP mode. UDP mode "
 456                   "does not currently support get ops.\n");
 457           exit(EXIT_FAILURE);
 458         }
 459         opt_test= GET_TEST ;
 460       }
 461       else if (strcmp(optarg, "set") == 0)
 462       {
 463         opt_test= SET_TEST;
 464       }
 465       else if (strcmp(optarg, "mget") == 0)
 466       {
 467         opt_test= MGET_TEST;
 468       }
 469       else
 470       {
 471         fprintf(stderr, "Your test, %s, is not a known test\n", optarg);
 472         exit(EXIT_FAILURE);
 473       }
 474       break;
 475
 476     case OPT_SLAP_CONCURRENCY:
 477       errno= 0;
 478       opt_concurrency= (unsigned int)strtoul(optarg, (char **)NULL, 10);
 479       if (errno != 0)
 480       {
 481         fprintf(stderr, "Invalid value for concurrency: %s\n", optarg);
 482         exit(EXIT_FAILURE);
 483       }
 484       break;
 485
 486     case OPT_SLAP_EXECUTE_NUMBER:
 487       errno= 0;
 488       opt_execute_number= (unsigned int)strtoul(optarg, (char **)NULL, 10);
 489       if (errno != 0)
 490       {
 491         fprintf(stderr, "Invalid value for execute: %s\n", optarg);
 492         exit(EXIT_FAILURE);
 493       }
 494       break;
 495
 496     case OPT_SLAP_INITIAL_LOAD:
 497       errno= 0;
 498       opt_createial_load= (unsigned int)strtoul(optarg, (char **)NULL, 10);
 499       if (errno != 0)
 500       {
 501         fprintf(stderr, "Invalid value for initial load: %s\n", optarg);
 502         exit(EXIT_FAILURE);
 503       }
 504       break;
 505
 506     case OPT_QUIET:
 507       close_stdio();
 508       break;
 509
 510
 511     case '?':
 512       /* getopt_long already printed an error message. */
 513       exit(EXIT_FAILURE);
 514
 515     default:
 516       abort();
 517     }
 518   }
 519
 520   if (opt_version)
 521   {
 522     version_command(PROGRAM_NAME);
 523     exit(EXIT_SUCCESS);
 524   }
 525
 526   if (opt_help)
 527   {
 528     help_command(PROGRAM_NAME, PROGRAM_DESCRIPTION, long_options, help_options);
 529     exit(EXIT_SUCCESS);
 530   }
 531
 532   if ((opt_test == GET_TEST or opt_test == MGET_TEST) and opt_createial_load == 0)
 533     opt_createial_load= DEFAULT_INITIAL_LOAD;
 534
 535   if (opt_execute_number == 0)
 536     opt_execute_number= DEFAULT_EXECUTE_NUMBER;
 537
 538   if (opt_concurrency == 0)
 539     opt_concurrency= DEFAULT_CONCURRENCY;
 540 }
 541
 542 void conclusions_print(conclusions_st *conclusion)
 543 {
 544   printf("\tThreads connecting to servers %u\n", opt_concurrency);
 545 #ifdef NOT_FINISHED
 546   printf("\tLoaded %u rows\n", conclusion->rows_loaded);
 547   printf("\tRead %u rows\n", conclusion->rows_read);
 548 #endif
 549   if (opt_test == SET_TEST)
 550     printf("\tTook %ld.%03ld seconds to load data\n", conclusion->load_time / 1000,
 551            conclusion->load_time % 1000);
 552   else
 553     printf("\tTook %ld.%03ld seconds to read data\n", conclusion->read_time / 1000,
 554            conclusion->read_time % 1000);
 555 }
 556
 557 void flush_all(memcached_st *memc)
 558 {
 559   memcached_flush(memc, 0);
 560 }
 561
 562 pairs_st *load_create_data(memcached_st *memc, unsigned int number_of,
 563                            unsigned int *actual_loaded)
 564 {
 565   memcached_st *memc_clone= memcached_clone(NULL, memc);
 566   /* We always used non-blocking IO for load since it is faster */
 567   memcached_behavior_set(memc_clone, MEMCACHED_BEHAVIOR_NO_BLOCK, 0);
 568
 569   pairs_st *pairs= pairs_generate(number_of, 400);
 570   *actual_loaded= execute_set(memc_clone, pairs, number_of);
 571
 572   memcached_free(memc_clone);
 573
 574   return pairs;
 575 }