git.m6w6.name Git - m6w6/libmemcached/blob - libmemcachedutil/pool.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) 2010 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 <libmemcachedutil/common.h>
  40
  41 #include <cassert>
  42 #include <cerrno>
  43 #include <pthread.h>
  44 #include <memory>
  45
  46 struct memcached_pool_st
  47 {
  48   pthread_mutex_t mutex;
  49   pthread_cond_t cond;
  50   memcached_st *master;
  51   memcached_st **server_pool;
  52   int firstfree;
  53   const uint32_t size;
  54   uint32_t current_size;
  55   bool _owns_master;
  56   struct timespec _timeout;
  57
  58   memcached_pool_st(memcached_st *master_arg, size_t max_arg) :
  59     master(master_arg),
  60     server_pool(NULL),
  61     firstfree(-1),
  62     size(uint32_t(max_arg)),
  63     current_size(0),
  64     _owns_master(false)
  65   {
  66     pthread_mutex_init(&mutex, NULL);
  67     pthread_cond_init(&cond, NULL);
  68     _timeout.tv_sec= 5;
  69     _timeout.tv_nsec= 0;
  70   }
  71
  72   const struct timespec& timeout() const
  73   {
  74     return _timeout;
  75   }
  76
  77   bool release(memcached_st*, memcached_return_t& rc);
  78
  79   memcached_st *fetch(memcached_return_t& rc);
  80   memcached_st *fetch(const struct timespec&, memcached_return_t& rc);
  81
  82   bool init(uint32_t initial);
  83
  84   ~memcached_pool_st()
  85   {
  86     for (int x= 0; x <= firstfree; ++x)
  87     {
  88       memcached_free(server_pool[x]);
  89       server_pool[x] = NULL;
  90     }
  91
  92     pthread_mutex_destroy(&mutex);
  93     pthread_cond_destroy(&cond);
  94     delete [] server_pool;
  95     if (_owns_master)
  96     {
  97       memcached_free(master);
  98     }
  99   }
 100
 101   void increment_version()
 102   {
 103     ++master->configure.version;
 104   }
 105
 106   bool compare_version(const memcached_st *arg) const
 107   {
 108     return (arg->configure.version == version());
 109   }
 110
 111   int32_t version() const
 112   {
 113     return master->configure.version;
 114   }
 115 };
 116
 117
 118 /**
 119  * Grow the connection pool by creating a connection structure and clone the
 120  * original memcached handle.
 121  */
 122 static bool grow_pool(memcached_pool_st* pool)
 123 {
 124   assert(pool);
 125
 126   memcached_st *obj;
 127   if (not (obj= memcached_clone(NULL, pool->master)))
 128   {
 129     return false;
 130   }
 131
 132   pool->server_pool[++pool->firstfree]= obj;
 133   pool->current_size++;
 134   obj->configure.version= pool->version();
 135
 136   return true;
 137 }
 138
 139 bool memcached_pool_st::init(uint32_t initial)
 140 {
 141   server_pool= new (std::nothrow) memcached_st *[size];
 142   if (not server_pool)
 143     return false;
 144
 145   /*
 146     Try to create the initial size of the pool. An allocation failure at
 147     this time is not fatal..
 148   */
 149   for (unsigned int x= 0; x < initial; ++x)
 150   {
 151     if (grow_pool(this) == false)
 152     {
 153       break;
 154     }
 155   }
 156
 157   return true;
 158 }
 159
 160
 161 static inline memcached_pool_st *_pool_create(memcached_st* master, uint32_t initial, uint32_t max)
 162 {
 163   if (initial == 0 or max == 0 or (initial > max))
 164   {
 165     return NULL;
 166   }
 167
 168   memcached_pool_st *object= new (std::nothrow) memcached_pool_st(master, max);
 169   if (object == NULL)
 170   {
 171     return NULL;
 172   }
 173
 174   /*
 175     Try to create the initial size of the pool. An allocation failure at
 176     this time is not fatal..
 177   */
 178   if (not object->init(initial))
 179   {
 180     delete object;
 181     return NULL;
 182   }
 183
 184   return object;
 185 }
 186
 187 memcached_pool_st *memcached_pool_create(memcached_st* master, uint32_t initial, uint32_t max)
 188 {
 189   return _pool_create(master, initial, max);
 190 }
 191
 192 memcached_pool_st * memcached_pool(const char *option_string, size_t option_string_length)
 193 {
 194   memcached_st *memc= memcached(option_string, option_string_length);
 195
 196   if (memc == NULL)
 197   {
 198     return NULL;
 199   }
 200
 201   memcached_pool_st *self= memcached_pool_create(memc, memc->configure.initial_pool_size, memc->configure.max_pool_size);
 202   if (self == NULL)
 203   {
 204     memcached_free(memc);
 205     return NULL;
 206   }
 207
 208   self->_owns_master= true;
 209
 210   return self;
 211 }
 212
 213 memcached_st*  memcached_pool_destroy(memcached_pool_st* pool)
 214 {
 215   if (pool == NULL)
 216   {
 217     return NULL;
 218   }
 219
 220   // Legacy that we return the original structure
 221   memcached_st *ret= NULL;
 222   if (pool->_owns_master)
 223   { }
 224   else
 225   {
 226     ret= pool->master;
 227   }
 228
 229   delete pool;
 230
 231   return ret;
 232 }
 233
 234 memcached_st* memcached_pool_st::fetch(memcached_return_t& rc)
 235 {
 236   static struct timespec relative_time= { 0, 0 };
 237   return fetch(relative_time, rc);
 238 }
 239
 240 memcached_st* memcached_pool_st::fetch(const struct timespec& relative_time, memcached_return_t& rc)
 241 {
 242   rc= MEMCACHED_SUCCESS;
 243
 244   int error;
 245   if ((error= pthread_mutex_lock(&mutex)) != 0)
 246   {
 247     rc= MEMCACHED_IN_PROGRESS;
 248     return NULL;
 249   }
 250
 251   memcached_st *ret= NULL;
 252   do
 253   {
 254     if (firstfree > -1)
 255     {
 256       ret= server_pool[firstfree--];
 257     }
 258     else if (current_size == size)
 259     {
 260       if (relative_time.tv_sec == 0 and relative_time.tv_nsec == 0)
 261       {
 262         error= pthread_mutex_unlock(&mutex);
 263         rc= MEMCACHED_NOTFOUND;
 264
 265         return NULL;
 266       }
 267
 268       struct timespec time_to_wait= {0, 0};
 269       time_to_wait.tv_sec= time(NULL) +relative_time.tv_sec;
 270       time_to_wait.tv_nsec= relative_time.tv_nsec;
 271
 272       int thread_ret;
 273       if ((thread_ret= pthread_cond_timedwait(&cond, &mutex, &time_to_wait)) != 0)
 274       {
 275         int unlock_error;
 276         if ((unlock_error= pthread_mutex_unlock(&mutex)) != 0)
 277         {
 278         }
 279
 280         if (thread_ret == ETIMEDOUT)
 281         {
 282           rc= MEMCACHED_TIMEOUT;
 283         }
 284         else
 285         {
 286           errno= thread_ret;
 287           rc= MEMCACHED_ERRNO;
 288         }
 289
 290         return NULL;
 291       }
 292     }
 293     else if (grow_pool(this) == false)
 294     {
 295       int unlock_error;
 296       if ((unlock_error= pthread_mutex_unlock(&mutex)) != 0)
 297       {
 298       }
 299
 300       return NULL;
 301     }
 302   } while (ret == NULL);
 303
 304   if ((error= pthread_mutex_unlock(&mutex)) != 0)
 305   {
 306   }
 307
 308   return ret;
 309 }
 310
 311 bool memcached_pool_st::release(memcached_st *released, memcached_return_t& rc)
 312 {
 313   rc= MEMCACHED_SUCCESS;
 314   if (released == NULL)
 315   {
 316     rc= MEMCACHED_INVALID_ARGUMENTS;
 317     return false;
 318   }
 319
 320   int error;
 321   if ((error= pthread_mutex_lock(&mutex)))
 322   {
 323     rc= MEMCACHED_IN_PROGRESS;
 324     return false;
 325   }
 326
 327   /*
 328     Someone updated the behavior on the object, so we clone a new memcached_st with the new settings. If we fail to clone, we keep the old one around.
 329   */
 330   if (compare_version(released) == false)
 331   {
 332     memcached_st *memc;
 333     if ((memc= memcached_clone(NULL, master)))
 334     {
 335       memcached_free(released);
 336       released= memc;
 337     }
 338   }
 339
 340   server_pool[++firstfree]= released;
 341
 342   if (firstfree == 0 and current_size == size)
 343   {
 344     /* we might have people waiting for a connection.. wake them up :-) */
 345     if ((error= pthread_cond_broadcast(&cond)) != 0)
 346     {
 347     }
 348   }
 349
 350   if ((error= pthread_mutex_unlock(&mutex)) != 0)
 351   {
 352   }
 353
 354   return true;
 355 }
 356
 357 memcached_st* memcached_pool_fetch(memcached_pool_st* pool, struct timespec* relative_time, memcached_return_t* rc)
 358 {
 359   if (pool == NULL)
 360   {
 361     return NULL;
 362   }
 363
 364   memcached_return_t unused;
 365   if (rc == NULL)
 366   {
 367     rc= &unused;
 368   }
 369
 370   if (relative_time == NULL)
 371   {
 372     return pool->fetch(*rc);
 373   }
 374
 375   return pool->fetch(*relative_time, *rc);
 376 }
 377
 378 memcached_st* memcached_pool_pop(memcached_pool_st* pool,
 379                                  bool block,
 380                                  memcached_return_t *rc)
 381 {
 382   if (pool == NULL)
 383   {
 384     return NULL;
 385   }
 386
 387   memcached_return_t unused;
 388   if (rc == NULL)
 389   {
 390     rc= &unused;
 391   }
 392
 393   memcached_st *memc;
 394   if (block)
 395   {
 396     memc= pool->fetch(pool->timeout(), *rc);
 397   }
 398   else
 399   {
 400     memc= pool->fetch(*rc);
 401   }
 402
 403   return memc;
 404 }
 405
 406 memcached_return_t memcached_pool_release(memcached_pool_st* pool, memcached_st *released)
 407 {
 408   if (pool == NULL)
 409   {
 410     return MEMCACHED_INVALID_ARGUMENTS;
 411   }
 412
 413   memcached_return_t rc;
 414
 415   (void) pool->release(released, rc);
 416
 417   return rc;
 418 }
 419
 420 memcached_return_t memcached_pool_push(memcached_pool_st* pool, memcached_st *released)
 421 {
 422   return memcached_pool_release(pool, released);
 423 }
 424
 425
 426 memcached_return_t memcached_pool_behavior_set(memcached_pool_st *pool,
 427                                                memcached_behavior_t flag,
 428                                                uint64_t data)
 429 {
 430   if (pool == NULL)
 431   {
 432     return MEMCACHED_INVALID_ARGUMENTS;
 433   }
 434
 435   int error;
 436   if ((error= pthread_mutex_lock(&pool->mutex)))
 437   {
 438     return MEMCACHED_IN_PROGRESS;
 439   }
 440
 441   /* update the master */
 442   memcached_return_t rc= memcached_behavior_set(pool->master, flag, data);
 443   if (memcached_failed(rc))
 444   {
 445     if ((error= pthread_mutex_unlock(&pool->mutex)) != 0)
 446     {
 447     }
 448     return rc;
 449   }
 450
 451   pool->increment_version();
 452   /* update the clones */
 453   for (int xx= 0; xx <= pool->firstfree; ++xx)
 454   {
 455     if (memcached_success(memcached_behavior_set(pool->server_pool[xx], flag, data)))
 456     {
 457       pool->server_pool[xx]->configure.version= pool->version();
 458     }
 459     else
 460     {
 461       memcached_st *memc;
 462       if ((memc= memcached_clone(NULL, pool->master)))
 463       {
 464         memcached_free(pool->server_pool[xx]);
 465         pool->server_pool[xx]= memc;
 466         /* I'm not sure what to do in this case.. this would happen
 467           if we fail to push the server list inside the client..
 468           I should add a testcase for this, but I believe the following
 469           would work, except that you would add a hole in the pool list..
 470           in theory you could end up with an empty pool....
 471         */
 472       }
 473     }
 474   }
 475
 476   if ((error= pthread_mutex_unlock(&pool->mutex)) != 0)
 477   {
 478   }
 479
 480   return rc;
 481 }
 482
 483 memcached_return_t memcached_pool_behavior_get(memcached_pool_st *pool,
 484                                                memcached_behavior_t flag,
 485                                                uint64_t *value)
 486 {
 487   if (pool == NULL)
 488   {
 489     return MEMCACHED_INVALID_ARGUMENTS;
 490   }
 491
 492   int error;
 493   if ((error= pthread_mutex_lock(&pool->mutex)))
 494   {
 495     return MEMCACHED_IN_PROGRESS;
 496   }
 497
 498   *value= memcached_behavior_get(pool->master, flag);
 499
 500   if ((error= pthread_mutex_unlock(&pool->mutex)) != 0)
 501   {
 502   }
 503
 504   return MEMCACHED_SUCCESS;
 505 }