Extend lp:655423 just to see if we can trigger it via any other method.
[awesomized/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 || ! max || 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 (not object)
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 (not memc)
203 return NULL;
204
205 memcached_pool_st *self;
206 self= memcached_pool_create(memc, memc->configure.initial_pool_size, memc->configure.max_pool_size);
207 if (not self)
208 {
209 memcached_free(memc);
210 errno= ENOMEM;
211 return NULL;
212 }
213 errno= 0;
214
215 self->_owns_master= true;
216
217 return self;
218 }
219
220 memcached_st* memcached_pool_destroy(memcached_pool_st* pool)
221 {
222 if (not pool)
223 return NULL;
224
225 // Legacy that we return the original structure
226 memcached_st *ret= NULL;
227 if (pool->_owns_master)
228 { }
229 else
230 {
231 ret= pool->master;
232 }
233
234 delete pool;
235
236 return ret;
237 }
238
239 memcached_st* memcached_pool_pop(memcached_pool_st* pool,
240 bool block,
241 memcached_return_t *rc)
242 {
243 assert(pool);
244 assert(rc);
245 if (not pool || not rc)
246 {
247 errno= EINVAL;
248 return NULL;
249 }
250
251 if ((*rc= mutex_enter(&pool->mutex)) != MEMCACHED_SUCCESS)
252 {
253 return NULL;
254 }
255
256 memcached_st *ret= NULL;
257 do
258 {
259 if (pool->firstfree > -1)
260 {
261 ret= pool->server_pool[pool->firstfree--];
262 }
263 else if (pool->current_size == pool->size)
264 {
265 if (not block)
266 {
267 *rc= mutex_exit(&pool->mutex); // this should be a different error
268 return NULL;
269 }
270
271 if (pthread_cond_wait(&pool->cond, &pool->mutex) == -1)
272 {
273 int err= errno;
274 mutex_exit(&pool->mutex);
275 errno= err;
276 *rc= MEMCACHED_ERRNO;
277 return NULL;
278 }
279 }
280 else if (not grow_pool(pool))
281 {
282 (void)mutex_exit(&pool->mutex);
283 *rc= MEMCACHED_MEMORY_ALLOCATION_FAILURE;
284 return NULL;
285 }
286 }
287 while (ret == NULL);
288
289 *rc= mutex_exit(&pool->mutex);
290
291 return ret;
292 }
293
294 memcached_return_t memcached_pool_push(memcached_pool_st* pool, memcached_st *released)
295 {
296 if (not pool)
297 return MEMCACHED_INVALID_ARGUMENTS;
298
299 memcached_return_t rc= mutex_enter(&pool->mutex);
300
301 if (rc != MEMCACHED_SUCCESS)
302 return rc;
303
304 /* Someone updated the behavior on the object.. */
305 if (not pool->compare_version(released))
306 {
307 memcached_free(released);
308 if (not (released= memcached_clone(NULL, pool->master)))
309 {
310 rc= MEMCACHED_SOME_ERRORS;
311 }
312 }
313
314 pool->server_pool[++pool->firstfree]= released;
315
316 if (pool->firstfree == 0 && pool->current_size == pool->size)
317 {
318 /* we might have people waiting for a connection.. wake them up :-) */
319 pthread_cond_broadcast(&pool->cond);
320 }
321
322 memcached_return_t rval= mutex_exit(&pool->mutex);
323 if (rc == MEMCACHED_SOME_ERRORS)
324 return rc;
325
326 return rval;
327 }
328
329
330 memcached_return_t memcached_pool_behavior_set(memcached_pool_st *pool,
331 memcached_behavior_t flag,
332 uint64_t data)
333 {
334 if (not pool)
335 return MEMCACHED_INVALID_ARGUMENTS;
336
337 memcached_return_t rc= mutex_enter(&pool->mutex);
338 if (rc != MEMCACHED_SUCCESS)
339 return rc;
340
341 /* update the master */
342 rc= memcached_behavior_set(pool->master, flag, data);
343 if (rc != MEMCACHED_SUCCESS)
344 {
345 mutex_exit(&pool->mutex);
346 return rc;
347 }
348
349 pool->increment_version();
350 /* update the clones */
351 for (int xx= 0; xx <= pool->firstfree; ++xx)
352 {
353 rc= memcached_behavior_set(pool->server_pool[xx], flag, data);
354 if (rc == MEMCACHED_SUCCESS)
355 {
356 pool->server_pool[xx]->configure.version= pool->version();
357 }
358 else
359 {
360 memcached_free(pool->server_pool[xx]);
361 if (not (pool->server_pool[xx]= memcached_clone(NULL, pool->master)))
362 {
363 /* I'm not sure what to do in this case.. this would happen
364 if we fail to push the server list inside the client..
365 I should add a testcase for this, but I believe the following
366 would work, except that you would add a hole in the pool list..
367 in theory you could end up with an empty pool....
368 */
369 }
370 }
371 }
372
373 return mutex_exit(&pool->mutex);
374 }
375
376 memcached_return_t memcached_pool_behavior_get(memcached_pool_st *pool,
377 memcached_behavior_t flag,
378 uint64_t *value)
379 {
380 if (! pool)
381 return MEMCACHED_INVALID_ARGUMENTS;
382
383 memcached_return_t rc= mutex_enter(&pool->mutex);
384 if (rc != MEMCACHED_SUCCESS)
385 {
386 return rc;
387 }
388
389 *value= memcached_behavior_get(pool->master, flag);
390
391 return mutex_exit(&pool->mutex);
392 }