git.m6w6.name Git - awesomized/libmemcached/blob - libmemcached/util/pool.c

   1 /* LibMemcached
   2  * Copyright (C) 2010 Brian Aker
   3  * All rights reserved.
   4  *
   5  * Use and distribution licensed under the BSD license.  See
   6  * the COPYING file in the parent directory for full text.
   7  *
   8  * Summary: connects to a host, and makes sure it is alive.
   9  *
  10  */
  11
  12 /* -*- Mode: C; tab-width: 2; c-basic-offset: 2; indent-tabs-mode: nil -*- */
  13 #include "libmemcached/common.h"
  14 #include "libmemcached/memcached_util.h"
  15
  16 #include <errno.h>
  17 #include <pthread.h>
  18
  19 struct memcached_pool_st
  20 {
  21   pthread_mutex_t mutex;
  22   pthread_cond_t cond;
  23   memcached_st *master;
  24   memcached_st **mmc;
  25   int firstfree;
  26   uint32_t size;
  27   uint32_t current_size;
  28   bool _owns_master;
  29   char *version;
  30 };
  31
  32 static memcached_return_t mutex_enter(pthread_mutex_t *mutex)
  33 {
  34   int ret;
  35   do
  36     ret= pthread_mutex_lock(mutex);
  37   while (ret == -1 && errno == EINTR);
  38
  39   return (ret == -1) ? MEMCACHED_ERRNO : MEMCACHED_SUCCESS;
  40 }
  41
  42 static memcached_return_t mutex_exit(pthread_mutex_t *mutex)
  43 {
  44   int ret;
  45   do
  46     ret= pthread_mutex_unlock(mutex);
  47   while (ret == -1 && errno == EINTR);
  48
  49   return (ret == -1) ? MEMCACHED_ERRNO : MEMCACHED_SUCCESS;
  50 }
  51
  52 /**
  53  * Grow the connection pool by creating a connection structure and clone the
  54  * original memcached handle.
  55  */
  56 static int grow_pool(memcached_pool_st* pool)
  57 {
  58   memcached_st *obj= calloc(1, sizeof(*obj));
  59
  60   if (obj == NULL)
  61     return -1;
  62
  63   if (memcached_clone(obj, pool->master) == NULL)
  64   {
  65     free(obj);
  66     return -1;
  67   }
  68
  69   pool->mmc[++pool->firstfree] = obj;
  70   pool->current_size++;
  71
  72   return EXIT_SUCCESS;
  73 }
  74
  75 static inline memcached_pool_st *_pool_create(memcached_st* mmc, uint32_t initial, uint32_t max)
  76 {
  77   memcached_pool_st* ret= NULL;
  78
  79   if (! initial || ! max || initial > max)
  80   {
  81     errno= EINVAL;
  82     return NULL;
  83   }
  84
  85   memcached_pool_st object= { .mutex = PTHREAD_MUTEX_INITIALIZER,
  86     .cond= PTHREAD_COND_INITIALIZER,
  87     .master= mmc,
  88     .mmc= calloc(max, sizeof(memcached_st*)),
  89     .firstfree= -1,
  90     .size= max,
  91     .current_size= 0,
  92     ._owns_master= false};
  93
  94   if (object.mmc != NULL)
  95   {
  96     ret= (memcached_pool_st*)calloc(1, sizeof(memcached_pool_st));
  97     if (ret == NULL)
  98     {
  99       free(object.mmc);
 100       errno= ENOMEM; // Set this for the failed calloc
 101       return NULL;
 102     }
 103
 104     *ret= object;
 105
 106     /*
 107       Try to create the initial size of the pool. An allocation failure at
 108       this time is not fatal..
 109     */
 110     for (unsigned int ii= 0; ii < initial; ++ii)
 111     {
 112       if (grow_pool(ret) == -1)
 113         break;
 114     }
 115   }
 116
 117   return ret;
 118 }
 119
 120 memcached_pool_st *memcached_pool_create(memcached_st* mmc, uint32_t initial, uint32_t max)
 121 {
 122   return _pool_create(mmc, initial, max);
 123 }
 124
 125 memcached_pool_st * memcached_pool(const char *option_string, size_t option_string_length)
 126 {
 127   memcached_pool_st *self;
 128   memcached_st *memc= memcached(option_string, option_string_length);
 129
 130   if (! memc)
 131     return NULL;
 132
 133   self= memcached_pool_create(memc, memc->configure.initial_pool_size, memc->configure.max_pool_size);
 134   if (! self)
 135   {
 136     memcached_free(memc);
 137     errno= ENOMEM;
 138     return NULL;
 139   }
 140   errno= 0;
 141
 142   self->_owns_master= true;
 143
 144   return self;
 145 }
 146
 147 memcached_st*  memcached_pool_destroy(memcached_pool_st* pool)
 148 {
 149   if (! pool)
 150     return NULL;
 151
 152   memcached_st *ret= pool->master;
 153
 154   for (int xx= 0; xx <= pool->firstfree; ++xx)
 155   {
 156     memcached_free(pool->mmc[xx]);
 157     free(pool->mmc[xx]);
 158     pool->mmc[xx] = NULL;
 159   }
 160
 161   pthread_mutex_destroy(&pool->mutex);
 162   pthread_cond_destroy(&pool->cond);
 163   free(pool->mmc);
 164   if (pool->_owns_master)
 165   {
 166     memcached_free(pool->master);
 167     ret= NULL;
 168   }
 169   free(pool);
 170
 171   return ret;
 172 }
 173
 174 memcached_st* memcached_pool_pop(memcached_pool_st* pool,
 175                                  bool block,
 176                                  memcached_return_t *rc)
 177 {
 178   if (! pool || ! rc)
 179   {
 180     errno= EINVAL;
 181     return NULL;
 182   }
 183
 184   memcached_st *ret= NULL;
 185   if ((*rc= mutex_enter(&pool->mutex)) != MEMCACHED_SUCCESS)
 186   {
 187     return NULL;
 188   }
 189
 190   do
 191   {
 192     if (pool->firstfree > -1)
 193     {
 194       ret= pool->mmc[pool->firstfree--];
 195     }
 196     else if (pool->current_size == pool->size)
 197     {
 198       if (!block)
 199       {
 200         *rc= mutex_exit(&pool->mutex);
 201         return NULL;
 202       }
 203
 204       if (pthread_cond_wait(&pool->cond, &pool->mutex) == -1)
 205       {
 206         int err= errno;
 207         mutex_exit(&pool->mutex);
 208         errno= err;
 209         *rc= MEMCACHED_ERRNO;
 210         return NULL;
 211       }
 212     }
 213     else if (grow_pool(pool) == -1)
 214     {
 215       *rc= mutex_exit(&pool->mutex);
 216       return NULL;
 217     }
 218   }
 219   while (ret == NULL);
 220
 221   *rc= mutex_exit(&pool->mutex);
 222
 223   return ret;
 224 }
 225
 226 memcached_return_t memcached_pool_push(memcached_pool_st* pool,
 227                                        memcached_st *mmc)
 228 {
 229   if (! pool)
 230     return MEMCACHED_INVALID_ARGUMENTS;
 231
 232   memcached_return_t rc= mutex_enter(&pool->mutex);
 233
 234   if (rc != MEMCACHED_SUCCESS)
 235     return rc;
 236
 237   char* version= memcached_get_user_data(mmc);
 238   /* Someone updated the behavior on the object.. */
 239   if (version != pool->version)
 240   {
 241     memcached_free(mmc);
 242     memset(mmc, 0, sizeof(*mmc));
 243     if (memcached_clone(mmc, pool->master) == NULL)
 244     {
 245       rc= MEMCACHED_SOME_ERRORS;
 246     }
 247   }
 248
 249   pool->mmc[++pool->firstfree]= mmc;
 250
 251   if (pool->firstfree == 0 && pool->current_size == pool->size)
 252   {
 253     /* we might have people waiting for a connection.. wake them up :-) */
 254     pthread_cond_broadcast(&pool->cond);
 255   }
 256
 257   memcached_return_t rval= mutex_exit(&pool->mutex);
 258   if (rc == MEMCACHED_SOME_ERRORS)
 259     return rc;
 260
 261   return rval;
 262 }
 263
 264
 265 memcached_return_t memcached_pool_behavior_set(memcached_pool_st *pool,
 266                                                memcached_behavior_t flag,
 267                                                uint64_t data)
 268 {
 269   if (! pool)
 270     return MEMCACHED_INVALID_ARGUMENTS;
 271
 272   memcached_return_t rc= mutex_enter(&pool->mutex);
 273   if (rc != MEMCACHED_SUCCESS)
 274     return rc;
 275
 276   /* update the master */
 277   rc= memcached_behavior_set(pool->master, flag, data);
 278   if (rc != MEMCACHED_SUCCESS)
 279   {
 280     mutex_exit(&pool->mutex);
 281     return rc;
 282   }
 283
 284   ++pool->version;
 285   memcached_set_user_data(pool->master, pool->version);
 286   /* update the clones */
 287   for (int xx= 0; xx <= pool->firstfree; ++xx)
 288   {
 289     rc= memcached_behavior_set(pool->mmc[xx], flag, data);
 290     if (rc == MEMCACHED_SUCCESS)
 291     {
 292       memcached_set_user_data(pool->mmc[xx], pool->version);
 293     }
 294     else
 295     {
 296       memcached_free(pool->mmc[xx]);
 297       memset(pool->mmc[xx], 0, sizeof(*pool->mmc[xx]));
 298
 299       if (memcached_clone(pool->mmc[xx], pool->master) == NULL)
 300       {
 301         /* I'm not sure what to do in this case.. this would happen
 302           if we fail to push the server list inside the client..
 303           I should add a testcase for this, but I believe the following
 304           would work, except that you would add a hole in the pool list..
 305           in theory you could end up with an empty pool....
 306         */
 307         free(pool->mmc[xx]);
 308         pool->mmc[xx]= NULL;
 309       }
 310     }
 311   }
 312
 313   return mutex_exit(&pool->mutex);
 314 }
 315
 316 memcached_return_t memcached_pool_behavior_get(memcached_pool_st *pool,
 317                                                memcached_behavior_t flag,
 318                                                uint64_t *value)
 319 {
 320   if (! pool)
 321     return MEMCACHED_INVALID_ARGUMENTS;
 322
 323   memcached_return_t rc= mutex_enter(&pool->mutex);
 324   if (rc != MEMCACHED_SUCCESS)
 325   {
 326     return rc;
 327   }
 328
 329   *value= memcached_behavior_get(pool->master, flag);
 330
 331   return mutex_exit(&pool->mutex);
 332 }