Merge from trunk
[m6w6/libmemcached] / 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 }