4 /* Protoypes (static) */
5 static memcached_return
server_add(memcached_st
*ptr
, const char *hostname
,
8 memcached_connection type
);
9 memcached_return
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 (ptr
->number_of_hosts
)
31 qsort(ptr
->hosts
, ptr
->number_of_hosts
, sizeof(memcached_server_st
), compare_servers
);
32 ptr
->hosts
[0].count
= ptr
->number_of_hosts
;
37 memcached_return
run_distribution(memcached_st
*ptr
)
39 switch (ptr
->distribution
)
41 case MEMCACHED_DISTRIBUTION_CONSISTENT
:
42 case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA
:
43 return update_continuum(ptr
);
44 case MEMCACHED_DISTRIBUTION_MODULA
:
45 if (ptr
->flags
& MEM_USE_SORT_HOSTS
)
48 case MEMCACHED_DISTRIBUTION_RANDOM
:
51 WATCHPOINT_ASSERT(0); /* We have added a distribution without extending the logic */
54 return MEMCACHED_SUCCESS
;
57 void server_list_free(memcached_st
*ptr
, memcached_server_st
*servers
)
64 for (x
= 0; x
< servers
->count
; x
++)
65 if (servers
[x
].address_info
)
67 freeaddrinfo(servers
[x
].address_info
);
68 servers
[x
].address_info
= NULL
;
72 ptr
->call_free(ptr
, servers
);
77 static uint32_t ketama_server_hash(const char *key
, unsigned int key_length
, int alignment
)
79 unsigned char results
[16];
81 md5_signature((unsigned char*)key
, key_length
, results
);
82 return ((uint32_t) (results
[3 + alignment
* 4] & 0xFF) << 24)
83 | ((uint32_t) (results
[2 + alignment
* 4] & 0xFF) << 16)
84 | ((uint32_t) (results
[1 + alignment
* 4] & 0xFF) << 8)
85 | (results
[0 + alignment
* 4] & 0xFF);
88 static int continuum_item_cmp(const void *t1
, const void *t2
)
90 memcached_continuum_item_st
*ct1
= (memcached_continuum_item_st
*)t1
;
91 memcached_continuum_item_st
*ct2
= (memcached_continuum_item_st
*)t2
;
93 /* Why 153? Hmmm... */
94 WATCHPOINT_ASSERT(ct1
->value
!= 153);
95 if (ct1
->value
== ct2
->value
)
97 else if (ct1
->value
> ct2
->value
)
103 memcached_return
update_continuum(memcached_st
*ptr
)
107 uint32_t continuum_index
= 0;
109 memcached_server_st
*list
;
110 uint32_t pointer_counter
= 0;
111 uint32_t pointer_per_server
= MEMCACHED_POINTS_PER_SERVER
;
112 uint32_t pointer_per_hash
= 1;
113 uint64_t total_weight
= 0;
114 uint32_t is_ketama_weighted
= 0;
115 uint32_t is_auto_ejecting
= 0;
116 uint32_t points_per_server
= 0;
117 uint32_t live_servers
= 0;
120 if (gettimeofday(&now
, NULL
) != 0)
122 ptr
->cached_errno
= errno
;
123 return MEMCACHED_ERRNO
;
128 /* count live servers (those without a retry delay set) */
129 is_auto_ejecting
= memcached_behavior_get(ptr
, MEMCACHED_BEHAVIOR_AUTO_EJECT_HOSTS
);
130 if (is_auto_ejecting
)
133 ptr
->next_distribution_rebuild
= 0;
134 for (host_index
= 0; host_index
< ptr
->number_of_hosts
; ++host_index
)
136 if (list
[host_index
].next_retry
<= now
.tv_sec
)
140 if (ptr
->next_distribution_rebuild
== 0 || list
[host_index
].next_retry
< ptr
->next_distribution_rebuild
)
141 ptr
->next_distribution_rebuild
= list
[host_index
].next_retry
;
146 live_servers
= ptr
->number_of_hosts
;
148 is_ketama_weighted
= memcached_behavior_get(ptr
, MEMCACHED_BEHAVIOR_KETAMA_WEIGHTED
);
149 points_per_server
= is_ketama_weighted
? MEMCACHED_POINTS_PER_SERVER_KETAMA
: MEMCACHED_POINTS_PER_SERVER
;
151 if (live_servers
== 0)
152 return MEMCACHED_SUCCESS
;
154 if (live_servers
> ptr
->continuum_count
)
156 memcached_continuum_item_st
*new_ptr
;
158 new_ptr
= ptr
->call_realloc(ptr
, ptr
->continuum
,
159 sizeof(memcached_continuum_item_st
) * (live_servers
+ MEMCACHED_CONTINUUM_ADDITION
) * points_per_server
);
162 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
164 ptr
->continuum
= new_ptr
;
165 ptr
->continuum_count
= live_servers
+ MEMCACHED_CONTINUUM_ADDITION
;
168 if (is_ketama_weighted
)
170 for (host_index
= 0; host_index
< ptr
->number_of_hosts
; ++host_index
)
172 if (list
[host_index
].weight
== 0)
174 list
[host_index
].weight
= 1;
176 if (!is_auto_ejecting
|| list
[host_index
].next_retry
<= now
.tv_sec
)
177 total_weight
+= list
[host_index
].weight
;
181 for (host_index
= 0; host_index
< ptr
->number_of_hosts
; ++host_index
)
183 if (is_auto_ejecting
&& list
[host_index
].next_retry
> now
.tv_sec
)
186 if (is_ketama_weighted
)
188 float pct
= (float)list
[host_index
].weight
/ (float)total_weight
;
189 pointer_per_server
= floorf(pct
* MEMCACHED_POINTS_PER_SERVER_KETAMA
/ 4 * (float)live_servers
+ 0.0000000001) * 4;
192 printf("ketama_weighted:%s|%d|%llu|%u\n",
193 list
[host_index
].hostname
,
194 list
[host_index
].port
,
195 (unsigned long long)list
[host_index
].weight
,
199 for (index
= 1; index
<= pointer_per_server
/ pointer_per_hash
; ++index
)
201 char sort_host
[MEMCACHED_MAX_HOST_SORT_LENGTH
]= "";
202 size_t sort_host_length
;
204 if (list
[host_index
].port
== MEMCACHED_DEFAULT_PORT
)
206 sort_host_length
= snprintf(sort_host
, MEMCACHED_MAX_HOST_SORT_LENGTH
, "%s-%d",
207 list
[host_index
].hostname
, index
- 1);
212 sort_host_length
= snprintf(sort_host
, MEMCACHED_MAX_HOST_SORT_LENGTH
, "%s:%d-%d",
213 list
[host_index
].hostname
, list
[host_index
].port
, index
- 1);
215 WATCHPOINT_ASSERT(sort_host_length
);
217 if (is_ketama_weighted
)
220 for (i
= 0; i
< pointer_per_hash
; i
++)
222 value
= ketama_server_hash(sort_host
, sort_host_length
, i
);
223 ptr
->continuum
[continuum_index
].index
= host_index
;
224 ptr
->continuum
[continuum_index
++].value
= value
;
229 value
= memcached_generate_hash_value(sort_host
, sort_host_length
, ptr
->hash_continuum
);
230 ptr
->continuum
[continuum_index
].index
= host_index
;
231 ptr
->continuum
[continuum_index
++].value
= value
;
234 pointer_counter
+= pointer_per_server
;
237 WATCHPOINT_ASSERT(ptr
);
238 WATCHPOINT_ASSERT(ptr
->continuum
);
239 WATCHPOINT_ASSERT(ptr
->number_of_hosts
* MEMCACHED_POINTS_PER_SERVER
<= MEMCACHED_CONTINUUM_SIZE
);
240 ptr
->continuum_points_counter
= pointer_counter
;
241 qsort(ptr
->continuum
, ptr
->continuum_points_counter
, sizeof(memcached_continuum_item_st
), continuum_item_cmp
);
244 for (index
= 0; ptr
->number_of_hosts
&& index
< ((live_servers
* MEMCACHED_POINTS_PER_SERVER
) - 1); index
++)
246 WATCHPOINT_ASSERT(ptr
->continuum
[index
].value
<= ptr
->continuum
[index
+ 1].value
);
250 return MEMCACHED_SUCCESS
;
254 memcached_return
memcached_server_push(memcached_st
*ptr
, memcached_server_st
*list
)
258 memcached_server_st
*new_host_list
;
261 return MEMCACHED_SUCCESS
;
263 count
= list
[0].count
;
264 new_host_list
= ptr
->call_realloc(ptr
, ptr
->hosts
,
265 sizeof(memcached_server_st
) * (count
+ ptr
->number_of_hosts
));
268 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
270 ptr
->hosts
= new_host_list
;
272 for (x
= 0; x
< count
; x
++)
274 if ((ptr
->flags
& MEM_USE_UDP
&& list
[x
].type
!= MEMCACHED_CONNECTION_UDP
)
275 || ((list
[x
].type
== MEMCACHED_CONNECTION_UDP
)
276 && ! (ptr
->flags
& MEM_USE_UDP
)) )
277 return MEMCACHED_INVALID_HOST_PROTOCOL
;
279 WATCHPOINT_ASSERT(list
[x
].hostname
[0] != 0);
280 memcached_server_create(ptr
, &ptr
->hosts
[ptr
->number_of_hosts
]);
281 /* TODO check return type */
282 (void)memcached_server_create_with(ptr
, &ptr
->hosts
[ptr
->number_of_hosts
], list
[x
].hostname
,
283 list
[x
].port
, list
[x
].weight
, list
[x
].type
);
284 ptr
->number_of_hosts
++;
286 ptr
->hosts
[0].count
= ptr
->number_of_hosts
;
288 return run_distribution(ptr
);
291 memcached_return
memcached_server_add_unix_socket(memcached_st
*ptr
,
292 const char *filename
)
294 return memcached_server_add_unix_socket_with_weight(ptr
, filename
, 0);
297 memcached_return
memcached_server_add_unix_socket_with_weight(memcached_st
*ptr
,
298 const char *filename
,
302 return MEMCACHED_FAILURE
;
304 return server_add(ptr
, filename
, 0, weight
, MEMCACHED_CONNECTION_UNIX_SOCKET
);
307 memcached_return
memcached_server_add_udp(memcached_st
*ptr
,
308 const char *hostname
,
311 return memcached_server_add_udp_with_weight(ptr
, hostname
, port
, 0);
314 memcached_return
memcached_server_add_udp_with_weight(memcached_st
*ptr
,
315 const char *hostname
,
320 port
= MEMCACHED_DEFAULT_PORT
;
323 hostname
= "localhost";
325 return server_add(ptr
, hostname
, port
, weight
, MEMCACHED_CONNECTION_UDP
);
328 memcached_return
memcached_server_add(memcached_st
*ptr
,
329 const char *hostname
,
332 return memcached_server_add_with_weight(ptr
, hostname
, port
, 0);
335 memcached_return
memcached_server_add_with_weight(memcached_st
*ptr
,
336 const char *hostname
,
341 port
= MEMCACHED_DEFAULT_PORT
;
344 hostname
= "localhost";
346 return server_add(ptr
, hostname
, port
, weight
, MEMCACHED_CONNECTION_TCP
);
349 static memcached_return
server_add(memcached_st
*ptr
, const char *hostname
,
352 memcached_connection type
)
354 memcached_server_st
*new_host_list
;
356 if ( (ptr
->flags
& MEM_USE_UDP
&& type
!= MEMCACHED_CONNECTION_UDP
)
357 || ( (type
== MEMCACHED_CONNECTION_UDP
) && !(ptr
->flags
& MEM_USE_UDP
) ) )
358 return MEMCACHED_INVALID_HOST_PROTOCOL
;
360 new_host_list
= ptr
->call_realloc(ptr
, ptr
->hosts
,
361 sizeof(memcached_server_st
) * (ptr
->number_of_hosts
+1));
363 if (new_host_list
== NULL
)
364 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
366 ptr
->hosts
= new_host_list
;
368 /* TODO: Check return type */
369 (void)memcached_server_create_with(ptr
, &ptr
->hosts
[ptr
->number_of_hosts
], hostname
, port
, weight
, type
);
370 ptr
->number_of_hosts
++;
371 ptr
->hosts
[0].count
= ptr
->number_of_hosts
;
373 return run_distribution(ptr
);
376 memcached_return
memcached_server_remove(memcached_server_st
*st_ptr
)
379 memcached_st
*ptr
= st_ptr
->root
;
380 memcached_server_st
*list
= ptr
->hosts
;
382 for (x
= 0, index
= 0; x
< ptr
->number_of_hosts
; x
++)
384 if (strncmp(list
[x
].hostname
, st_ptr
->hostname
, MEMCACHED_MAX_HOST_LENGTH
) != 0 || list
[x
].port
!= st_ptr
->port
)
387 memcpy(list
+index
, list
+x
, sizeof(memcached_server_st
));
391 ptr
->number_of_hosts
= index
;
393 if (st_ptr
->address_info
)
395 freeaddrinfo(st_ptr
->address_info
);
396 st_ptr
->address_info
= NULL
;
398 run_distribution(ptr
);
400 return MEMCACHED_SUCCESS
;
403 memcached_server_st
*memcached_server_list_append(memcached_server_st
*ptr
,
404 const char *hostname
, unsigned int port
,
405 memcached_return
*error
)
407 return memcached_server_list_append_with_weight(ptr
, hostname
, port
, 0, error
);
410 memcached_server_st
*memcached_server_list_append_with_weight(memcached_server_st
*ptr
,
411 const char *hostname
, unsigned int port
,
413 memcached_return
*error
)
416 memcached_server_st
*new_host_list
;
418 if (hostname
== NULL
|| error
== NULL
)
422 port
= MEMCACHED_DEFAULT_PORT
;
424 /* Increment count for hosts */
428 count
+= ptr
[0].count
;
431 new_host_list
= (memcached_server_st
*)realloc(ptr
, sizeof(memcached_server_st
) * count
);
434 *error
= MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
438 /* TODO: Check return type */
439 memcached_server_create_with(NULL
, &new_host_list
[count
-1], hostname
, port
, weight
, MEMCACHED_CONNECTION_TCP
);
441 /* Backwards compatibility hack */
442 new_host_list
[0].count
= count
;
444 *error
= MEMCACHED_SUCCESS
;
445 return new_host_list
;
448 unsigned int memcached_server_list_count(memcached_server_st
*ptr
)
456 void memcached_server_list_free(memcached_server_st
*ptr
)
458 server_list_free(NULL
, ptr
);