Fix for solaris.
[m6w6/libmemcached] / libmemcached / hosts.c
1 #include "common.h"
2 #include <math.h>
3
4 /* Protoypes (static) */
5 static memcached_return_t server_add(memcached_st *ptr, const char *hostname,
6 in_port_t port,
7 uint32_t weight,
8 memcached_connection_t type);
9 memcached_return_t update_continuum(memcached_st *ptr);
10
11 static int compare_servers(const void *p1, const void *p2)
12 {
13 int return_value;
14 memcached_server_st *a= (memcached_server_st *)p1;
15 memcached_server_st *b= (memcached_server_st *)p2;
16
17 return_value= strcmp(a->hostname, b->hostname);
18
19 if (return_value == 0)
20 {
21 return_value= (int) (a->port - b->port);
22 }
23
24 return return_value;
25 }
26
27 static void sort_hosts(memcached_st *ptr)
28 {
29 if (ptr->number_of_hosts)
30 {
31 qsort(ptr->hosts, ptr->number_of_hosts, sizeof(memcached_server_st), compare_servers);
32 ptr->hosts[0].count= (uint16_t) ptr->number_of_hosts;
33 }
34 }
35
36
37 memcached_return_t run_distribution(memcached_st *ptr)
38 {
39 switch (ptr->distribution)
40 {
41 case MEMCACHED_DISTRIBUTION_CONSISTENT:
42 case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA:
43 case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA_SPY:
44 return update_continuum(ptr);
45 case MEMCACHED_DISTRIBUTION_MODULA:
46 if (ptr->flags.use_sort_hosts)
47 sort_hosts(ptr);
48 break;
49 case MEMCACHED_DISTRIBUTION_RANDOM:
50 srandom((uint32_t) time(NULL));
51 break;
52 case MEMCACHED_DISTRIBUTION_CONSISTENT_MAX:
53 default:
54 WATCHPOINT_ASSERT(0); /* We have added a distribution without extending the logic */
55 }
56
57 ptr->last_disconnected_server = NULL;
58
59 return MEMCACHED_SUCCESS;
60 }
61
62 void server_list_free(memcached_st *ptr, memcached_server_st *servers)
63 {
64 unsigned int x;
65
66 if (servers == NULL)
67 return;
68
69 for (x= 0; x < servers->count; x++)
70 if (servers[x].address_info)
71 {
72 freeaddrinfo(servers[x].address_info);
73 servers[x].address_info= NULL;
74 }
75
76 if (ptr)
77 ptr->call_free(ptr, servers);
78 else
79 free(servers);
80 }
81
82 static uint32_t ketama_server_hash(const char *key, unsigned int key_length, int alignment)
83 {
84 unsigned char results[16];
85
86 md5_signature((unsigned char*)key, key_length, results);
87 return ((uint32_t) (results[3 + alignment * 4] & 0xFF) << 24)
88 | ((uint32_t) (results[2 + alignment * 4] & 0xFF) << 16)
89 | ((uint32_t) (results[1 + alignment * 4] & 0xFF) << 8)
90 | (results[0 + alignment * 4] & 0xFF);
91 }
92
93 static int continuum_item_cmp(const void *t1, const void *t2)
94 {
95 memcached_continuum_item_st *ct1= (memcached_continuum_item_st *)t1;
96 memcached_continuum_item_st *ct2= (memcached_continuum_item_st *)t2;
97
98 /* Why 153? Hmmm... */
99 WATCHPOINT_ASSERT(ct1->value != 153);
100 if (ct1->value == ct2->value)
101 return 0;
102 else if (ct1->value > ct2->value)
103 return 1;
104 else
105 return -1;
106 }
107
108 memcached_return_t update_continuum(memcached_st *ptr)
109 {
110 uint32_t host_index;
111 uint32_t continuum_index= 0;
112 uint32_t value;
113 memcached_server_st *list;
114 uint32_t pointer_index;
115 uint32_t pointer_counter= 0;
116 uint32_t pointer_per_server= MEMCACHED_POINTS_PER_SERVER;
117 uint32_t pointer_per_hash= 1;
118 uint64_t total_weight= 0;
119 uint64_t is_ketama_weighted= 0;
120 uint64_t is_auto_ejecting= 0;
121 uint32_t points_per_server= 0;
122 uint32_t live_servers= 0;
123 struct timeval now;
124
125 if (gettimeofday(&now, NULL) != 0)
126 {
127 ptr->cached_errno = errno;
128 return MEMCACHED_ERRNO;
129 }
130
131 list = ptr->hosts;
132
133 /* count live servers (those without a retry delay set) */
134 is_auto_ejecting= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_AUTO_EJECT_HOSTS);
135 if (is_auto_ejecting)
136 {
137 live_servers= 0;
138 ptr->next_distribution_rebuild= 0;
139 for (host_index= 0; host_index < ptr->number_of_hosts; ++host_index)
140 {
141 if (list[host_index].next_retry <= now.tv_sec)
142 live_servers++;
143 else
144 {
145 if (ptr->next_distribution_rebuild == 0 || list[host_index].next_retry < ptr->next_distribution_rebuild)
146 ptr->next_distribution_rebuild= list[host_index].next_retry;
147 }
148 }
149 }
150 else
151 live_servers= ptr->number_of_hosts;
152
153 is_ketama_weighted= memcached_behavior_get(ptr, MEMCACHED_BEHAVIOR_KETAMA_WEIGHTED);
154 points_per_server= (uint32_t) (is_ketama_weighted ? MEMCACHED_POINTS_PER_SERVER_KETAMA : MEMCACHED_POINTS_PER_SERVER);
155
156 if (live_servers == 0)
157 return MEMCACHED_SUCCESS;
158
159 if (live_servers > ptr->continuum_count)
160 {
161 memcached_continuum_item_st *new_ptr;
162
163 new_ptr= ptr->call_realloc(ptr, ptr->continuum,
164 sizeof(memcached_continuum_item_st) * (live_servers + MEMCACHED_CONTINUUM_ADDITION) * points_per_server);
165
166 if (new_ptr == 0)
167 return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
168
169 ptr->continuum= new_ptr;
170 ptr->continuum_count= live_servers + MEMCACHED_CONTINUUM_ADDITION;
171 }
172
173 if (is_ketama_weighted)
174 {
175 for (host_index = 0; host_index < ptr->number_of_hosts; ++host_index)
176 {
177 if (list[host_index].weight == 0)
178 {
179 list[host_index].weight = 1;
180 }
181 if (!is_auto_ejecting || list[host_index].next_retry <= now.tv_sec)
182 total_weight += list[host_index].weight;
183 }
184 }
185
186 for (host_index = 0; host_index < ptr->number_of_hosts; ++host_index)
187 {
188 if (is_auto_ejecting && list[host_index].next_retry > now.tv_sec)
189 continue;
190
191 if (is_ketama_weighted)
192 {
193 float pct = (float)list[host_index].weight / (float)total_weight;
194 pointer_per_server= (uint32_t) ((floorf((float) (pct * MEMCACHED_POINTS_PER_SERVER_KETAMA / 4 * (float)live_servers + 0.0000000001))) * 4);
195 pointer_per_hash= 4;
196 #ifdef DEBUG
197 printf("ketama_weighted:%s|%d|%llu|%u\n",
198 list[host_index].hostname,
199 list[host_index].port,
200 (unsigned long long)list[host_index].weight,
201 pointer_per_server);
202 #endif
203 }
204
205
206 if (ptr->distribution == MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA_SPY)
207 {
208 for (pointer_index= 0;
209 pointer_index < pointer_per_server / pointer_per_hash;
210 pointer_index++)
211 {
212 char sort_host[MEMCACHED_MAX_HOST_SORT_LENGTH]= "";
213 size_t sort_host_length;
214
215 // Spymemcached ketema key format is: hostname/ip:port-index
216 // If hostname is not available then: /ip:port-index
217 sort_host_length= (size_t) snprintf(sort_host, MEMCACHED_MAX_HOST_SORT_LENGTH,
218 "/%s:%d-%d",
219 list[host_index].hostname,
220 list[host_index].port,
221 pointer_index);
222 #ifdef DEBUG
223 printf("update_continuum: key is %s\n", sort_host);
224 #endif
225
226 WATCHPOINT_ASSERT(sort_host_length);
227
228 if (is_ketama_weighted)
229 {
230 unsigned int i;
231 for (i = 0; i < pointer_per_hash; i++)
232 {
233 value= ketama_server_hash(sort_host, (uint32_t) sort_host_length, (int) i);
234 ptr->continuum[continuum_index].index= host_index;
235 ptr->continuum[continuum_index++].value= value;
236 }
237 }
238 else
239 {
240 value= memcached_generate_hash_value(sort_host, sort_host_length, ptr->distribution_hash);
241 ptr->continuum[continuum_index].index= host_index;
242 ptr->continuum[continuum_index++].value= value;
243 }
244 }
245 }
246 else
247 {
248 for (pointer_index= 1;
249 pointer_index <= pointer_per_server / pointer_per_hash;
250 pointer_index++)
251 {
252 char sort_host[MEMCACHED_MAX_HOST_SORT_LENGTH]= "";
253 size_t sort_host_length;
254
255 if (list[host_index].port == MEMCACHED_DEFAULT_PORT)
256 {
257 sort_host_length= (size_t) snprintf(sort_host, MEMCACHED_MAX_HOST_SORT_LENGTH,
258 "%s-%d",
259 list[host_index].hostname,
260 pointer_index - 1);
261 }
262 else
263 {
264 sort_host_length= (size_t) snprintf(sort_host, MEMCACHED_MAX_HOST_SORT_LENGTH,
265 "%s:%d-%d",
266 list[host_index].hostname,
267 list[host_index].port, pointer_index - 1);
268 }
269
270 WATCHPOINT_ASSERT(sort_host_length);
271
272 if (is_ketama_weighted)
273 {
274 unsigned int i;
275 for (i = 0; i < pointer_per_hash; i++)
276 {
277 value= ketama_server_hash(sort_host, (uint32_t) sort_host_length, (int) i);
278 ptr->continuum[continuum_index].index= host_index;
279 ptr->continuum[continuum_index++].value= value;
280 }
281 }
282 else
283 {
284 value= memcached_generate_hash_value(sort_host, sort_host_length, ptr->distribution_hash);
285 ptr->continuum[continuum_index].index= host_index;
286 ptr->continuum[continuum_index++].value= value;
287 }
288 }
289 }
290
291 pointer_counter+= pointer_per_server;
292 }
293
294 WATCHPOINT_ASSERT(ptr);
295 WATCHPOINT_ASSERT(ptr->continuum);
296 WATCHPOINT_ASSERT(ptr->number_of_hosts * MEMCACHED_POINTS_PER_SERVER <= MEMCACHED_CONTINUUM_SIZE);
297 ptr->continuum_points_counter= pointer_counter;
298 qsort(ptr->continuum, ptr->continuum_points_counter, sizeof(memcached_continuum_item_st), continuum_item_cmp);
299
300 #ifdef DEBUG
301 for (pointer_index= 0; ptr->number_of_hosts && pointer_index < ((live_servers * MEMCACHED_POINTS_PER_SERVER) - 1); pointer_index++)
302 {
303 WATCHPOINT_ASSERT(ptr->continuum[pointer_index].value <= ptr->continuum[pointer_index + 1].value);
304 }
305 #endif
306
307 return MEMCACHED_SUCCESS;
308 }
309
310
311 memcached_return_t memcached_server_push(memcached_st *ptr, memcached_server_st *list)
312 {
313 unsigned int x;
314 uint16_t count;
315 memcached_server_st *new_host_list;
316
317 if (!list)
318 return MEMCACHED_SUCCESS;
319
320 count= list[0].count;
321 new_host_list= ptr->call_realloc(ptr, ptr->hosts,
322 sizeof(memcached_server_st) * (count + ptr->number_of_hosts));
323
324 if (!new_host_list)
325 return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
326
327 ptr->hosts= new_host_list;
328
329 for (x= 0; x < count; x++)
330 {
331 if ((ptr->flags.use_udp && list[x].type != MEMCACHED_CONNECTION_UDP)
332 || ((list[x].type == MEMCACHED_CONNECTION_UDP)
333 && ! (ptr->flags.use_udp)) )
334 return MEMCACHED_INVALID_HOST_PROTOCOL;
335
336 WATCHPOINT_ASSERT(list[x].hostname[0] != 0);
337 memcached_server_create(ptr, &ptr->hosts[ptr->number_of_hosts]);
338 /* TODO check return type */
339 (void)memcached_server_create_with(ptr, &ptr->hosts[ptr->number_of_hosts], list[x].hostname,
340 list[x].port, list[x].weight, list[x].type);
341 ptr->number_of_hosts++;
342 }
343 ptr->hosts[0].count= (uint16_t) ptr->number_of_hosts;
344
345 return run_distribution(ptr);
346 }
347
348 memcached_return_t memcached_server_add_unix_socket(memcached_st *ptr,
349 const char *filename)
350 {
351 return memcached_server_add_unix_socket_with_weight(ptr, filename, 0);
352 }
353
354 memcached_return_t memcached_server_add_unix_socket_with_weight(memcached_st *ptr,
355 const char *filename,
356 uint32_t weight)
357 {
358 if (!filename)
359 return MEMCACHED_FAILURE;
360
361 return server_add(ptr, filename, 0, weight, MEMCACHED_CONNECTION_UNIX_SOCKET);
362 }
363
364 memcached_return_t memcached_server_add_udp(memcached_st *ptr,
365 const char *hostname,
366 in_port_t port)
367 {
368 return memcached_server_add_udp_with_weight(ptr, hostname, port, 0);
369 }
370
371 memcached_return_t memcached_server_add_udp_with_weight(memcached_st *ptr,
372 const char *hostname,
373 in_port_t port,
374 uint32_t weight)
375 {
376 if (!port)
377 port= MEMCACHED_DEFAULT_PORT;
378
379 if (!hostname)
380 hostname= "localhost";
381
382 return server_add(ptr, hostname, port, weight, MEMCACHED_CONNECTION_UDP);
383 }
384
385 memcached_return_t memcached_server_add(memcached_st *ptr,
386 const char *hostname,
387 in_port_t port)
388 {
389 return memcached_server_add_with_weight(ptr, hostname, port, 0);
390 }
391
392 memcached_return_t memcached_server_add_with_weight(memcached_st *ptr,
393 const char *hostname,
394 in_port_t port,
395 uint32_t weight)
396 {
397 if (!port)
398 port= MEMCACHED_DEFAULT_PORT;
399
400 if (!hostname)
401 hostname= "localhost";
402
403 return server_add(ptr, hostname, port, weight, MEMCACHED_CONNECTION_TCP);
404 }
405
406 static memcached_return_t server_add(memcached_st *ptr, const char *hostname,
407 in_port_t port,
408 uint32_t weight,
409 memcached_connection_t type)
410 {
411 memcached_server_st *new_host_list;
412
413 if ( (ptr->flags.use_udp && type != MEMCACHED_CONNECTION_UDP)
414 || ( (type == MEMCACHED_CONNECTION_UDP) && (! ptr->flags.use_udp) ) )
415 return MEMCACHED_INVALID_HOST_PROTOCOL;
416
417 new_host_list= ptr->call_realloc(ptr, ptr->hosts,
418 sizeof(memcached_server_st) * (ptr->number_of_hosts+1));
419
420 if (new_host_list == NULL)
421 return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
422
423 ptr->hosts= new_host_list;
424
425 /* TODO: Check return type */
426 (void)memcached_server_create_with(ptr, &ptr->hosts[ptr->number_of_hosts], hostname, port, weight, type);
427 ptr->number_of_hosts++;
428 ptr->hosts[0].count= (uint16_t) ptr->number_of_hosts;
429
430 return run_distribution(ptr);
431 }
432
433 memcached_return_t memcached_server_remove(memcached_server_st *st_ptr)
434 {
435 uint32_t x, host_index;
436 memcached_st *ptr= st_ptr->root;
437 memcached_server_st *list= ptr->hosts;
438
439 for (x= 0, host_index= 0; x < ptr->number_of_hosts; x++)
440 {
441 if (strncmp(list[x].hostname, st_ptr->hostname, MEMCACHED_MAX_HOST_LENGTH) != 0 || list[x].port != st_ptr->port)
442 {
443 if (host_index != x)
444 memcpy(list+host_index, list+x, sizeof(memcached_server_st));
445 host_index++;
446 }
447 }
448 ptr->number_of_hosts= host_index;
449
450 if (st_ptr->address_info)
451 {
452 freeaddrinfo(st_ptr->address_info);
453 st_ptr->address_info= NULL;
454 }
455 run_distribution(ptr);
456
457 return MEMCACHED_SUCCESS;
458 }
459
460 memcached_server_st *memcached_server_list_append(memcached_server_st *ptr,
461 const char *hostname, in_port_t port,
462 memcached_return_t *error)
463 {
464 return memcached_server_list_append_with_weight(ptr, hostname, port, 0, error);
465 }
466
467 memcached_server_st *memcached_server_list_append_with_weight(memcached_server_st *ptr,
468 const char *hostname, in_port_t port,
469 uint32_t weight,
470 memcached_return_t *error)
471 {
472 unsigned int count;
473 memcached_server_st *new_host_list;
474
475 if (hostname == NULL || error == NULL)
476 return NULL;
477
478 if (!port)
479 port= MEMCACHED_DEFAULT_PORT;
480
481 /* Increment count for hosts */
482 count= 1;
483 if (ptr != NULL)
484 {
485 count+= ptr[0].count;
486 }
487
488 new_host_list= (memcached_server_st *)realloc(ptr, sizeof(memcached_server_st) * count);
489 if (!new_host_list)
490 {
491 *error= MEMCACHED_MEMORY_ALLOCATION_FAILURE;
492 return NULL;
493 }
494
495 /* TODO: Check return type */
496 memcached_server_create_with(NULL, &new_host_list[count-1], hostname, port, weight, MEMCACHED_CONNECTION_TCP);
497
498 /* Backwards compatibility hack */
499 new_host_list[0].count= (uint16_t) count;
500
501 *error= MEMCACHED_SUCCESS;
502 return new_host_list;
503 }
504
505 unsigned int memcached_server_list_count(memcached_server_st *ptr)
506 {
507 if (ptr == NULL)
508 return 0;
509
510 return ptr[0].count;
511 }
512
513 void memcached_server_list_free(memcached_server_st *ptr)
514 {
515 server_list_free(NULL, ptr);
516 }