git.m6w6.name Git - m6w6/libmemcached/blob - tests/libmemcached-1.0/pool.cc

   1 /*  vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
   2  *
   3  *  Libmemcached Client and Server
   4  *
   5  *  Copyright (C) 2011 Data Differential, http://datadifferential.com/
   6  *  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 #include <config.h>
  39 #include <libtest/test.hpp>
  40
  41 using namespace libtest;
  42
  43 #include <vector>
  44 #include <iostream>
  45 #include <string>
  46 #include <cerrno>
  47
  48 #include <semaphore.h>
  49
  50 #include <libmemcached-1.0/memcached.h>
  51 #include <libmemcachedutil-1.0/util.h>
  52 #include <libmemcached/is.h>
  53 #include <tests/pool.h>
  54
  55 #include <pthread.h>
  56 #include <poll.h>
  57
  58 #include "libmemcached/instance.h"
  59
  60 #ifndef __INTEL_COMPILER
  61 #pragma GCC diagnostic ignored "-Wstrict-aliasing"
  62 #endif
  63
  64
  65 test_return_t memcached_pool_test(memcached_st *)
  66 {
  67   const char *config_string= "--SERVER=host10.example.com --SERVER=host11.example.com --SERVER=host10.example.com --POOL-MIN=10 --POOL-MAX=32";
  68
  69   char buffer[2048];
  70
  71   test_compare(libmemcached_check_configuration(config_string, sizeof(config_string) -1, buffer, sizeof(buffer)), MEMCACHED_PARSE_ERROR);
  72
  73   memcached_pool_st* pool= memcached_pool(config_string, strlen(config_string));
  74   test_true(pool);
  75
  76   memcached_return_t rc;
  77   memcached_st *memc= memcached_pool_pop(pool, false, &rc);
  78
  79   test_compare(rc, MEMCACHED_SUCCESS);
  80   test_true(memc);
  81
  82   /*
  83     Release the memc_ptr that was pulled from the pool
  84   */
  85   memcached_pool_push(pool, memc);
  86
  87   /*
  88     Destroy the pool.
  89   */
  90   memcached_pool_destroy(pool);
  91
  92   return TEST_SUCCESS;
  93 }
  94
  95
  96 #define POOL_SIZE 10
  97 test_return_t connection_pool_test(memcached_st *memc)
  98 {
  99   memcached_pool_st* pool= memcached_pool_create(memc, 5, POOL_SIZE);
 100   test_true(pool);
 101   memcached_st *mmc[POOL_SIZE];
 102
 103   // Fill up our array that we will store the memc that are in the pool
 104   for (size_t x= 0; x < POOL_SIZE; ++x)
 105   {
 106     memcached_return_t rc;
 107     mmc[x]= memcached_pool_fetch(pool, NULL, &rc);
 108     test_compare(MEMCACHED_SUCCESS, rc);
 109     test_true(mmc[x]);
 110   }
 111
 112   // All memc should be gone
 113   {
 114     memcached_return_t rc;
 115     test_null(memcached_pool_fetch(pool, NULL, &rc));
 116     test_compare(MEMCACHED_NOTFOUND, rc);
 117   }
 118
 119   // Release them..
 120   for (size_t x= 0; x < POOL_SIZE; ++x)
 121   {
 122     if (mmc[x])
 123     {
 124       test_compare(MEMCACHED_SUCCESS, memcached_pool_release(pool, mmc[x]));
 125     }
 126   }
 127   test_true(memcached_pool_destroy(pool) == memc);
 128
 129   return TEST_SUCCESS;
 130 }
 131
 132 test_return_t connection_pool2_test(memcached_st *memc)
 133 {
 134   memcached_pool_st* pool= memcached_pool_create(memc, 5, POOL_SIZE);
 135   test_true(pool);
 136   memcached_st *mmc[POOL_SIZE];
 137
 138   // Fill up our array that we will store the memc that are in the pool
 139   for (size_t x= 0; x < POOL_SIZE; ++x)
 140   {
 141     memcached_return_t rc;
 142     mmc[x]= memcached_pool_fetch(pool, NULL, &rc);
 143     test_compare(MEMCACHED_SUCCESS, rc);
 144     test_true(mmc[x]);
 145   }
 146
 147   // All memc should be gone
 148   {
 149     memcached_return_t rc;
 150     test_null(memcached_pool_fetch(pool, NULL, &rc));
 151     test_compare(MEMCACHED_NOTFOUND, rc);
 152   }
 153
 154   // verify that I can do ops with all connections
 155   test_compare(MEMCACHED_SUCCESS,
 156                memcached_set(mmc[0],
 157                              test_literal_param("key"),
 158                              "0", 1, 0, 0));
 159
 160   for (uint64_t x= 0; x < POOL_SIZE; ++x)
 161   {
 162     uint64_t number_value;
 163     test_compare(MEMCACHED_SUCCESS,
 164                  memcached_increment(mmc[x],
 165                                      test_literal_param("key"),
 166                                      1, &number_value));
 167     test_compare(number_value, (x+1));
 168   }
 169
 170   // Release them..
 171   for (size_t x= 0; x < POOL_SIZE; ++x)
 172   {
 173     test_compare(MEMCACHED_SUCCESS, memcached_pool_release(pool, mmc[x]));
 174   }
 175
 176
 177   /* verify that I can set behaviors on the pool when I don't have all
 178    * of the connections in the pool. It should however be enabled
 179    * when I push the item into the pool
 180  */
 181   mmc[0]= memcached_pool_fetch(pool, NULL, NULL);
 182   test_true(mmc[0]);
 183
 184   test_compare(MEMCACHED_SUCCESS,
 185                memcached_pool_behavior_set(pool, MEMCACHED_BEHAVIOR_IO_MSG_WATERMARK, 9999));
 186
 187   {
 188     memcached_return_t rc;
 189     mmc[1]= memcached_pool_fetch(pool, NULL, &rc);
 190     test_true(mmc[1]);
 191     test_compare(MEMCACHED_SUCCESS, rc);
 192   }
 193
 194   test_compare(UINT64_C(9999), memcached_behavior_get(mmc[1], MEMCACHED_BEHAVIOR_IO_MSG_WATERMARK));
 195   test_compare(MEMCACHED_SUCCESS, memcached_pool_release(pool, mmc[1]));
 196   test_compare(MEMCACHED_SUCCESS, memcached_pool_release(pool, mmc[0]));
 197
 198   {
 199     memcached_return_t rc;
 200     mmc[0]= memcached_pool_fetch(pool, NULL, &rc);
 201     test_true(mmc[0]);
 202     test_compare(MEMCACHED_SUCCESS, rc);
 203   }
 204
 205   test_compare(UINT64_C(9999), memcached_behavior_get(mmc[0], MEMCACHED_BEHAVIOR_IO_MSG_WATERMARK));
 206   test_compare(MEMCACHED_SUCCESS, memcached_pool_release(pool, mmc[0]));
 207
 208   test_true(memcached_pool_destroy(pool) == memc);
 209
 210   return TEST_SUCCESS;
 211 }
 212
 213 struct test_pool_context_st {
 214   volatile memcached_return_t rc;
 215   memcached_pool_st* pool;
 216   memcached_st* mmc;
 217   sem_t _lock;
 218
 219   test_pool_context_st(memcached_pool_st *pool_arg, memcached_st *memc_arg):
 220     rc(MEMCACHED_FAILURE),
 221     pool(pool_arg),
 222     mmc(memc_arg)
 223   {
 224     sem_init(&_lock, 0, 0);
 225   }
 226
 227   void wait()
 228   {
 229     sem_wait(&_lock);
 230   }
 231
 232   void release()
 233   {
 234     sem_post(&_lock);
 235   }
 236
 237   ~test_pool_context_st()
 238   {
 239     sem_destroy(&_lock);
 240   }
 241 };
 242
 243 static __attribute__((noreturn)) void* connection_release(void *arg)
 244 {
 245   test_pool_context_st *resource= static_cast<test_pool_context_st *>(arg);
 246   if (resource == NULL)
 247   {
 248     fatal_message("resource == NULL");
 249   }
 250
 251   // Release all of the memc we are holding
 252   resource->rc= memcached_pool_release(resource->pool, resource->mmc);
 253   resource->release();
 254
 255   pthread_exit(arg);
 256 }
 257
 258 test_return_t connection_pool3_test(memcached_st *memc)
 259 {
 260 #ifdef __APPLE__
 261   return TEST_SKIPPED;
 262 #endif
 263
 264   memcached_pool_st* pool= memcached_pool_create(memc, 1, 1);
 265   test_true(pool);
 266
 267   memcached_st *pool_memc;
 268   {
 269     memcached_return_t rc;
 270     pool_memc= memcached_pool_fetch(pool, NULL, &rc);
 271     test_compare(MEMCACHED_SUCCESS, rc);
 272     test_true(pool_memc);
 273   }
 274
 275   /*
 276     @note This comment was written to describe what was believed to be the original authors intent.
 277
 278     This portion of the test creates a thread that will wait until told to free a memcached_st
 279     that will be grabbed by the main thread.
 280
 281     It is believed that this tests whether or not we are handling ownership correctly.
 282   */
 283   pthread_t tid;
 284   test_pool_context_st item(pool, pool_memc);
 285
 286   test_zero(pthread_create(&tid, NULL, connection_release, &item));
 287   item.wait();
 288
 289   memcached_return_t rc;
 290   memcached_st *pop_memc;
 291   // We do a hard loop, and try N times
 292   int counter= 5;
 293   do
 294   {
 295     struct timespec relative_time= { 0, 0 };
 296     pop_memc= memcached_pool_fetch(pool, &relative_time, &rc);
 297
 298     if (memcached_success(rc))
 299     {
 300       break;
 301     }
 302
 303     if (memcached_failed(rc))
 304     {
 305       test_null(pop_memc);
 306       test_true(rc != MEMCACHED_TIMEOUT); // As long as relative_time is zero, MEMCACHED_TIMEOUT is invalid
 307     }
 308   } while (--counter);
 309
 310   if (memcached_failed(rc)) // Cleanup thread since we will exit once we test.
 311   {
 312     pthread_join(tid, NULL);
 313     test_compare(MEMCACHED_SUCCESS, rc);
 314   }
 315
 316   {
 317     int pthread_ret= pthread_join(tid, NULL);
 318     test_true(pthread_ret == 0 or pthread_ret == ESRCH);
 319   }
 320   test_compare(MEMCACHED_SUCCESS, rc);
 321   test_true(pool_memc == pop_memc);
 322
 323   test_true(memcached_pool_destroy(pool) == memc);
 324
 325   return TEST_SUCCESS;
 326 }
 327
 328 static memcached_st * create_single_instance_memcached(const memcached_st *original_memc, const char *options)
 329 {
 330   /*
 331     If no options are given, copy over at least the binary flag.
 332   */
 333   char options_buffer[1024]= { 0 };
 334   if (options == NULL)
 335   {
 336     if (memcached_is_binary(original_memc))
 337     {
 338       snprintf(options_buffer, sizeof(options_buffer), "--BINARY");
 339     }
 340   }
 341
 342   /*
 343    * I only want to hit _one_ server so I know the number of requests I'm
 344    * sending in the pipeline.
 345    */
 346   memcached_server_instance_st instance= memcached_server_instance_by_position(original_memc, 0);
 347
 348   char server_string[1024];
 349   int server_string_length;
 350   if (instance->type == MEMCACHED_CONNECTION_UNIX_SOCKET)
 351   {
 352     if (options)
 353     {
 354       server_string_length= snprintf(server_string, sizeof(server_string), "--SOCKET=\"%s\" %s",
 355                                      memcached_server_name(instance), options);
 356     }
 357     else
 358     {
 359       server_string_length= snprintf(server_string, sizeof(server_string), "--SOCKET=\"%s\"",
 360                                      memcached_server_name(instance));
 361     }
 362   }
 363   else
 364   {
 365     if (options)
 366     {
 367       server_string_length= snprintf(server_string, sizeof(server_string), "--server=%s:%d %s",
 368                                      memcached_server_name(instance), int(memcached_server_port(instance)),
 369                                      options);
 370     }
 371     else
 372     {
 373       server_string_length= snprintf(server_string, sizeof(server_string), "--server=%s:%d",
 374                                      memcached_server_name(instance), int(memcached_server_port(instance)));
 375     }
 376   }
 377
 378   if (server_string_length <= 0)
 379   {
 380     return NULL;
 381   }
 382
 383   char errror_buffer[1024];
 384   if (memcached_failed(libmemcached_check_configuration(server_string, server_string_length, errror_buffer, sizeof(errror_buffer))))
 385   {
 386     Error << "Failed to parse (" << server_string << ") " << errror_buffer;
 387     return NULL;
 388   }
 389
 390   return memcached(server_string, server_string_length);
 391 }
 392
 393 pthread_mutex_t mutex= PTHREAD_MUTEX_INITIALIZER;
 394 static bool _running= false;
 395
 396 static void set_running(const bool arg)
 397 {
 398   int error;
 399   if ((error= pthread_mutex_lock(&mutex)) != 0)
 400   {
 401     fatal_message(strerror(error));
 402   }
 403
 404   _running= arg;
 405
 406   if ((error= pthread_mutex_unlock(&mutex)) != 0)
 407   {
 408     fatal_message(strerror(error));
 409   }
 410 }
 411
 412 static bool running()
 413 {
 414   int error;
 415   bool ret;
 416
 417   if ((error= pthread_mutex_lock(&mutex)) != 0)
 418   {
 419     fatal_message(strerror(error));
 420   }
 421
 422   ret= _running;
 423
 424   if ((error= pthread_mutex_unlock(&mutex)) != 0)
 425   {
 426     fatal_message(strerror(error));
 427   }
 428
 429   return ret;
 430 }
 431
 432 static void *worker_thread(void *ctx)
 433 {
 434   memcached_pool_st *pool= (memcached_pool_st *)ctx;
 435
 436   while (running())
 437   {
 438     memcached_return_t rc;
 439     memcached_st *mc= memcached_pool_pop(pool, true, &rc);
 440
 441     if (mc == NULL)
 442     {
 443       Error << "failed to fetch a connection from the pool" << memcached_strerror(NULL, rc);
 444       dream(1, 0);
 445       continue;
 446     }
 447
 448     rc= memcached_set(mc, "test:kv", 7, "value", 5, 600, 0);
 449     if (memcached_failed(rc))
 450     {
 451       Out << "failed memcached_set()";
 452     }
 453
 454     rc= memcached_pool_push(pool, mc);
 455     if (memcached_failed(rc))
 456     {
 457       Error << "failed to release a connection to the pool" << memcached_strerror(NULL, rc);
 458     }
 459   }
 460
 461   return NULL;
 462 }
 463
 464 #define NUM_THREADS 20
 465 test_return_t regression_bug_962815(memcached_st *memc)
 466 {
 467   pthread_t pid[NUM_THREADS];
 468
 469   test_false(running());
 470
 471   memcached_st *master = create_single_instance_memcached(memc, 0);
 472   test_true(master);
 473
 474   memcached_pool_st *pool= memcached_pool_create(master, 5, 10);
 475
 476   test_true(pool);
 477
 478   set_running(true);
 479
 480   for (size_t x=0; x < NUM_THREADS; x++)
 481   {
 482     test_compare(0, pthread_create(&pid[x], NULL, worker_thread, (void*)pool));
 483   }
 484
 485   {
 486     pollfd fds[1];
 487     memset(fds, 0, sizeof(pollfd));
 488     fds[0].fd= -1; //STDIN_FILENO;
 489     fds[0].events= POLLIN;
 490     fds[0].revents= 0;
 491
 492     int active_fd;
 493     if ((active_fd= poll(fds, 1, 5000)) == -1)
 494     {
 495       Error << "poll() failed with:" << strerror(errno);
 496     }
 497     test_zero(active_fd);
 498
 499     set_running(false);
 500   }
 501
 502   for (size_t x=0; x < NUM_THREADS; x++)
 503   {
 504     test_compare(0, pthread_join(pid[x], NULL));
 505   }
 506
 507   if (pool)
 508   {
 509     memcached_pool_destroy(pool);
 510   }
 511
 512   if (master)
 513   {
 514     memcached_free(master);
 515   }
 516
 517   return TEST_SUCCESS;
 518 }