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 host_reset(memcached_st
*ptr
, memcached_server_st
*host
,
58 const char *hostname
, unsigned int port
, uint32_t weight
,
59 memcached_connection type
)
61 memset(host
, 0, sizeof(memcached_server_st
));
62 strncpy(host
->hostname
, hostname
, MEMCACHED_MAX_HOST_LENGTH
- 1);
63 host
->root
= ptr
? ptr
: NULL
;
68 host
->read_ptr
= host
->read_buffer
;
70 host
->next_retry
= ptr
->retry_timeout
;
71 host
->sockaddr_inited
= MEMCACHED_NOT_ALLOCATED
;
74 void server_list_free(memcached_st
*ptr
, memcached_server_st
*servers
)
81 for (x
= 0; x
< servers
->count
; x
++)
82 if (servers
[x
].address_info
)
84 freeaddrinfo(servers
[x
].address_info
);
85 servers
[x
].address_info
= NULL
;
88 if (ptr
&& ptr
->call_free
)
89 ptr
->call_free(ptr
, servers
);
94 static uint32_t ketama_server_hash(const char *key
, unsigned int key_length
, int alignment
)
96 unsigned char results
[16];
98 md5_signature((unsigned char*)key
, key_length
, results
);
99 return ((uint32_t) (results
[3 + alignment
* 4] & 0xFF) << 24)
100 | ((uint32_t) (results
[2 + alignment
* 4] & 0xFF) << 16)
101 | ((uint32_t) (results
[1 + alignment
* 4] & 0xFF) << 8)
102 | (results
[0 + alignment
* 4] & 0xFF);
105 static int continuum_item_cmp(const void *t1
, const void *t2
)
107 memcached_continuum_item_st
*ct1
= (memcached_continuum_item_st
*)t1
;
108 memcached_continuum_item_st
*ct2
= (memcached_continuum_item_st
*)t2
;
110 /* Why 153? Hmmm... */
111 WATCHPOINT_ASSERT(ct1
->value
!= 153);
112 if (ct1
->value
== ct2
->value
)
114 else if (ct1
->value
> ct2
->value
)
120 memcached_return
update_continuum(memcached_st
*ptr
)
124 uint32_t continuum_index
= 0;
126 memcached_server_st
*list
;
127 uint32_t pointer_counter
= 0;
128 uint32_t pointer_per_server
= MEMCACHED_POINTS_PER_SERVER
;
129 uint32_t pointer_per_hash
= 1;
130 uint64_t total_weight
= 0;
131 uint32_t is_ketama_weighted
= 0;
132 uint32_t points_per_server
= 0;
134 is_ketama_weighted
= memcached_behavior_get(ptr
, MEMCACHED_BEHAVIOR_KETAMA_WEIGHTED
);
135 points_per_server
= is_ketama_weighted
? MEMCACHED_POINTS_PER_SERVER_KETAMA
: MEMCACHED_POINTS_PER_SERVER
;
137 if (ptr
->number_of_hosts
> ptr
->continuum_count
)
139 memcached_continuum_item_st
*new_ptr
;
141 if (ptr
->call_realloc
)
142 new_ptr
= (memcached_continuum_item_st
*)ptr
->call_realloc(ptr
, ptr
->continuum
, sizeof(memcached_continuum_item_st
) * (ptr
->number_of_hosts
+ MEMCACHED_CONTINUUM_ADDITION
) * points_per_server
);
144 new_ptr
= (memcached_continuum_item_st
*)realloc(ptr
->continuum
, sizeof(memcached_continuum_item_st
) * (ptr
->number_of_hosts
+ MEMCACHED_CONTINUUM_ADDITION
) * points_per_server
);
147 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
149 ptr
->continuum
= new_ptr
;
150 ptr
->continuum_count
= ptr
->number_of_hosts
+ MEMCACHED_CONTINUUM_ADDITION
;
155 if (is_ketama_weighted
)
157 for (host_index
= 0; host_index
< ptr
->number_of_hosts
; ++host_index
)
159 if (list
[host_index
].weight
== 0)
161 list
[host_index
].weight
= 1;
163 total_weight
+= list
[host_index
].weight
;
167 for (host_index
= 0; host_index
< ptr
->number_of_hosts
; ++host_index
)
169 if (is_ketama_weighted
)
171 float pct
= (float)list
[host_index
].weight
/ (float)total_weight
;
172 pointer_per_server
= floorf(pct
* MEMCACHED_POINTS_PER_SERVER_KETAMA
/ 4 * (float)(ptr
->number_of_hosts
) + 0.0000000001) * 4;
175 printf("ketama_weighted:%s|%d|%llu|%u\n",
176 list
[host_index
].hostname
,
177 list
[host_index
].port
,
178 (unsigned long long)list
[host_index
].weight
,
182 for (index
= 1; index
<= pointer_per_server
/ pointer_per_hash
; ++index
)
184 char sort_host
[MEMCACHED_MAX_HOST_SORT_LENGTH
]= "";
185 size_t sort_host_length
;
187 if (list
[host_index
].port
== MEMCACHED_DEFAULT_PORT
)
189 sort_host_length
= snprintf(sort_host
, MEMCACHED_MAX_HOST_SORT_LENGTH
, "%s-%d",
190 list
[host_index
].hostname
, index
- 1);
195 sort_host_length
= snprintf(sort_host
, MEMCACHED_MAX_HOST_SORT_LENGTH
, "%s:%d-%d",
196 list
[host_index
].hostname
, list
[host_index
].port
, index
- 1);
198 WATCHPOINT_ASSERT(sort_host_length
);
200 if (is_ketama_weighted
)
203 for (i
= 0; i
< pointer_per_hash
; i
++)
205 value
= ketama_server_hash(sort_host
, sort_host_length
, i
);
206 ptr
->continuum
[continuum_index
].index
= host_index
;
207 ptr
->continuum
[continuum_index
++].value
= value
;
212 value
= generate_hash_value(sort_host
, sort_host_length
, ptr
->hash_continuum
);
213 ptr
->continuum
[continuum_index
].index
= host_index
;
214 ptr
->continuum
[continuum_index
++].value
= value
;
217 pointer_counter
+= pointer_per_server
;
220 WATCHPOINT_ASSERT(ptr
);
221 WATCHPOINT_ASSERT(ptr
->continuum
);
222 WATCHPOINT_ASSERT(ptr
->number_of_hosts
* MEMCACHED_POINTS_PER_SERVER
<= MEMCACHED_CONTINUUM_SIZE
);
223 ptr
->continuum_points_counter
= pointer_counter
;
224 qsort(ptr
->continuum
, ptr
->continuum_points_counter
, sizeof(memcached_continuum_item_st
), continuum_item_cmp
);
227 for (index
= 0; ptr
->number_of_hosts
&& index
< ((ptr
->number_of_hosts
* MEMCACHED_POINTS_PER_SERVER
) - 1); index
++)
229 WATCHPOINT_ASSERT(ptr
->continuum
[index
].value
<= ptr
->continuum
[index
+ 1].value
);
233 return MEMCACHED_SUCCESS
;
237 memcached_return
memcached_server_push(memcached_st
*ptr
, memcached_server_st
*list
)
241 memcached_server_st
*new_host_list
;
244 return MEMCACHED_SUCCESS
;
246 count
= list
[0].count
;
247 if (ptr
->call_realloc
)
249 (memcached_server_st
*)ptr
->call_realloc(ptr
, ptr
->hosts
,
250 sizeof(memcached_server_st
) * (count
+ ptr
->number_of_hosts
));
253 (memcached_server_st
*)realloc(ptr
->hosts
,
254 sizeof(memcached_server_st
) * (count
+ ptr
->number_of_hosts
));
257 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
259 ptr
->hosts
= new_host_list
;
261 for (x
= 0; x
< count
; x
++)
263 WATCHPOINT_ASSERT(list
[x
].hostname
[0] != 0);
264 host_reset(ptr
, &ptr
->hosts
[ptr
->number_of_hosts
], list
[x
].hostname
,
265 list
[x
].port
, list
[x
].weight
, list
[x
].type
);
266 ptr
->number_of_hosts
++;
268 ptr
->hosts
[0].count
= ptr
->number_of_hosts
;
270 return run_distribution(ptr
);
273 memcached_return
memcached_server_add_unix_socket(memcached_st
*ptr
,
274 const char *filename
)
276 return memcached_server_add_unix_socket_with_weight(ptr
, filename
, 0);
279 memcached_return
memcached_server_add_unix_socket_with_weight(memcached_st
*ptr
,
280 const char *filename
,
284 return MEMCACHED_FAILURE
;
286 return server_add(ptr
, filename
, 0, weight
, MEMCACHED_CONNECTION_UNIX_SOCKET
);
289 memcached_return
memcached_server_add_udp(memcached_st
*ptr
,
290 const char *hostname
,
293 return memcached_server_add_udp_with_weight(ptr
, hostname
, port
, 0);
296 memcached_return
memcached_server_add_udp_with_weight(memcached_st
*ptr
,
297 const char *hostname
,
302 port
= MEMCACHED_DEFAULT_PORT
;
305 hostname
= "localhost";
307 return server_add(ptr
, hostname
, port
, weight
, MEMCACHED_CONNECTION_UDP
);
310 memcached_return
memcached_server_add(memcached_st
*ptr
,
311 const char *hostname
,
314 return memcached_server_add_with_weight(ptr
, hostname
, port
, 0);
317 memcached_return
memcached_server_add_with_weight(memcached_st
*ptr
,
318 const char *hostname
,
323 port
= MEMCACHED_DEFAULT_PORT
;
326 hostname
= "localhost";
328 return server_add(ptr
, hostname
, port
, weight
, MEMCACHED_CONNECTION_TCP
);
331 static memcached_return
server_add(memcached_st
*ptr
, const char *hostname
,
334 memcached_connection type
)
336 memcached_server_st
*new_host_list
;
338 if (ptr
->call_realloc
)
339 new_host_list
= (memcached_server_st
*)ptr
->call_realloc(ptr
, ptr
->hosts
,
340 sizeof(memcached_server_st
) * (ptr
->number_of_hosts
+1));
342 new_host_list
= (memcached_server_st
*)realloc(ptr
->hosts
,
343 sizeof(memcached_server_st
) * (ptr
->number_of_hosts
+1));
344 if (new_host_list
== NULL
)
345 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
347 ptr
->hosts
= new_host_list
;
349 host_reset(ptr
, &ptr
->hosts
[ptr
->number_of_hosts
], hostname
, port
, weight
, type
);
350 ptr
->number_of_hosts
++;
351 ptr
->hosts
[0].count
= ptr
->number_of_hosts
;
353 return run_distribution(ptr
);
356 memcached_return
memcached_server_remove(memcached_server_st
*st_ptr
)
359 memcached_st
*ptr
= st_ptr
->root
;
360 memcached_server_st
*list
= ptr
->hosts
;
362 for (x
= 0, index
= 0; x
< ptr
->number_of_hosts
; x
++)
364 if (strncmp(list
[x
].hostname
, st_ptr
->hostname
, MEMCACHED_MAX_HOST_LENGTH
) != 0 || list
[x
].port
!= st_ptr
->port
)
367 memcpy(list
+index
, list
+x
, sizeof(memcached_server_st
));
371 ptr
->number_of_hosts
= index
;
373 if (st_ptr
->address_info
)
375 freeaddrinfo(st_ptr
->address_info
);
376 st_ptr
->address_info
= NULL
;
378 run_distribution(ptr
);
380 return MEMCACHED_SUCCESS
;
383 memcached_server_st
*memcached_server_list_append(memcached_server_st
*ptr
,
384 const char *hostname
, unsigned int port
,
385 memcached_return
*error
)
387 return memcached_server_list_append_with_weight(ptr
, hostname
, port
, 0, error
);
390 memcached_server_st
*memcached_server_list_append_with_weight(memcached_server_st
*ptr
,
391 const char *hostname
, unsigned int port
,
393 memcached_return
*error
)
396 memcached_server_st
*new_host_list
;
398 if (hostname
== NULL
|| error
== NULL
)
402 port
= MEMCACHED_DEFAULT_PORT
;
404 /* Increment count for hosts */
408 count
+= ptr
[0].count
;
411 new_host_list
= (memcached_server_st
*)realloc(ptr
, sizeof(memcached_server_st
) * count
);
414 *error
= MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
418 host_reset(NULL
, &new_host_list
[count
-1], hostname
, port
, weight
, MEMCACHED_CONNECTION_TCP
);
420 /* Backwards compatibility hack */
421 new_host_list
[0].count
= count
;
423 *error
= MEMCACHED_SUCCESS
;
424 return new_host_list
;
427 unsigned int memcached_server_list_count(memcached_server_st
*ptr
)
435 void memcached_server_list_free(memcached_server_st
*ptr
)
437 server_list_free(NULL
, ptr
);