git.m6w6.name Git - m6w6/libmemcached/blob - libmemcachedutil/memcached_pool.c

   1 #include "libmemcached/common.h"
   2 #include "libmemcached/memcached_pool.h"
   3 #include <pthread.h>
   4
   5 struct memcached_pool_st
   6 {
   7   pthread_mutex_t mutex;
   8   pthread_cond_t cond;
   9   memcached_st *master;
  10   memcached_st **mmc;
  11   int firstfree;
  12   uint32_t size;
  13   uint32_t current_size;
  14 };
  15
  16 static memcached_return mutex_enter(pthread_mutex_t *mutex)
  17 {
  18   int ret;
  19   do
  20     ret= pthread_mutex_lock(mutex);
  21   while (ret == -1 && errno == EINTR);
  22
  23   return (ret == -1) ? MEMCACHED_ERRNO : MEMCACHED_SUCCESS;
  24 }
  25
  26 static memcached_return mutex_exit(pthread_mutex_t *mutex) {
  27   int ret;
  28   do
  29     ret= pthread_mutex_unlock(mutex);
  30   while (ret == -1 && errno == EINTR);
  31
  32   return (ret == -1) ? MEMCACHED_ERRNO : MEMCACHED_SUCCESS;
  33 }
  34
  35 /**
  36  * Grow the connection pool by creating a connection structure and clone the
  37  * original memcached handle.
  38  */
  39 static int grow_pool(memcached_pool_st* pool) {
  40   memcached_st *obj= calloc(1, sizeof(*obj));
  41   if (obj == NULL)
  42     return -1;
  43
  44   if (memcached_clone(obj, pool->master) == NULL)
  45   {
  46     free(obj);
  47     return -1;
  48   }
  49
  50   pool->mmc[++pool->firstfree] = obj;
  51   pool->current_size++;
  52
  53   return 0;
  54 }
  55
  56 memcached_pool_st *memcached_pool_create(memcached_st* mmc,
  57                                          uint32_t initial, uint32_t max)
  58 {
  59   memcached_pool_st* ret = NULL;
  60   memcached_pool_st object = { .mutex = PTHREAD_MUTEX_INITIALIZER,
  61                                .cond = PTHREAD_COND_INITIALIZER,
  62                                .master = mmc,
  63                                .mmc = calloc(max, sizeof(memcached_st*)),
  64                                .firstfree = -1,
  65                                .size = max,
  66                                .current_size = 0 };
  67
  68   if (object.mmc != NULL)
  69   {
  70     ret= calloc(1, sizeof(*ret));
  71     if (ret == NULL)
  72     {
  73       free(object.mmc);
  74       return NULL;
  75     }
  76
  77     *ret = object;
  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 ii=0; ii < initial; ++ii)
  83       if (grow_pool(ret) == -1)
  84         break;
  85   }
  86
  87   return ret;
  88 }
  89
  90 memcached_st*  memcached_pool_destroy(memcached_pool_st* pool)
  91 {
  92   memcached_st *ret = pool->master;
  93
  94   for (int xx= 0; xx <= pool->firstfree; ++xx)
  95   {
  96     memcached_free(pool->mmc[xx]);
  97     free(pool->mmc[xx]);
  98     pool->mmc[xx] = NULL;
  99   }
 100
 101   pthread_mutex_destroy(&pool->mutex);
 102   pthread_cond_destroy(&pool->cond);
 103   free(pool->mmc);
 104   free(pool);
 105
 106   return ret;
 107 }
 108
 109 memcached_st* memcached_pool_pop(memcached_pool_st* pool,
 110                                  bool block,
 111                                  memcached_return *rc)
 112 {
 113   memcached_st *ret= NULL;
 114   if ((*rc= mutex_enter(&pool->mutex)) != MEMCACHED_SUCCESS)
 115     return NULL;
 116
 117   do
 118   {
 119     if (pool->firstfree > -1)
 120        ret= pool->mmc[pool->firstfree--];
 121     else if (pool->current_size == pool->size)
 122     {
 123       if (!block)
 124       {
 125         *rc= mutex_exit(&pool->mutex);
 126         return NULL;
 127       }
 128
 129       if (pthread_cond_wait(&pool->cond, &pool->mutex) == -1)
 130       {
 131         int err = errno;
 132         mutex_exit(&pool->mutex);
 133         errno = err;
 134         *rc= MEMCACHED_ERRNO;
 135         return NULL;
 136       }
 137     }
 138     else if (grow_pool(pool) == -1)
 139     {
 140        *rc= mutex_exit(&pool->mutex);
 141        return NULL;
 142     }
 143   }
 144   while (ret == NULL);
 145
 146   *rc= mutex_exit(&pool->mutex);
 147
 148   return ret;
 149 }
 150
 151 memcached_return memcached_pool_push(memcached_pool_st* pool,
 152                                      memcached_st *mmc)
 153 {
 154   memcached_return rc= mutex_enter(&pool->mutex);
 155
 156   if (rc != MEMCACHED_SUCCESS)
 157     return rc;
 158
 159   pool->mmc[++pool->firstfree]= mmc;
 160
 161   if (pool->firstfree == 0 && pool->current_size == pool->size)
 162   {
 163     /* we might have people waiting for a connection.. wake them up :-) */
 164     pthread_cond_broadcast(&pool->cond);
 165   }
 166
 167   return mutex_exit(&pool->mutex);
 168 }