2 * Copyright (C) 2006-2010 Brian Aker
5 * Use and distribution licensed under the BSD license. See
6 * the COPYING file in the parent directory for full text.
15 /* Protoypes (static) */
16 static memcached_return_t
server_add(memcached_st
*ptr
, const char *hostname
,
19 memcached_connection_t type
);
21 static memcached_return_t
update_continuum(memcached_st
*ptr
);
23 static int compare_servers(const void *p1
, const void *p2
)
26 memcached_server_instance_st a
= (memcached_server_instance_st
)p1
;
27 memcached_server_instance_st b
= (memcached_server_instance_st
)p2
;
29 return_value
= strcmp(a
->hostname
, b
->hostname
);
31 if (return_value
== 0)
33 return_value
= (int) (a
->port
- b
->port
);
39 static void sort_hosts(memcached_st
*ptr
)
41 if (memcached_server_count(ptr
))
43 memcached_server_write_instance_st instance
;
45 qsort(memcached_server_list(ptr
), memcached_server_count(ptr
), sizeof(memcached_server_st
), compare_servers
);
46 instance
= memcached_server_instance_fetch(ptr
, 0);
47 instance
->number_of_hosts
= memcached_server_count(ptr
);
52 memcached_return_t
run_distribution(memcached_st
*ptr
)
54 if (ptr
->flags
.use_sort_hosts
)
57 switch (ptr
->distribution
)
59 case MEMCACHED_DISTRIBUTION_CONSISTENT
:
60 case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA
:
61 case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA_SPY
:
62 case MEMCACHED_DISTRIBUTION_CONSISTENT_WEIGHTED
:
63 return update_continuum(ptr
);
64 case MEMCACHED_DISTRIBUTION_VIRTUAL_BUCKET
:
65 case MEMCACHED_DISTRIBUTION_MODULA
:
67 case MEMCACHED_DISTRIBUTION_RANDOM
:
68 srandom((uint32_t) time(NULL
));
70 case MEMCACHED_DISTRIBUTION_CONSISTENT_MAX
:
72 WATCHPOINT_ASSERT(0); /* We have added a distribution without extending the logic */
75 return MEMCACHED_SUCCESS
;
78 static uint32_t ketama_server_hash(const char *key
, size_t key_length
, uint32_t alignment
)
80 unsigned char results
[16];
82 libhashkit_md5_signature((unsigned char*)key
, key_length
, results
);
84 return ((uint32_t) (results
[3 + alignment
* 4] & 0xFF) << 24)
85 | ((uint32_t) (results
[2 + alignment
* 4] & 0xFF) << 16)
86 | ((uint32_t) (results
[1 + alignment
* 4] & 0xFF) << 8)
87 | (results
[0 + alignment
* 4] & 0xFF);
90 static int continuum_item_cmp(const void *t1
, const void *t2
)
92 memcached_continuum_item_st
*ct1
= (memcached_continuum_item_st
*)t1
;
93 memcached_continuum_item_st
*ct2
= (memcached_continuum_item_st
*)t2
;
95 /* Why 153? Hmmm... */
96 WATCHPOINT_ASSERT(ct1
->value
!= 153);
97 if (ct1
->value
== ct2
->value
)
99 else if (ct1
->value
> ct2
->value
)
105 static memcached_return_t
update_continuum(memcached_st
*ptr
)
108 uint32_t continuum_index
= 0;
110 memcached_server_st
*list
;
111 uint32_t pointer_index
;
112 uint32_t pointer_counter
= 0;
113 uint32_t pointer_per_server
= MEMCACHED_POINTS_PER_SERVER
;
114 uint32_t pointer_per_hash
= 1;
115 uint64_t total_weight
= 0;
116 uint64_t is_ketama_weighted
= 0;
117 uint64_t is_auto_ejecting
= 0;
118 uint32_t points_per_server
= 0;
119 uint32_t live_servers
= 0;
122 if (gettimeofday(&now
, NULL
) != 0)
124 memcached_set_errno(ptr
, errno
, NULL
);
125 return MEMCACHED_ERRNO
;
128 list
= memcached_server_list(ptr
);
130 /* count live servers (those without a retry delay set) */
131 is_auto_ejecting
= _is_auto_eject_host(ptr
);
132 if (is_auto_ejecting
)
135 ptr
->ketama
.next_distribution_rebuild
= 0;
136 for (host_index
= 0; host_index
< memcached_server_count(ptr
); ++host_index
)
138 if (list
[host_index
].next_retry
<= now
.tv_sec
)
142 if (ptr
->ketama
.next_distribution_rebuild
== 0 || list
[host_index
].next_retry
< ptr
->ketama
.next_distribution_rebuild
)
143 ptr
->ketama
.next_distribution_rebuild
= list
[host_index
].next_retry
;
149 live_servers
= memcached_server_count(ptr
);
152 is_ketama_weighted
= memcached_behavior_get(ptr
, MEMCACHED_BEHAVIOR_KETAMA_WEIGHTED
);
153 points_per_server
= (uint32_t) (is_ketama_weighted
? MEMCACHED_POINTS_PER_SERVER_KETAMA
: MEMCACHED_POINTS_PER_SERVER
);
155 if (live_servers
== 0)
156 return MEMCACHED_SUCCESS
;
158 if (live_servers
> ptr
->ketama
.continuum_count
)
160 memcached_continuum_item_st
*new_ptr
;
162 new_ptr
= libmemcached_realloc(ptr
, ptr
->ketama
.continuum
,
163 sizeof(memcached_continuum_item_st
) * (live_servers
+ MEMCACHED_CONTINUUM_ADDITION
) * points_per_server
);
166 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
168 ptr
->ketama
.continuum
= new_ptr
;
169 ptr
->ketama
.continuum_count
= live_servers
+ MEMCACHED_CONTINUUM_ADDITION
;
172 if (is_ketama_weighted
)
174 for (host_index
= 0; host_index
< memcached_server_count(ptr
); ++host_index
)
176 if (list
[host_index
].weight
== 0)
178 list
[host_index
].weight
= 1;
180 if (! is_auto_ejecting
|| list
[host_index
].next_retry
<= now
.tv_sec
)
181 total_weight
+= list
[host_index
].weight
;
185 for (host_index
= 0; host_index
< memcached_server_count(ptr
); ++host_index
)
187 if (is_auto_ejecting
&& list
[host_index
].next_retry
> now
.tv_sec
)
190 if (is_ketama_weighted
)
192 float pct
= (float)list
[host_index
].weight
/ (float)total_weight
;
193 pointer_per_server
= (uint32_t) ((floorf((float) (pct
* MEMCACHED_POINTS_PER_SERVER_KETAMA
/ 4 * (float)live_servers
+ 0.0000000001))) * 4);
196 printf("ketama_weighted:%s|%d|%llu|%u\n",
197 list
[host_index
].hostname
,
198 list
[host_index
].port
,
199 (unsigned long long)list
[host_index
].weight
,
205 if (ptr
->distribution
== MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA_SPY
)
207 for (pointer_index
= 0;
208 pointer_index
< pointer_per_server
/ pointer_per_hash
;
211 char sort_host
[MEMCACHED_MAX_HOST_SORT_LENGTH
]= "";
212 int sort_host_length
;
214 // Spymemcached ketema key format is: hostname/ip:port-index
215 // If hostname is not available then: /ip:port-index
216 sort_host_length
= snprintf(sort_host
, MEMCACHED_MAX_HOST_SORT_LENGTH
,
218 list
[host_index
].hostname
,
219 (uint32_t)list
[host_index
].port
,
222 if (sort_host_length
>= MEMCACHED_MAX_HOST_SORT_LENGTH
|| sort_host_length
< 0)
224 return MEMCACHED_FAILURE
;
227 printf("update_continuum: key is %s\n", sort_host
);
230 WATCHPOINT_ASSERT(sort_host_length
);
232 if (is_ketama_weighted
)
234 for (uint32_t x
= 0; x
< pointer_per_hash
; x
++)
236 value
= ketama_server_hash(sort_host
, (size_t)sort_host_length
, x
);
237 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
238 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
243 value
= hashkit_digest(&ptr
->distribution_hashkit
, sort_host
, (size_t)sort_host_length
);
244 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
245 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
251 for (pointer_index
= 1;
252 pointer_index
<= pointer_per_server
/ pointer_per_hash
;
255 char sort_host
[MEMCACHED_MAX_HOST_SORT_LENGTH
]= "";
256 int sort_host_length
;
258 if (list
[host_index
].port
== MEMCACHED_DEFAULT_PORT
)
260 sort_host_length
= snprintf(sort_host
, MEMCACHED_MAX_HOST_SORT_LENGTH
,
262 list
[host_index
].hostname
,
267 sort_host_length
= snprintf(sort_host
, MEMCACHED_MAX_HOST_SORT_LENGTH
,
269 list
[host_index
].hostname
,
270 (uint32_t)list
[host_index
].port
,
274 if (sort_host_length
>= MEMCACHED_MAX_HOST_SORT_LENGTH
|| sort_host_length
< 0)
276 return MEMCACHED_FAILURE
;
279 WATCHPOINT_ASSERT(sort_host_length
);
281 if (is_ketama_weighted
)
283 for (uint32_t x
= 0; x
< pointer_per_hash
; x
++)
285 value
= ketama_server_hash(sort_host
, (size_t)sort_host_length
, x
);
286 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
287 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
292 value
= hashkit_digest(&ptr
->distribution_hashkit
, sort_host
, (size_t)sort_host_length
);
293 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
294 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
299 pointer_counter
+= pointer_per_server
;
302 WATCHPOINT_ASSERT(ptr
);
303 WATCHPOINT_ASSERT(ptr
->continuum
);
304 WATCHPOINT_ASSERT(memcached_server_count(ptr
) * MEMCACHED_POINTS_PER_SERVER
<= MEMCACHED_CONTINUUM_SIZE
);
305 ptr
->ketama
.continuum_points_counter
= pointer_counter
;
306 qsort(ptr
->ketama
.continuum
, ptr
->ketama
.continuum_points_counter
, sizeof(memcached_continuum_item_st
), continuum_item_cmp
);
309 for (pointer_index
= 0; memcached_server_count(ptr
) && pointer_index
< ((live_servers
* MEMCACHED_POINTS_PER_SERVER
) - 1); pointer_index
++)
311 WATCHPOINT_ASSERT(ptr
->continuum
[pointer_index
].value
<= ptr
->continuum
[pointer_index
+ 1].value
);
315 return MEMCACHED_SUCCESS
;
319 memcached_return_t
memcached_server_push(memcached_st
*ptr
, const memcached_server_list_st list
)
322 memcached_server_st
*new_host_list
;
325 return MEMCACHED_SUCCESS
;
327 count
= memcached_server_list_count(list
);
328 new_host_list
= libmemcached_realloc(ptr
, memcached_server_list(ptr
),
329 sizeof(memcached_server_st
) * (count
+ memcached_server_count(ptr
)));
332 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
334 memcached_server_list_set(ptr
, new_host_list
);
336 for (uint32_t x
= 0; x
< count
; x
++)
338 memcached_server_write_instance_st instance
;
340 if ((ptr
->flags
.use_udp
&& list
[x
].type
!= MEMCACHED_CONNECTION_UDP
)
341 || ((list
[x
].type
== MEMCACHED_CONNECTION_UDP
)
342 && ! (ptr
->flags
.use_udp
)) )
344 return MEMCACHED_INVALID_HOST_PROTOCOL
;
347 WATCHPOINT_ASSERT(list
[x
].hostname
[0] != 0);
349 instance
= memcached_server_instance_fetch(ptr
, memcached_server_count(ptr
));
351 /* TODO check return type */
352 (void)memcached_server_create_with(ptr
, instance
, list
[x
].hostname
,
353 list
[x
].port
, list
[x
].weight
, list
[x
].type
);
354 ptr
->number_of_hosts
++;
357 // Provides backwards compatibility with server list.
359 memcached_server_write_instance_st instance
;
360 instance
= memcached_server_instance_fetch(ptr
, 0);
361 instance
->number_of_hosts
= memcached_server_count(ptr
);
364 return run_distribution(ptr
);
367 memcached_return_t
memcached_server_add_unix_socket(memcached_st
*ptr
,
368 const char *filename
)
370 return memcached_server_add_unix_socket_with_weight(ptr
, filename
, 0);
373 memcached_return_t
memcached_server_add_unix_socket_with_weight(memcached_st
*ptr
,
374 const char *filename
,
378 return MEMCACHED_FAILURE
;
380 return server_add(ptr
, filename
, 0, weight
, MEMCACHED_CONNECTION_UNIX_SOCKET
);
383 memcached_return_t
memcached_server_add_udp(memcached_st
*ptr
,
384 const char *hostname
,
387 return memcached_server_add_udp_with_weight(ptr
, hostname
, port
, 0);
390 memcached_return_t
memcached_server_add_udp_with_weight(memcached_st
*ptr
,
391 const char *hostname
,
396 port
= MEMCACHED_DEFAULT_PORT
;
399 hostname
= "localhost";
401 return server_add(ptr
, hostname
, port
, weight
, MEMCACHED_CONNECTION_UDP
);
404 memcached_return_t
memcached_server_add(memcached_st
*ptr
,
405 const char *hostname
,
408 return memcached_server_add_with_weight(ptr
, hostname
, port
, 0);
411 memcached_return_t
memcached_server_add_with_weight(memcached_st
*ptr
,
412 const char *hostname
,
417 port
= MEMCACHED_DEFAULT_PORT
;
420 hostname
= "localhost";
422 return server_add(ptr
, hostname
, port
, weight
, MEMCACHED_CONNECTION_TCP
);
425 static memcached_return_t
server_add(memcached_st
*ptr
, const char *hostname
,
428 memcached_connection_t type
)
430 memcached_server_st
*new_host_list
;
431 memcached_server_write_instance_st instance
;
433 if ( (ptr
->flags
.use_udp
&& type
!= MEMCACHED_CONNECTION_UDP
)
434 || ( (type
== MEMCACHED_CONNECTION_UDP
) && (! ptr
->flags
.use_udp
) ) )
435 return MEMCACHED_INVALID_HOST_PROTOCOL
;
437 new_host_list
= libmemcached_realloc(ptr
, memcached_server_list(ptr
),
438 sizeof(memcached_server_st
) * (ptr
->number_of_hosts
+ 1));
440 if (new_host_list
== NULL
)
441 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
443 memcached_server_list_set(ptr
, new_host_list
);
445 /* TODO: Check return type */
446 instance
= memcached_server_instance_fetch(ptr
, memcached_server_count(ptr
));
447 (void)memcached_server_create_with(ptr
, instance
, hostname
, port
, weight
, type
);
448 ptr
->number_of_hosts
++;
450 instance
= memcached_server_instance_fetch(ptr
, 0);
451 memcached_servers_set_count(instance
, memcached_server_count(ptr
));
453 return run_distribution(ptr
);
456 memcached_return_t
memcached_server_add_parsed(memcached_st
*ptr
,
457 const char *hostname
,
458 size_t hostname_length
,
462 char buffer
[NI_MAXHOST
];
464 memcpy(buffer
, hostname
, hostname_length
);
465 buffer
[hostname_length
]= 0;
467 return server_add(ptr
, buffer
,
470 MEMCACHED_CONNECTION_TCP
);