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

   1 #include "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   int size;
  13   int current_size;
  14 };
  15
  16 /**
  17  * Lock a the pthread mutex and handle error conditions. If we fail to
  18  * lock the mutex there must be something really wrong and we cannot continue.
  19  */
  20 static void 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   if (ret == -1)
  28   {
  29     /*
  30     ** This means something is seriously wrong (deadlock or an internal
  31     ** error in the posix library. Print out an error message and abort
  32     ** the program.
  33     */
  34     fprintf(stderr, "pthread_mutex_lock failed: %s\n", strerror(errno));
  35     fflush(stderr);
  36     abort();
  37   }
  38 }
  39
  40 /**
  41  * Unlock a the pthread mutex and handle error conditions.
  42  * All errors except EINTR is fatal errors and will terminate the program
  43  * with a coredump.
  44  */
  45 static void mutex_exit(pthread_mutex_t *mutex) {
  46   int ret;
  47   do
  48     ret = pthread_mutex_unlock(mutex);
  49   while (ret == -1 && errno == EINTR);
  50
  51   if (ret == -1)
  52   {
  53     /*
  54     ** This means something is seriously wrong (deadlock or an internal
  55     ** error in the posix library. Print out an error message and abort
  56     ** the program.
  57     */
  58     fprintf(stderr, "pthread_mutex_unlock %s\n", strerror(errno));
  59     fflush(stderr);
  60     abort();
  61   }
  62 }
  63
  64 /**
  65  * Grow the connection pool by creating a connection structure and clone the
  66  * original memcached handle.
  67  */
  68 static int grow_pool(memcached_pool_st* pool) {
  69   memcached_st *obj = calloc(1, sizeof(*obj));
  70   if (obj == NULL)
  71     return -1;
  72
  73   if (memcached_clone(obj, pool->master) == NULL)
  74   {
  75     free(obj);
  76     return -1;
  77   }
  78
  79   pool->mmc[++pool->firstfree] = obj;
  80   pool->current_size++;
  81
  82   return 0;
  83 }
  84
  85 memcached_pool_st *memcached_pool_create(memcached_st* mmc, int initial, int max)
  86 {
  87   memcached_pool_st* ret = NULL;
  88   memcached_pool_st object = { .mutex = PTHREAD_MUTEX_INITIALIZER,
  89                                .cond = PTHREAD_COND_INITIALIZER,
  90                                .master = mmc,
  91                                .mmc = calloc(max, sizeof(memcached_st*)),
  92                                .firstfree = -1,
  93                                .size = max,
  94                                .current_size = 0 };
  95
  96   if (object.mmc != NULL)
  97   {
  98     ret = calloc(1, sizeof(*ret));
  99     if (ret == NULL)
 100     {
 101       free(object.mmc);
 102       return NULL;
 103     }
 104
 105     *ret = object;
 106
 107     /* Try to create the initial size of the pool. An allocation failure at
 108      * this time is not fatal..
 109      */
 110     for (int ii=0; ii < initial; ++ii)
 111       if (grow_pool(ret) == -1)
 112         break;
 113   }
 114
 115   return ret;
 116 }
 117
 118 memcached_st*  memcached_pool_destroy(memcached_pool_st* pool)
 119 {
 120   memcached_st *ret = pool->master;
 121
 122   for (int ii = 0; ii <= pool->firstfree; ++ii)
 123   {
 124     memcached_free(pool->mmc[ii]);
 125     free(pool->mmc[ii]);
 126     pool->mmc[ii] = NULL;
 127   }
 128
 129   pthread_mutex_destroy(&pool->mutex);
 130   pthread_cond_destroy(&pool->cond);
 131   free(pool->mmc);
 132   free(pool);
 133
 134   return ret;
 135 }
 136
 137 memcached_st* memcached_pool_pop(memcached_pool_st* pool, bool block) {
 138   memcached_st *ret = NULL;
 139   mutex_enter(&pool->mutex);
 140   do
 141   {
 142     if (pool->firstfree > -1)
 143       ret = pool->mmc[pool->firstfree--];
 144     else if (pool->current_size == pool->size)
 145     {
 146       if (!block)
 147       {
 148         mutex_exit(&pool->mutex);
 149         return NULL;
 150       }
 151
 152       if (pthread_cond_wait(&pool->cond, &pool->mutex) == -1)
 153       {
 154         /*
 155         ** This means something is seriously wrong (an internal error in the
 156         ** posix library. Print out an error message and abort the program.
 157         */
 158         fprintf(stderr, "pthread cond_wait %s\n", strerror(errno));
 159         fflush(stderr);
 160         abort();
 161       }
 162     }
 163     else if (grow_pool(pool) == -1)
 164     {
 165       mutex_exit(&pool->mutex);
 166       return NULL;
 167     }
 168   }
 169   while (ret == NULL);
 170
 171   mutex_exit(&pool->mutex);
 172   return ret;
 173 }
 174
 175 void memcached_pool_push(memcached_pool_st* pool, memcached_st *mmc)
 176 {
 177   mutex_enter(&pool->mutex);
 178   pool->mmc[++pool->firstfree] = mmc;
 179
 180   if (pool->firstfree == 0 && pool->current_size == pool->size)
 181   {
 182     /* we might have people waiting for a connection.. wake them up :-) */
 183     pthread_cond_broadcast(&pool->cond);
 184   }
 185   mutex_exit(&pool->mutex);
 186 }