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