4 /* Protoypes (static) */
5 static memcached_return_t
server_add(memcached_st
*ptr
, const char *hostname
,
8 memcached_connection_t type
);
9 static memcached_return_t
update_continuum(memcached_st
*ptr
);
11 static int compare_servers(const void *p1
, const void *p2
)
14 memcached_server_st
*a
= (memcached_server_st
*)p1
;
15 memcached_server_st
*b
= (memcached_server_st
*)p2
;
17 return_value
= strcmp(a
->hostname
, b
->hostname
);
19 if (return_value
== 0)
21 return_value
= (int) (a
->port
- b
->port
);
27 static void sort_hosts(memcached_st
*ptr
)
29 if (memcached_server_count(ptr
))
31 qsort(ptr
->hosts
, memcached_server_count(ptr
), sizeof(memcached_server_st
), compare_servers
);
32 ptr
->hosts
[0].number_of_hosts
= memcached_server_count(ptr
);
37 memcached_return_t
run_distribution(memcached_st
*ptr
)
39 if (ptr
->flags
.use_sort_hosts
)
42 switch (ptr
->distribution
)
44 case MEMCACHED_DISTRIBUTION_CONSISTENT
:
45 case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA
:
46 case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA_SPY
:
47 return update_continuum(ptr
);
48 case MEMCACHED_DISTRIBUTION_MODULA
:
50 case MEMCACHED_DISTRIBUTION_RANDOM
:
51 srandom((uint32_t) time(NULL
));
53 case MEMCACHED_DISTRIBUTION_CONSISTENT_MAX
:
55 WATCHPOINT_ASSERT(0); /* We have added a distribution without extending the logic */
58 ptr
->last_disconnected_server
= NULL
;
60 return MEMCACHED_SUCCESS
;
63 static uint32_t ketama_server_hash(const char *key
, unsigned int key_length
, int alignment
)
65 unsigned char results
[16];
67 md5_signature((unsigned char*)key
, key_length
, results
);
68 return ((uint32_t) (results
[3 + alignment
* 4] & 0xFF) << 24)
69 | ((uint32_t) (results
[2 + alignment
* 4] & 0xFF) << 16)
70 | ((uint32_t) (results
[1 + alignment
* 4] & 0xFF) << 8)
71 | (results
[0 + alignment
* 4] & 0xFF);
74 static int continuum_item_cmp(const void *t1
, const void *t2
)
76 memcached_continuum_item_st
*ct1
= (memcached_continuum_item_st
*)t1
;
77 memcached_continuum_item_st
*ct2
= (memcached_continuum_item_st
*)t2
;
79 /* Why 153? Hmmm... */
80 WATCHPOINT_ASSERT(ct1
->value
!= 153);
81 if (ct1
->value
== ct2
->value
)
83 else if (ct1
->value
> ct2
->value
)
89 static memcached_return_t
update_continuum(memcached_st
*ptr
)
92 uint32_t continuum_index
= 0;
94 memcached_server_st
*list
;
95 uint32_t pointer_index
;
96 uint32_t pointer_counter
= 0;
97 uint32_t pointer_per_server
= MEMCACHED_POINTS_PER_SERVER
;
98 uint32_t pointer_per_hash
= 1;
99 uint64_t total_weight
= 0;
100 uint64_t is_ketama_weighted
= 0;
101 uint64_t is_auto_ejecting
= 0;
102 uint32_t points_per_server
= 0;
103 uint32_t live_servers
= 0;
106 if (gettimeofday(&now
, NULL
) != 0)
108 ptr
->cached_errno
= errno
;
109 return MEMCACHED_ERRNO
;
114 /* count live servers (those without a retry delay set) */
115 is_auto_ejecting
= memcached_behavior_get(ptr
, MEMCACHED_BEHAVIOR_AUTO_EJECT_HOSTS
);
116 if (is_auto_ejecting
)
119 ptr
->next_distribution_rebuild
= 0;
120 for (host_index
= 0; host_index
< memcached_server_count(ptr
); ++host_index
)
122 if (list
[host_index
].next_retry
<= now
.tv_sec
)
126 if (ptr
->next_distribution_rebuild
== 0 || list
[host_index
].next_retry
< ptr
->next_distribution_rebuild
)
127 ptr
->next_distribution_rebuild
= list
[host_index
].next_retry
;
132 live_servers
= memcached_server_count(ptr
);
134 is_ketama_weighted
= memcached_behavior_get(ptr
, MEMCACHED_BEHAVIOR_KETAMA_WEIGHTED
);
135 points_per_server
= (uint32_t) (is_ketama_weighted
? MEMCACHED_POINTS_PER_SERVER_KETAMA
: MEMCACHED_POINTS_PER_SERVER
);
137 if (live_servers
== 0)
138 return MEMCACHED_SUCCESS
;
140 if (live_servers
> ptr
->continuum_count
)
142 memcached_continuum_item_st
*new_ptr
;
144 new_ptr
= ptr
->call_realloc(ptr
, ptr
->continuum
,
145 sizeof(memcached_continuum_item_st
) * (live_servers
+ MEMCACHED_CONTINUUM_ADDITION
) * points_per_server
);
148 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
150 ptr
->continuum
= new_ptr
;
151 ptr
->continuum_count
= live_servers
+ MEMCACHED_CONTINUUM_ADDITION
;
154 if (is_ketama_weighted
)
156 for (host_index
= 0; host_index
< memcached_server_count(ptr
); ++host_index
)
158 if (list
[host_index
].weight
== 0)
160 list
[host_index
].weight
= 1;
162 if (!is_auto_ejecting
|| list
[host_index
].next_retry
<= now
.tv_sec
)
163 total_weight
+= list
[host_index
].weight
;
167 for (host_index
= 0; host_index
< memcached_server_count(ptr
); ++host_index
)
169 if (is_auto_ejecting
&& list
[host_index
].next_retry
> now
.tv_sec
)
172 if (is_ketama_weighted
)
174 float pct
= (float)list
[host_index
].weight
/ (float)total_weight
;
175 pointer_per_server
= (uint32_t) ((floorf((float) (pct
* MEMCACHED_POINTS_PER_SERVER_KETAMA
/ 4 * (float)live_servers
+ 0.0000000001))) * 4);
178 printf("ketama_weighted:%s|%d|%llu|%u\n",
179 list
[host_index
].hostname
,
180 list
[host_index
].port
,
181 (unsigned long long)list
[host_index
].weight
,
187 if (ptr
->distribution
== MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA_SPY
)
189 for (pointer_index
= 0;
190 pointer_index
< pointer_per_server
/ pointer_per_hash
;
193 char sort_host
[MEMCACHED_MAX_HOST_SORT_LENGTH
]= "";
194 size_t sort_host_length
;
196 // Spymemcached ketema key format is: hostname/ip:port-index
197 // If hostname is not available then: /ip:port-index
198 sort_host_length
= (size_t) snprintf(sort_host
, MEMCACHED_MAX_HOST_SORT_LENGTH
,
200 list
[host_index
].hostname
,
201 (uint32_t)list
[host_index
].port
,
204 printf("update_continuum: key is %s\n", sort_host
);
207 WATCHPOINT_ASSERT(sort_host_length
);
209 if (is_ketama_weighted
)
212 for (i
= 0; i
< pointer_per_hash
; i
++)
214 value
= ketama_server_hash(sort_host
, (uint32_t) sort_host_length
, (int) i
);
215 ptr
->continuum
[continuum_index
].index
= host_index
;
216 ptr
->continuum
[continuum_index
++].value
= value
;
221 value
= memcached_generate_hash_value(sort_host
, sort_host_length
, ptr
->distribution_hash
);
222 ptr
->continuum
[continuum_index
].index
= host_index
;
223 ptr
->continuum
[continuum_index
++].value
= value
;
229 for (pointer_index
= 1;
230 pointer_index
<= pointer_per_server
/ pointer_per_hash
;
233 char sort_host
[MEMCACHED_MAX_HOST_SORT_LENGTH
]= "";
234 size_t sort_host_length
;
236 if (list
[host_index
].port
== MEMCACHED_DEFAULT_PORT
)
238 sort_host_length
= (size_t) snprintf(sort_host
, MEMCACHED_MAX_HOST_SORT_LENGTH
,
240 list
[host_index
].hostname
,
245 sort_host_length
= (size_t) snprintf(sort_host
, MEMCACHED_MAX_HOST_SORT_LENGTH
,
247 list
[host_index
].hostname
,
248 (uint32_t)list
[host_index
].port
,
252 WATCHPOINT_ASSERT(sort_host_length
);
254 if (is_ketama_weighted
)
257 for (i
= 0; i
< pointer_per_hash
; i
++)
259 value
= ketama_server_hash(sort_host
, (uint32_t) sort_host_length
, (int) i
);
260 ptr
->continuum
[continuum_index
].index
= host_index
;
261 ptr
->continuum
[continuum_index
++].value
= value
;
266 value
= memcached_generate_hash_value(sort_host
, sort_host_length
, ptr
->distribution_hash
);
267 ptr
->continuum
[continuum_index
].index
= host_index
;
268 ptr
->continuum
[continuum_index
++].value
= value
;
273 pointer_counter
+= pointer_per_server
;
276 WATCHPOINT_ASSERT(ptr
);
277 WATCHPOINT_ASSERT(ptr
->continuum
);
278 WATCHPOINT_ASSERT(memcached_server_count(ptr
) * MEMCACHED_POINTS_PER_SERVER
<= MEMCACHED_CONTINUUM_SIZE
);
279 ptr
->continuum_points_counter
= pointer_counter
;
280 qsort(ptr
->continuum
, ptr
->continuum_points_counter
, sizeof(memcached_continuum_item_st
), continuum_item_cmp
);
283 for (pointer_index
= 0; memcached_server_count(ptr
) && pointer_index
< ((live_servers
* MEMCACHED_POINTS_PER_SERVER
) - 1); pointer_index
++)
285 WATCHPOINT_ASSERT(ptr
->continuum
[pointer_index
].value
<= ptr
->continuum
[pointer_index
+ 1].value
);
289 return MEMCACHED_SUCCESS
;
293 memcached_return_t
memcached_server_push(memcached_st
*ptr
, memcached_server_st
*list
)
297 memcached_server_st
*new_host_list
;
300 return MEMCACHED_SUCCESS
;
302 count
= memcached_servers_count(list
);
303 new_host_list
= ptr
->call_realloc(ptr
, ptr
->hosts
,
304 sizeof(memcached_server_st
) * (count
+ memcached_server_count(ptr
)));
307 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
309 ptr
->hosts
= new_host_list
;
311 for (x
= 0; x
< count
; x
++)
313 if ((ptr
->flags
.use_udp
&& list
[x
].type
!= MEMCACHED_CONNECTION_UDP
)
314 || ((list
[x
].type
== MEMCACHED_CONNECTION_UDP
)
315 && ! (ptr
->flags
.use_udp
)) )
316 return MEMCACHED_INVALID_HOST_PROTOCOL
;
318 WATCHPOINT_ASSERT(list
[x
].hostname
[0] != 0);
319 memcached_server_create(ptr
, &ptr
->hosts
[memcached_server_count(ptr
)]);
320 /* TODO check return type */
321 (void)memcached_server_create_with(ptr
, &ptr
->hosts
[memcached_server_count(ptr
)], list
[x
].hostname
,
322 list
[x
].port
, list
[x
].weight
, list
[x
].type
);
323 ptr
->number_of_hosts
++;
325 ptr
->hosts
[0].number_of_hosts
= memcached_server_count(ptr
);
327 return run_distribution(ptr
);
330 memcached_return_t
memcached_server_add_unix_socket(memcached_st
*ptr
,
331 const char *filename
)
333 return memcached_server_add_unix_socket_with_weight(ptr
, filename
, 0);
336 memcached_return_t
memcached_server_add_unix_socket_with_weight(memcached_st
*ptr
,
337 const char *filename
,
341 return MEMCACHED_FAILURE
;
343 return server_add(ptr
, filename
, 0, weight
, MEMCACHED_CONNECTION_UNIX_SOCKET
);
346 memcached_return_t
memcached_server_add_udp(memcached_st
*ptr
,
347 const char *hostname
,
350 return memcached_server_add_udp_with_weight(ptr
, hostname
, port
, 0);
353 memcached_return_t
memcached_server_add_udp_with_weight(memcached_st
*ptr
,
354 const char *hostname
,
359 port
= MEMCACHED_DEFAULT_PORT
;
362 hostname
= "localhost";
364 return server_add(ptr
, hostname
, port
, weight
, MEMCACHED_CONNECTION_UDP
);
367 memcached_return_t
memcached_server_add(memcached_st
*ptr
,
368 const char *hostname
,
371 return memcached_server_add_with_weight(ptr
, hostname
, port
, 0);
374 memcached_return_t
memcached_server_add_with_weight(memcached_st
*ptr
,
375 const char *hostname
,
380 port
= MEMCACHED_DEFAULT_PORT
;
383 hostname
= "localhost";
385 return server_add(ptr
, hostname
, port
, weight
, MEMCACHED_CONNECTION_TCP
);
388 static memcached_return_t
server_add(memcached_st
*ptr
, const char *hostname
,
391 memcached_connection_t type
)
393 memcached_server_st
*new_host_list
;
395 if ( (ptr
->flags
.use_udp
&& type
!= MEMCACHED_CONNECTION_UDP
)
396 || ( (type
== MEMCACHED_CONNECTION_UDP
) && (! ptr
->flags
.use_udp
) ) )
397 return MEMCACHED_INVALID_HOST_PROTOCOL
;
399 new_host_list
= ptr
->call_realloc(ptr
, ptr
->hosts
,
400 sizeof(memcached_server_st
) * (ptr
->number_of_hosts
+ 1));
402 if (new_host_list
== NULL
)
403 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
405 ptr
->hosts
= new_host_list
;
407 /* TODO: Check return type */
408 (void)memcached_server_create_with(ptr
, &ptr
->hosts
[memcached_server_count(ptr
)], hostname
, port
, weight
, type
);
409 ptr
->number_of_hosts
++;
411 memcached_servers_set_count(&ptr
->hosts
[0], memcached_server_count(ptr
));
413 return run_distribution(ptr
);
416 memcached_return_t
memcached_server_remove(memcached_server_st
*st_ptr
)
418 uint32_t x
, host_index
;
419 memcached_st
*ptr
= st_ptr
->root
;
420 memcached_server_st
*list
= ptr
->hosts
;
422 for (x
= 0, host_index
= 0; x
< memcached_server_count(ptr
); x
++)
424 if (strncmp(list
[x
].hostname
, st_ptr
->hostname
, MEMCACHED_MAX_HOST_LENGTH
) != 0 || list
[x
].port
!= st_ptr
->port
)
427 memcpy(list
+host_index
, list
+x
, sizeof(memcached_server_st
));
431 ptr
->number_of_hosts
= host_index
;
433 if (st_ptr
->address_info
)
435 freeaddrinfo(st_ptr
->address_info
);
436 st_ptr
->address_info
= NULL
;
438 run_distribution(ptr
);
440 return MEMCACHED_SUCCESS
;
443 memcached_server_st
*memcached_server_list_append(memcached_server_st
*ptr
,
444 const char *hostname
, in_port_t port
,
445 memcached_return_t
*error
)
447 return memcached_server_list_append_with_weight(ptr
, hostname
, port
, 0, error
);
450 memcached_server_st
*memcached_server_list_append_with_weight(memcached_server_st
*ptr
,
451 const char *hostname
, in_port_t port
,
453 memcached_return_t
*error
)
456 memcached_server_st
*new_host_list
;
458 if (hostname
== NULL
|| error
== NULL
)
462 port
= MEMCACHED_DEFAULT_PORT
;
464 /* Increment count for hosts */
468 count
+= memcached_servers_count(ptr
);
471 new_host_list
= (memcached_server_st
*)realloc(ptr
, sizeof(memcached_server_st
) * count
);
474 *error
= MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
478 /* TODO: Check return type */
479 memcached_server_create_with(NULL
, &new_host_list
[count
-1], hostname
, port
, weight
, MEMCACHED_CONNECTION_TCP
);
481 /* Backwards compatibility hack */
482 memcached_servers_set_count(new_host_list
, count
);
484 *error
= MEMCACHED_SUCCESS
;
485 return new_host_list
;