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

   1 /* -*- Mode: C; tab-width: 2; c-basic-offset: 2; indent-tabs-mode: nil -*- */
   2 #include "libmemcached/common.h"
   3 #include "libmemcached/pool.h"
   4
   5 #include <errno.h>
   6 #include <pthread.h>
   7
   8 struct memcached_pool_st
   9 {
  10   pthread_mutex_t mutex;
  11   pthread_cond_t cond;
  12   memcached_st *master;
  13   memcached_st **mmc;
  14   int firstfree;
  15   uint32_t size;
  16   uint32_t current_size;
  17   char *version;
  18 };
  19
  20 static memcached_return_t mutex_enter(pthread_mutex_t *mutex)
  21 {
  22   int ret;
  23   do
  24     ret= pthread_mutex_lock(mutex);
  25   while (ret == -1 && errno == EINTR);
  26
  27   return (ret == -1) ? MEMCACHED_ERRNO : MEMCACHED_SUCCESS;
  28 }
  29
  30 static memcached_return_t mutex_exit(pthread_mutex_t *mutex)
  31 {
  32   int ret;
  33   do
  34     ret= pthread_mutex_unlock(mutex);
  35   while (ret == -1 && errno == EINTR);
  36
  37   return (ret == -1) ? MEMCACHED_ERRNO : MEMCACHED_SUCCESS;
  38 }
  39
  40 /**
  41  * Grow the connection pool by creating a connection structure and clone the
  42  * original memcached handle.
  43  */
  44 static int grow_pool(memcached_pool_st* pool)
  45 {
  46   memcached_st *obj= calloc(1, sizeof(*obj));
  47
  48   if (obj == NULL)
  49     return -1;
  50
  51   if (memcached_clone(obj, pool->master) == NULL)
  52   {
  53     free(obj);
  54     return -1;
  55   }
  56
  57   pool->mmc[++pool->firstfree] = obj;
  58   pool->current_size++;
  59
  60   return 0;
  61 }
  62
  63 memcached_pool_st *memcached_pool_create(memcached_st* mmc,
  64                                          uint32_t initial, uint32_t max)
  65 {
  66   memcached_pool_st* ret = NULL;
  67   memcached_pool_st object = { .mutex = PTHREAD_MUTEX_INITIALIZER,
  68     .cond = PTHREAD_COND_INITIALIZER,
  69     .master = mmc,
  70     .mmc = calloc(max, sizeof(memcached_st*)),
  71     .firstfree = -1,
  72     .size = max,
  73     .current_size = 0 };
  74
  75   if (object.mmc != NULL)
  76   {
  77     ret= calloc(1, sizeof(*ret));
  78     if (ret == NULL)
  79     {
  80       free(object.mmc);
  81       return NULL;
  82     }
  83
  84     *ret = object;
  85
  86     /*
  87       Try to create the initial size of the pool. An allocation failure at
  88       this time is not fatal..
  89     */
  90     for (unsigned int ii= 0; ii < initial; ++ii)
  91     {
  92       if (grow_pool(ret) == -1)
  93         break;
  94     }
  95   }
  96
  97   return ret;
  98 }
  99
 100 memcached_st*  memcached_pool_destroy(memcached_pool_st* pool)
 101 {
 102   memcached_st *ret = pool->master;
 103
 104   for (int xx= 0; xx <= pool->firstfree; ++xx)
 105   {
 106     memcached_free(pool->mmc[xx]);
 107     free(pool->mmc[xx]);
 108     pool->mmc[xx] = NULL;
 109   }
 110
 111   pthread_mutex_destroy(&pool->mutex);
 112   pthread_cond_destroy(&pool->cond);
 113   free(pool->mmc);
 114   free(pool);
 115
 116   return ret;
 117 }
 118
 119 memcached_st* memcached_pool_pop(memcached_pool_st* pool,
 120                                  bool block,
 121                                  memcached_return_t *rc)
 122 {
 123   memcached_st *ret= NULL;
 124   if ((*rc= mutex_enter(&pool->mutex)) != MEMCACHED_SUCCESS)
 125     return NULL;
 126
 127   do
 128   {
 129     if (pool->firstfree > -1)
 130       ret= pool->mmc[pool->firstfree--];
 131     else if (pool->current_size == pool->size)
 132     {
 133       if (!block)
 134       {
 135         *rc= mutex_exit(&pool->mutex);
 136         return NULL;
 137       }
 138
 139       if (pthread_cond_wait(&pool->cond, &pool->mutex) == -1)
 140       {
 141         int err = errno;
 142         mutex_exit(&pool->mutex);
 143         errno = err;
 144         *rc= MEMCACHED_ERRNO;
 145         return NULL;
 146       }
 147     }
 148     else if (grow_pool(pool) == -1)
 149     {
 150       *rc= mutex_exit(&pool->mutex);
 151       return NULL;
 152     }
 153   }
 154   while (ret == NULL);
 155
 156   *rc= mutex_exit(&pool->mutex);
 157
 158   return ret;
 159 }
 160
 161 memcached_return_t memcached_pool_push(memcached_pool_st* pool,
 162                                        memcached_st *mmc)
 163 {
 164   memcached_return_t rc= mutex_enter(&pool->mutex);
 165
 166   if (rc != MEMCACHED_SUCCESS)
 167     return rc;
 168
 169   char* version= memcached_get_user_data(mmc);
 170   /* Someone updated the behavior on the object.. */
 171   if (version != pool->version)
 172   {
 173     memcached_free(mmc);
 174     memset(mmc, 0, sizeof(*mmc));
 175     if (memcached_clone(mmc, pool->master) == NULL)
 176     {
 177       rc= MEMCACHED_SOME_ERRORS;
 178     }
 179   }
 180
 181   pool->mmc[++pool->firstfree]= mmc;
 182
 183   if (pool->firstfree == 0 && pool->current_size == pool->size)
 184   {
 185     /* we might have people waiting for a connection.. wake them up :-) */
 186     pthread_cond_broadcast(&pool->cond);
 187   }
 188
 189   memcached_return_t rval= mutex_exit(&pool->mutex);
 190   if (rc == MEMCACHED_SOME_ERRORS)
 191     return rc;
 192
 193   return rval;
 194 }
 195
 196
 197 memcached_return_t memcached_pool_behavior_set(memcached_pool_st *pool,
 198                                                memcached_behavior_t flag,
 199                                                uint64_t data)
 200 {
 201
 202   memcached_return_t rc= mutex_enter(&pool->mutex);
 203   if (rc != MEMCACHED_SUCCESS)
 204     return rc;
 205
 206   /* update the master */
 207   rc= memcached_behavior_set(pool->master, flag, data);
 208   if (rc != MEMCACHED_SUCCESS)
 209   {
 210     mutex_exit(&pool->mutex);
 211     return rc;
 212   }
 213
 214   ++pool->version;
 215   memcached_set_user_data(pool->master, pool->version);
 216   /* update the clones */
 217   for (int xx= 0; xx <= pool->firstfree; ++xx)
 218   {
 219     rc= memcached_behavior_set(pool->mmc[xx], flag, data);
 220     if (rc == MEMCACHED_SUCCESS)
 221       memcached_set_user_data(pool->mmc[xx], pool->version);
 222     else
 223     {
 224       memcached_free(pool->mmc[xx]);
 225       memset(pool->mmc[xx], 0, sizeof(*pool->mmc[xx]));
 226       if (memcached_clone(pool->mmc[xx], pool->master) == NULL)
 227       {
 228         /* I'm not sure what to do in this case.. this would happen
 229           if we fail to push the server list inside the client..
 230           I should add a testcase for this, but I believe the following
 231           would work, except that you would add a hole in the pool list..
 232           in theory you could end up with an empty pool....
 233         */
 234         free(pool->mmc[xx]);
 235         pool->mmc[xx]= NULL;
 236       }
 237     }
 238   }
 239
 240   return mutex_exit(&pool->mutex);
 241 }
 242
 243 memcached_return_t memcached_pool_behavior_get(memcached_pool_st *pool,
 244                                                memcached_behavior_t flag,
 245                                                uint64_t *value)
 246 {
 247   memcached_return_t rc= mutex_enter(&pool->mutex);
 248
 249   if (rc != MEMCACHED_SUCCESS)
 250   {
 251     return rc;
 252   }
 253
 254   *value= memcached_behavior_get(pool->master, flag);
 255
 256   return mutex_exit(&pool->mutex);
 257 }