git.m6w6.name Git - m6w6/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 == 0 or max == 0 or (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 (object == NULL)
 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 (memc == NULL)
 203   {
 204     return NULL;
 205   }
 206
 207   memcached_pool_st *self= memcached_pool_create(memc, memc->configure.initial_pool_size, memc->configure.max_pool_size);
 208   if (self == NULL)
 209   {
 210     memcached_free(memc);
 211     errno= ENOMEM;
 212     return NULL;
 213   }
 214   errno= 0;
 215
 216   self->_owns_master= true;
 217
 218   return self;
 219 }
 220
 221 memcached_st*  memcached_pool_destroy(memcached_pool_st* pool)
 222 {
 223   if (pool == NULL)
 224   {
 225     return NULL;
 226   }
 227
 228   // Legacy that we return the original structure
 229   memcached_st *ret= NULL;
 230   if (pool->_owns_master)
 231   { }
 232   else
 233   {
 234     ret= pool->master;
 235   }
 236
 237   delete pool;
 238
 239   return ret;
 240 }
 241
 242 memcached_st* memcached_pool_pop(memcached_pool_st* pool,
 243                                  bool block,
 244                                  memcached_return_t *rc)
 245 {
 246   memcached_return_t not_used;
 247   if (rc == NULL)
 248   {
 249     rc= &not_used;
 250   }
 251
 252   if (pool == NULL)
 253   {
 254     *rc= MEMCACHED_INVALID_ARGUMENTS;
 255     errno= EINVAL;
 256
 257     return NULL;
 258   }
 259
 260   if (memcached_failed((*rc= mutex_enter(&pool->mutex))))
 261   {
 262     return NULL;
 263   }
 264
 265   memcached_st *ret= NULL;
 266   do
 267   {
 268     if (pool->firstfree > -1)
 269     {
 270       ret= pool->server_pool[pool->firstfree--];
 271     }
 272     else if (pool->current_size == pool->size)
 273     {
 274       if (block == false)
 275       {
 276         *rc= mutex_exit(&pool->mutex); // this should be a different error
 277         return NULL;
 278       }
 279
 280       struct timespec time_to_wait= {0, 0};
 281       time_to_wait.tv_sec = time(NULL) +5;
 282       int thread_ret;
 283       if ((thread_ret= pthread_cond_timedwait(&pool->cond, &pool->mutex, &time_to_wait)) != 0)
 284       {
 285         mutex_exit(&pool->mutex);
 286         errno= thread_ret;
 287
 288         if (thread_ret == ETIMEDOUT)
 289         {
 290           *rc= MEMCACHED_TIMEOUT;
 291         }
 292         else
 293         {
 294           *rc= MEMCACHED_ERRNO;
 295         }
 296
 297         return NULL;
 298       }
 299     }
 300     else if (not grow_pool(pool))
 301     {
 302       (void)mutex_exit(&pool->mutex);
 303       *rc= MEMCACHED_MEMORY_ALLOCATION_FAILURE;
 304       return NULL;
 305     }
 306   } while (ret == NULL);
 307
 308   *rc= mutex_exit(&pool->mutex);
 309
 310   return ret;
 311 }
 312
 313 memcached_return_t memcached_pool_push(memcached_pool_st* pool, memcached_st *released)
 314 {
 315   if (pool == NULL)
 316   {
 317     return MEMCACHED_INVALID_ARGUMENTS;
 318   }
 319
 320   if (released == NULL)
 321   {
 322     return MEMCACHED_INVALID_ARGUMENTS;
 323   }
 324
 325   memcached_return_t rc= mutex_enter(&pool->mutex);
 326
 327   if (rc != MEMCACHED_SUCCESS)
 328   {
 329     return rc;
 330   }
 331
 332   /* Someone updated the behavior on the object.. */
 333   if (pool->compare_version(released) == false)
 334   {
 335     memcached_free(released);
 336     if (not (released= memcached_clone(NULL, pool->master)))
 337     {
 338       rc= MEMCACHED_SOME_ERRORS;
 339     }
 340   }
 341
 342   pool->server_pool[++pool->firstfree]= released;
 343
 344   if (pool->firstfree == 0 and pool->current_size == pool->size)
 345   {
 346     /* we might have people waiting for a connection.. wake them up :-) */
 347     pthread_cond_broadcast(&pool->cond);
 348   }
 349
 350   memcached_return_t rval= mutex_exit(&pool->mutex);
 351   if (rc == MEMCACHED_SOME_ERRORS)
 352   {
 353     return rc;
 354   }
 355
 356   return rval;
 357 }
 358
 359
 360 memcached_return_t memcached_pool_behavior_set(memcached_pool_st *pool,
 361                                                memcached_behavior_t flag,
 362                                                uint64_t data)
 363 {
 364   if (pool == NULL)
 365   {
 366     return MEMCACHED_INVALID_ARGUMENTS;
 367   }
 368
 369   memcached_return_t rc= mutex_enter(&pool->mutex);
 370   if (rc != MEMCACHED_SUCCESS)
 371     return rc;
 372
 373   /* update the master */
 374   rc= memcached_behavior_set(pool->master, flag, data);
 375   if (rc != MEMCACHED_SUCCESS)
 376   {
 377     mutex_exit(&pool->mutex);
 378     return rc;
 379   }
 380
 381   pool->increment_version();
 382   /* update the clones */
 383   for (int xx= 0; xx <= pool->firstfree; ++xx)
 384   {
 385     rc= memcached_behavior_set(pool->server_pool[xx], flag, data);
 386     if (rc == MEMCACHED_SUCCESS)
 387     {
 388       pool->server_pool[xx]->configure.version= pool->version();
 389     }
 390     else
 391     {
 392       memcached_free(pool->server_pool[xx]);
 393       if (not (pool->server_pool[xx]= memcached_clone(NULL, pool->master)))
 394       {
 395         /* I'm not sure what to do in this case.. this would happen
 396           if we fail to push the server list inside the client..
 397           I should add a testcase for this, but I believe the following
 398           would work, except that you would add a hole in the pool list..
 399           in theory you could end up with an empty pool....
 400         */
 401       }
 402     }
 403   }
 404
 405   return mutex_exit(&pool->mutex);
 406 }
 407
 408 memcached_return_t memcached_pool_behavior_get(memcached_pool_st *pool,
 409                                                memcached_behavior_t flag,
 410                                                uint64_t *value)
 411 {
 412   if (pool == NULL)
 413   {
 414     return MEMCACHED_INVALID_ARGUMENTS;
 415   }
 416
 417   memcached_return_t rc= mutex_enter(&pool->mutex);
 418   if (rc != MEMCACHED_SUCCESS)
 419   {
 420     return rc;
 421   }
 422
 423   *value= memcached_behavior_get(pool->master, flag);
 424
 425   return mutex_exit(&pool->mutex);
 426 }