git.m6w6.name Git - awesomized/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(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   if (pthread_mutex_lock(&mutex))
 245   {
 246     rc= MEMCACHED_IN_PROGRESS;
 247     return NULL;
 248   }
 249
 250   memcached_st *ret= NULL;
 251   do
 252   {
 253     if (firstfree > -1)
 254     {
 255       ret= server_pool[firstfree--];
 256     }
 257     else if (current_size == size)
 258     {
 259       if (relative_time.tv_sec == 0 and relative_time.tv_nsec == 0)
 260       {
 261         pthread_mutex_unlock(&mutex);
 262         rc= MEMCACHED_NOTFOUND;
 263
 264         return NULL;
 265       }
 266
 267       struct timespec time_to_wait= {0, 0};
 268       time_to_wait.tv_sec= time(NULL) +relative_time.tv_sec;
 269       time_to_wait.tv_nsec= relative_time.tv_nsec;
 270
 271       int thread_ret;
 272       if ((thread_ret= pthread_cond_timedwait(&cond, &mutex, &time_to_wait)) != 0)
 273       {
 274         pthread_mutex_unlock(&mutex);
 275
 276         if (thread_ret == ETIMEDOUT)
 277         {
 278           rc= MEMCACHED_TIMEOUT;
 279         }
 280         else
 281         {
 282           errno= thread_ret;
 283           rc= MEMCACHED_ERRNO;
 284         }
 285
 286         return NULL;
 287       }
 288     }
 289     else if (grow_pool(this) == false)
 290     {
 291       (void)pthread_mutex_unlock(&mutex);
 292       return NULL;
 293     }
 294   } while (ret == NULL);
 295
 296   pthread_mutex_unlock(&mutex);
 297
 298   return ret;
 299 }
 300
 301 bool memcached_pool_st::release(memcached_st *released, memcached_return_t& rc)
 302 {
 303   rc= MEMCACHED_SUCCESS;
 304   if (released == NULL)
 305   {
 306     rc= MEMCACHED_INVALID_ARGUMENTS;
 307     return false;
 308   }
 309
 310   if (pthread_mutex_lock(&mutex))
 311   {
 312     rc= MEMCACHED_IN_PROGRESS;
 313     return false;
 314   }
 315
 316   /*
 317     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.
 318   */
 319   if (compare_version(released) == false)
 320   {
 321     memcached_st *memc;
 322     if ((memc= memcached_clone(NULL, master)))
 323     {
 324       memcached_free(released);
 325       released= memc;
 326     }
 327   }
 328
 329   server_pool[++firstfree]= released;
 330
 331   if (firstfree == 0 and current_size == size)
 332   {
 333     /* we might have people waiting for a connection.. wake them up :-) */
 334     pthread_cond_broadcast(&cond);
 335   }
 336
 337   (void)pthread_mutex_unlock(&mutex);
 338
 339   return true;
 340 }
 341
 342 memcached_st* memcached_pool_fetch(memcached_pool_st* pool, struct timespec* relative_time, memcached_return_t* rc)
 343 {
 344   if (pool == NULL)
 345   {
 346     return NULL;
 347   }
 348
 349   memcached_return_t unused;
 350   if (rc == NULL)
 351   {
 352     rc= &unused;
 353   }
 354
 355   if (relative_time == NULL)
 356   {
 357     return pool->fetch(*rc);
 358   }
 359
 360   return pool->fetch(*relative_time, *rc);
 361 }
 362
 363 memcached_st* memcached_pool_pop(memcached_pool_st* pool,
 364                                  bool block,
 365                                  memcached_return_t *rc)
 366 {
 367   if (pool == NULL)
 368   {
 369     return NULL;
 370   }
 371
 372   memcached_return_t unused;
 373   if (rc == NULL)
 374   {
 375     rc= &unused;
 376   }
 377
 378   memcached_st *memc;
 379   if (block)
 380   {
 381     memc= pool->fetch(pool->timeout(), *rc);
 382   }
 383   else
 384   {
 385     memc= pool->fetch(*rc);
 386   }
 387
 388   return memc;
 389 }
 390
 391 memcached_return_t memcached_pool_release(memcached_pool_st* pool, memcached_st *released)
 392 {
 393   if (pool == NULL)
 394   {
 395     return MEMCACHED_INVALID_ARGUMENTS;
 396   }
 397
 398   memcached_return_t rc;
 399
 400   (void) pool->release(released, rc);
 401
 402   return rc;
 403 }
 404
 405 memcached_return_t memcached_pool_push(memcached_pool_st* pool, memcached_st *released)
 406 {
 407   return memcached_pool_release(pool, released);
 408 }
 409
 410
 411 memcached_return_t memcached_pool_behavior_set(memcached_pool_st *pool,
 412                                                memcached_behavior_t flag,
 413                                                uint64_t data)
 414 {
 415   if (pool == NULL)
 416   {
 417     return MEMCACHED_INVALID_ARGUMENTS;
 418   }
 419
 420   if (pthread_mutex_lock(&pool->mutex))
 421   {
 422     return MEMCACHED_IN_PROGRESS;
 423   }
 424
 425   /* update the master */
 426   memcached_return_t rc= memcached_behavior_set(pool->master, flag, data);
 427   if (memcached_failed(rc))
 428   {
 429     (void)pthread_mutex_unlock(&pool->mutex);
 430     return rc;
 431   }
 432
 433   pool->increment_version();
 434   /* update the clones */
 435   for (int xx= 0; xx <= pool->firstfree; ++xx)
 436   {
 437     if (memcached_success(memcached_behavior_set(pool->server_pool[xx], flag, data)))
 438     {
 439       pool->server_pool[xx]->configure.version= pool->version();
 440     }
 441     else
 442     {
 443       memcached_st *memc;
 444       if ((memc= memcached_clone(NULL, pool->master)))
 445       {
 446         memcached_free(pool->server_pool[xx]);
 447         pool->server_pool[xx]= memc;
 448         /* I'm not sure what to do in this case.. this would happen
 449           if we fail to push the server list inside the client..
 450           I should add a testcase for this, but I believe the following
 451           would work, except that you would add a hole in the pool list..
 452           in theory you could end up with an empty pool....
 453         */
 454       }
 455     }
 456   }
 457
 458   (void)pthread_mutex_unlock(&pool->mutex);
 459
 460   return rc;
 461 }
 462
 463 memcached_return_t memcached_pool_behavior_get(memcached_pool_st *pool,
 464                                                memcached_behavior_t flag,
 465                                                uint64_t *value)
 466 {
 467   if (pool == NULL)
 468   {
 469     return MEMCACHED_INVALID_ARGUMENTS;
 470   }
 471
 472   if (pthread_mutex_lock(&pool->mutex))
 473   {
 474     return MEMCACHED_IN_PROGRESS;
 475   }
 476
 477   *value= memcached_behavior_get(pool->master, flag);
 478
 479   (void)pthread_mutex_unlock(&pool->mutex);
 480
 481   return MEMCACHED_SUCCESS;
 482 }