git.m6w6.name Git - awesomized/libmemcached/blob - src/memslap.c

   1 #include <stdio.h>
   2 #include <stdlib.h>
   3 #include <sys/types.h>
   4 #include <sys/stat.h>
   5 #include <sys/types.h>
   6 #include <sys/mman.h>
   7 #include <fcntl.h>
   8 #include <sys/time.h>
   9 #include <getopt.h>
  10 #include <pthread.h>
  11
  12 #include <memcached.h>
  13
  14 #include "client_options.h"
  15 #include "utilities.h"
  16 #include "generator.h"
  17
  18 /* Global Thread counter */
  19 unsigned int thread_counter;
  20 pthread_mutex_t counter_mutex;
  21 pthread_cond_t count_threshhold;
  22 unsigned int master_wakeup;
  23 pthread_mutex_t sleeper_mutex;
  24 pthread_cond_t sleep_threshhold;
  25
  26 void *run_task(void *p);
  27
  28 /* Types */
  29 typedef struct conclusions_st conclusions_st;
  30 typedef struct thread_context_st thread_context_st;
  31 typedef enum {
  32   AC_SET,
  33   AC_GET,
  34 } run_action;
  35
  36 struct thread_context_st {
  37   pairs_st *pairs;
  38   run_action action;
  39   memcached_server_st *servers;
  40 };
  41
  42 struct conclusions_st {
  43   long int load_time;
  44   long int read_time;
  45   unsigned int rows_loaded;
  46   unsigned int rows_read;
  47 };
  48
  49 /* Prototypes */
  50 void options_parse(int argc, char *argv[]);
  51 void conclusions_print(conclusions_st *conclusion);
  52 void scheduler(memcached_server_st *servers, conclusions_st *conclusion);
  53
  54 static int opt_verbose= 0;
  55 static unsigned int opt_default_pairs= 100;
  56 static unsigned int opt_concurrency= 10;
  57 static int opt_displayflag= 0;
  58 static char *opt_servers= NULL;
  59
  60 int main(int argc, char *argv[])
  61 {
  62   conclusions_st conclusion;
  63   memcached_server_st *servers;
  64
  65   memset(&conclusion, 0, sizeof(conclusions_st));
  66
  67   srandom(time(NULL));
  68   options_parse(argc, argv);
  69
  70   if (!opt_servers)
  71     exit(0);
  72
  73   servers= parse_opt_servers(opt_servers);
  74
  75   pthread_mutex_init(&counter_mutex, NULL);
  76   pthread_cond_init(&count_threshhold, NULL);
  77   pthread_mutex_init(&sleeper_mutex, NULL);
  78   pthread_cond_init(&sleep_threshhold, NULL);
  79
  80   scheduler(servers, &conclusion);
  81
  82   free(opt_servers);
  83
  84   (void)pthread_mutex_init(&counter_mutex, NULL);
  85   (void)pthread_cond_init(&count_threshhold, NULL);
  86   (void)pthread_mutex_init(&sleeper_mutex, NULL);
  87   (void)pthread_cond_init(&sleep_threshhold, NULL);
  88   conclusions_print(&conclusion);
  89   memcached_server_list_free(servers);
  90
  91   return 0;
  92 }
  93
  94 void scheduler(memcached_server_st *servers, conclusions_st *conclusion)
  95 {
  96   unsigned int x;
  97   struct timeval start_time, end_time;
  98   pthread_t mainthread;            /* Thread descriptor */
  99   pthread_attr_t attr;          /* Thread attributes */
 100   pairs_st *pairs;
 101
 102   pthread_attr_init(&attr);
 103   pthread_attr_setdetachstate(&attr,
 104                               PTHREAD_CREATE_DETACHED);
 105
 106   pairs= pairs_generate(opt_default_pairs);
 107
 108   pthread_mutex_lock(&counter_mutex);
 109   thread_counter= 0;
 110
 111   pthread_mutex_lock(&sleeper_mutex);
 112   master_wakeup= 1;
 113   pthread_mutex_unlock(&sleeper_mutex);
 114
 115   for (x= 0; x < opt_concurrency; x++)
 116   {
 117     thread_context_st *context;
 118     context= (thread_context_st *)malloc(sizeof(thread_context_st));
 119
 120     context->servers= servers;
 121     context->pairs= pairs;
 122     context->action= AC_SET;
 123
 124     /* now you create the thread */
 125     if (pthread_create(&mainthread, &attr, run_task,
 126                        (void *)context) != 0)
 127     {
 128       fprintf(stderr,"Could not create thread\n");
 129       exit(1);
 130     }
 131     thread_counter++;
 132   }
 133
 134   pthread_mutex_unlock(&counter_mutex);
 135   pthread_attr_destroy(&attr);
 136
 137   pthread_mutex_lock(&sleeper_mutex);
 138   master_wakeup= 0;
 139   pthread_mutex_unlock(&sleeper_mutex);
 140   pthread_cond_broadcast(&sleep_threshhold);
 141
 142   gettimeofday(&start_time, NULL);
 143   /*
 144     We loop until we know that all children have cleaned up.
 145   */
 146   pthread_mutex_lock(&counter_mutex);
 147   while (thread_counter)
 148   {
 149     struct timespec abstime;
 150
 151     memset(&abstime, 0, sizeof(struct timespec));
 152     abstime.tv_sec= 1;
 153
 154     pthread_cond_timedwait(&count_threshhold, &counter_mutex, &abstime);
 155   }
 156   pthread_mutex_unlock(&counter_mutex);
 157
 158   gettimeofday(&end_time, NULL);
 159
 160   conclusion->load_time= timedif(end_time, start_time);
 161   conclusion->read_time= timedif(end_time, start_time);
 162   pairs_free(pairs);
 163 }
 164
 165 void options_parse(int argc, char *argv[])
 166 {
 167   static struct option long_options[]=
 168     {
 169       {"version", no_argument, NULL, OPT_VERSION},
 170       {"help", no_argument, NULL, OPT_HELP},
 171       {"verbose", no_argument, &opt_verbose, OPT_VERBOSE},
 172       {"debug", no_argument, &opt_verbose, OPT_DEBUG},
 173       {"servers", required_argument, NULL, OPT_SERVERS},
 174       {"flag", no_argument, &opt_displayflag, OPT_FLAG},
 175       {"default-pairs", required_argument, NULL, OPT_SLAP_DEFAULT_PAIRS},
 176       {0, 0, 0, 0},
 177     };
 178
 179   int option_index= 0;
 180   int option_rv;
 181
 182   while (1)
 183   {
 184     option_rv= getopt_long(argc, argv, "Vhvds:", long_options, &option_index);
 185     if (option_rv == -1) break;
 186     switch (option_rv)
 187     {
 188     case 0:
 189       break;
 190     case OPT_VERBOSE: /* --verbose or -v */
 191       opt_verbose = OPT_VERBOSE;
 192       break;
 193     case OPT_DEBUG: /* --debug or -d */
 194       opt_verbose = OPT_DEBUG;
 195       break;
 196     case OPT_VERSION: /* --version or -V */
 197       printf("memcache tools, memcat, v1.0\n");
 198       exit(0);
 199       break;
 200     case OPT_HELP: /* --help or -h */
 201       printf("useful help messages go here\n");
 202       exit(0);
 203       break;
 204     case OPT_SERVERS: /* --servers or -s */
 205       opt_servers= strdup(optarg);
 206       break;
 207     case OPT_SLAP_DEFAULT_PAIRS:
 208       opt_default_pairs= strtol(optarg, (char **)NULL, 10);
 209       break;
 210     case '?':
 211       /* getopt_long already printed an error message. */
 212       exit(1);
 213     default:
 214       abort();
 215     }
 216   }
 217 }
 218
 219 void conclusions_print(conclusions_st *conclusion)
 220 {
 221   printf("\tLoaded %u rows\n", conclusion->rows_loaded);
 222   printf("\tRead %u rows\n", conclusion->rows_read);
 223   printf("\tTook %ld.%03ld seconds to load data\n", conclusion->load_time / 1000,
 224          conclusion->load_time % 1000);
 225   printf("\tTook %ld.%03ld seconds to read data\n", conclusion->read_time / 1000,
 226          conclusion->read_time % 1000);
 227 }
 228
 229 void *run_task(void *p)
 230 {
 231   unsigned int x;
 232   thread_context_st *context= (thread_context_st *)p;
 233   memcached_return rc;
 234   memcached_st *memc;
 235   pairs_st *pairs= context->pairs;
 236
 237   memc= memcached_init(NULL);
 238
 239   memcached_server_push(memc, context->servers);
 240
 241   pthread_mutex_lock(&sleeper_mutex);
 242   while (master_wakeup)
 243   {
 244     pthread_cond_wait(&sleep_threshhold, &sleeper_mutex);
 245   }
 246   pthread_mutex_unlock(&sleeper_mutex);
 247
 248   /* Do Stuff */
 249   switch (context->action)
 250   {
 251   case AC_SET:
 252     for (x= 0; x < opt_default_pairs; x++)
 253     {
 254       rc= memcached_set(memc, pairs[x].key, pairs[x].key_length,
 255                         pairs[x].value, pairs[x].value_length,
 256                         0, 0);
 257       if (rc != MEMCACHED_SUCCESS)
 258         fprintf(stderr, "Failured on insert of %.*s\n",
 259                 (unsigned int)pairs[x].key_length, pairs[x].key);
 260   }
 261     break;
 262   case AC_GET:
 263     for (x= 0; x < opt_default_pairs; x++)
 264     {
 265       char *value;
 266       size_t value_length;
 267       uint16_t flags;
 268
 269       value= memcached_get(memc, pairs[x].key, pairs[x].key_length,
 270                            &value_length,
 271                            &flags, &rc);
 272
 273       if (rc != MEMCACHED_SUCCESS)
 274         fprintf(stderr, "Failured on read of %.*s\n",
 275                 (unsigned int)pairs[x].key_length, pairs[x].key);
 276       free(value);
 277     }
 278     break;
 279   }
 280
 281   pthread_mutex_lock(&counter_mutex);
 282   thread_counter--;
 283   pthread_cond_signal(&count_threshhold);
 284   pthread_mutex_unlock(&counter_mutex);
 285   memcached_deinit(memc);
 286
 287   free(context);
 288
 289   return NULL;
 290 }