git.m6w6.name Git - awesomized/libmemcached/blob - libmemcached/util/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 <libmemcached/common.h>
  40 #include <libmemcached/memcached_util.h>
  41
  42 #include <cassert>
  43 #include <cerrno>
  44 #include <pthread.h>
  45 #include <memory>
  46
  47 static bool grow_pool(memcached_pool_st* pool);
  48
  49 struct memcached_pool_st
  50 {
  51   pthread_mutex_t mutex;
  52   pthread_cond_t cond;
  53   memcached_st *master;
  54   memcached_st **server_pool;
  55   int firstfree;
  56   const uint32_t size;
  57   uint32_t current_size;
  58   bool _owns_master;
  59
  60   memcached_pool_st(memcached_st *master_arg, size_t max_arg) :
  61     master(master_arg),
  62     server_pool(NULL),
  63     firstfree(-1),
  64     size(max_arg),
  65     current_size(0),
  66     _owns_master(false)
  67   {
  68     pthread_mutex_init(&mutex, NULL);
  69     pthread_cond_init(&cond, NULL);
  70   }
  71
  72   bool init(uint32_t initial)
  73   {
  74     server_pool= new (std::nothrow) memcached_st *[size];
  75     if (not server_pool)
  76       return false;
  77
  78     /*
  79       Try to create the initial size of the pool. An allocation failure at
  80       this time is not fatal..
  81     */
  82     for (unsigned int x= 0; x < initial; ++x)
  83     {
  84       if (not grow_pool(this))
  85         break;
  86     }
  87
  88     return true;
  89   }
  90
  91   ~memcached_pool_st()
  92   {
  93     for (int x= 0; x <= firstfree; ++x)
  94     {
  95       memcached_free(server_pool[x]);
  96       server_pool[x] = NULL;
  97     }
  98
  99     pthread_mutex_destroy(&mutex);
 100     pthread_cond_destroy(&cond);
 101     delete [] server_pool;
 102     if (_owns_master)
 103     {
 104       memcached_free(master);
 105     }
 106   }
 107
 108   void increment_version()
 109   {
 110     ++master->configure.version;
 111   }
 112
 113   bool compare_version(const memcached_st *arg) const
 114   {
 115     return (arg->configure.version == version());
 116   }
 117
 118   int32_t version() const
 119   {
 120     return master->configure.version;
 121   }
 122 };
 123
 124 static memcached_return_t mutex_enter(pthread_mutex_t *mutex)
 125 {
 126   int ret;
 127   do
 128   {
 129     ret= pthread_mutex_lock(mutex);
 130   } while (ret == -1 && errno == EINTR);
 131
 132   return (ret == -1) ? MEMCACHED_ERRNO : MEMCACHED_SUCCESS;
 133 }
 134
 135 static memcached_return_t mutex_exit(pthread_mutex_t *mutex)
 136 {
 137   int ret;
 138   do
 139   {
 140     ret= pthread_mutex_unlock(mutex);
 141   } while (ret == -1 && errno == EINTR);
 142
 143   return (ret == -1) ? MEMCACHED_ERRNO : MEMCACHED_SUCCESS;
 144 }
 145
 146 /**
 147  * Grow the connection pool by creating a connection structure and clone the
 148  * original memcached handle.
 149  */
 150 static bool grow_pool(memcached_pool_st* pool)
 151 {
 152   memcached_st *obj;
 153   if (not (obj= memcached_clone(NULL, pool->master)))
 154   {
 155     return false;
 156   }
 157
 158   pool->server_pool[++pool->firstfree]= obj;
 159   pool->current_size++;
 160   obj->configure.version= pool->version();
 161
 162   return true;
 163 }
 164
 165 static inline memcached_pool_st *_pool_create(memcached_st* master, uint32_t initial, uint32_t max)
 166 {
 167   if (! initial || ! max || initial > max)
 168   {
 169     errno= EINVAL;
 170     return NULL;
 171   }
 172
 173   memcached_pool_st *object= new (std::nothrow) memcached_pool_st(master, max);
 174   if (not object)
 175   {
 176     errno= ENOMEM; // Set this for the failed calloc
 177     return NULL;
 178   }
 179
 180   /*
 181     Try to create the initial size of the pool. An allocation failure at
 182     this time is not fatal..
 183   */
 184   if (not object->init(initial))
 185   {
 186     delete object;
 187     return NULL;
 188   }
 189
 190   return object;
 191 }
 192
 193 memcached_pool_st *memcached_pool_create(memcached_st* master, uint32_t initial, uint32_t max)
 194 {
 195   return _pool_create(master, initial, max);
 196 }
 197
 198 memcached_pool_st * memcached_pool(const char *option_string, size_t option_string_length)
 199 {
 200   memcached_st *memc= memcached(option_string, option_string_length);
 201
 202   if (not memc)
 203     return NULL;
 204
 205   memcached_pool_st *self;
 206   self= memcached_pool_create(memc, memc->configure.initial_pool_size, memc->configure.max_pool_size);
 207   if (not self)
 208   {
 209     memcached_free(memc);
 210     errno= ENOMEM;
 211     return NULL;
 212   }
 213   errno= 0;
 214
 215   self->_owns_master= true;
 216
 217   return self;
 218 }
 219
 220 memcached_st*  memcached_pool_destroy(memcached_pool_st* pool)
 221 {
 222   if (not pool)
 223     return NULL;
 224
 225   // Legacy that we return the original structure
 226   memcached_st *ret= NULL;
 227   if (pool->_owns_master)
 228   { }
 229   else
 230   {
 231     ret= pool->master;
 232   }
 233
 234   delete pool;
 235
 236   return ret;
 237 }
 238
 239 memcached_st* memcached_pool_pop(memcached_pool_st* pool,
 240                                  bool block,
 241                                  memcached_return_t *rc)
 242 {
 243   assert(pool);
 244   assert(rc);
 245   if (not pool ||  not rc)
 246   {
 247     errno= EINVAL;
 248     return NULL;
 249   }
 250
 251   if ((*rc= mutex_enter(&pool->mutex)) != MEMCACHED_SUCCESS)
 252   {
 253     return NULL;
 254   }
 255
 256   memcached_st *ret= NULL;
 257   do
 258   {
 259     if (pool->firstfree > -1)
 260     {
 261       ret= pool->server_pool[pool->firstfree--];
 262     }
 263     else if (pool->current_size == pool->size)
 264     {
 265       if (not block)
 266       {
 267         *rc= mutex_exit(&pool->mutex); // this should be a different error
 268         return NULL;
 269       }
 270
 271       if (pthread_cond_wait(&pool->cond, &pool->mutex) == -1)
 272       {
 273         int err= errno;
 274         mutex_exit(&pool->mutex);
 275         errno= err;
 276         *rc= MEMCACHED_ERRNO;
 277         return NULL;
 278       }
 279     }
 280     else if (not grow_pool(pool))
 281     {
 282       (void)mutex_exit(&pool->mutex);
 283       *rc= MEMCACHED_MEMORY_ALLOCATION_FAILURE;
 284       return NULL;
 285     }
 286   }
 287   while (ret == NULL);
 288
 289   *rc= mutex_exit(&pool->mutex);
 290
 291   return ret;
 292 }
 293
 294 memcached_return_t memcached_pool_push(memcached_pool_st* pool, memcached_st *released)
 295 {
 296   if (not pool)
 297     return MEMCACHED_INVALID_ARGUMENTS;
 298
 299   memcached_return_t rc= mutex_enter(&pool->mutex);
 300
 301   if (rc != MEMCACHED_SUCCESS)
 302     return rc;
 303
 304   /* Someone updated the behavior on the object.. */
 305   if (not pool->compare_version(released))
 306   {
 307     memcached_free(released);
 308     if (not (released= memcached_clone(NULL, pool->master)))
 309     {
 310       rc= MEMCACHED_SOME_ERRORS;
 311     }
 312   }
 313
 314   pool->server_pool[++pool->firstfree]= released;
 315
 316   if (pool->firstfree == 0 && pool->current_size == pool->size)
 317   {
 318     /* we might have people waiting for a connection.. wake them up :-) */
 319     pthread_cond_broadcast(&pool->cond);
 320   }
 321
 322   memcached_return_t rval= mutex_exit(&pool->mutex);
 323   if (rc == MEMCACHED_SOME_ERRORS)
 324     return rc;
 325
 326   return rval;
 327 }
 328
 329
 330 memcached_return_t memcached_pool_behavior_set(memcached_pool_st *pool,
 331                                                memcached_behavior_t flag,
 332                                                uint64_t data)
 333 {
 334   if (not pool)
 335     return MEMCACHED_INVALID_ARGUMENTS;
 336
 337   memcached_return_t rc= mutex_enter(&pool->mutex);
 338   if (rc != MEMCACHED_SUCCESS)
 339     return rc;
 340
 341   /* update the master */
 342   rc= memcached_behavior_set(pool->master, flag, data);
 343   if (rc != MEMCACHED_SUCCESS)
 344   {
 345     mutex_exit(&pool->mutex);
 346     return rc;
 347   }
 348
 349   pool->increment_version();
 350   /* update the clones */
 351   for (int xx= 0; xx <= pool->firstfree; ++xx)
 352   {
 353     rc= memcached_behavior_set(pool->server_pool[xx], flag, data);
 354     if (rc == MEMCACHED_SUCCESS)
 355     {
 356       pool->server_pool[xx]->configure.version= pool->version();
 357     }
 358     else
 359     {
 360       memcached_free(pool->server_pool[xx]);
 361       if (not (pool->server_pool[xx]= memcached_clone(NULL, pool->master)))
 362       {
 363         /* I'm not sure what to do in this case.. this would happen
 364           if we fail to push the server list inside the client..
 365           I should add a testcase for this, but I believe the following
 366           would work, except that you would add a hole in the pool list..
 367           in theory you could end up with an empty pool....
 368         */
 369       }
 370     }
 371   }
 372
 373   return mutex_exit(&pool->mutex);
 374 }
 375
 376 memcached_return_t memcached_pool_behavior_get(memcached_pool_st *pool,
 377                                                memcached_behavior_t flag,
 378                                                uint64_t *value)
 379 {
 380   if (! pool)
 381     return MEMCACHED_INVALID_ARGUMENTS;
 382
 383   memcached_return_t rc= mutex_enter(&pool->mutex);
 384   if (rc != MEMCACHED_SUCCESS)
 385   {
 386     return rc;
 387   }
 388
 389   *value= memcached_behavior_get(pool->master, flag);
 390
 391   return mutex_exit(&pool->mutex);
 392 }