3 /* Protoypes (static) */
4 static memcached_return
server_add(memcached_st
*ptr
, char *hostname
,
6 memcached_connection type
);
7 memcached_return
update_continuum(memcached_st
*ptr
);
9 #define MEMCACHED_WHEEL_SIZE 1024
10 #define MEMCACHED_STRIDE 4
11 static void rebalance_wheel(memcached_st
*ptr
)
18 memset(ptr
->wheel
, 0, sizeof(unsigned int) * MEMCACHED_WHEEL_SIZE
);
20 for (latch
= y
= x
= 0; x
< MEMCACHED_WHEEL_SIZE
; x
++, latch
++)
22 if (latch
== MEMCACHED_STRIDE
)
25 if (y
== ptr
->number_of_hosts
)
34 static int compare_servers(const void *p1
, const void *p2
)
37 memcached_server_st
*a
= (memcached_server_st
*)p1
;
38 memcached_server_st
*b
= (memcached_server_st
*)p2
;
40 return_value
= strcmp(a
->hostname
, b
->hostname
);
42 if (return_value
== 0)
44 return_value
= (int) (a
->port
- b
->port
);
50 void sort_hosts(memcached_st
*ptr
)
52 if (ptr
->number_of_hosts
)
54 qsort(ptr
->hosts
, ptr
->number_of_hosts
, sizeof(memcached_server_st
), compare_servers
);
55 ptr
->hosts
[0].count
= ptr
->number_of_hosts
;
60 memcached_return
run_distribution(memcached_st
*ptr
)
62 switch (ptr
->distribution
)
64 case MEMCACHED_DISTRIBUTION_CONSISTENT
:
65 case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA
:
66 return update_continuum(ptr
);
67 case MEMCACHED_DISTRIBUTION_CONSISTENT_WHEEL
:
70 case MEMCACHED_DISTRIBUTION_MODULA
:
71 if (ptr
->flags
& MEM_USE_SORT_HOSTS
)
75 WATCHPOINT_ASSERT(0); /* We have added a distribution without extending the logic */
78 return MEMCACHED_SUCCESS
;
81 static void host_reset(memcached_st
*ptr
, memcached_server_st
*host
,
82 char *hostname
, unsigned int port
,
83 memcached_connection type
)
85 memset(host
, 0, sizeof(memcached_server_st
));
86 strncpy(host
->hostname
, hostname
, MEMCACHED_MAX_HOST_LENGTH
- 1);
87 host
->root
= ptr
? ptr
: NULL
;
91 host
->read_ptr
= host
->read_buffer
;
93 host
->next_retry
= ptr
->retry_timeout
;
94 host
->sockaddr_inited
= MEMCACHED_NOT_ALLOCATED
;
97 void server_list_free(memcached_st
*ptr
, memcached_server_st
*servers
)
104 for (x
= 0; x
< servers
->count
; x
++)
105 if (servers
[x
].address_info
)
106 freeaddrinfo(servers
[x
].address_info
);
108 if (ptr
&& ptr
->call_free
)
109 ptr
->call_free(ptr
, servers
);
114 static int continuum_item_cmp(const void *t1
, const void *t2
)
116 memcached_continuum_item_st
*ct1
= (memcached_continuum_item_st
*)t1
;
117 memcached_continuum_item_st
*ct2
= (memcached_continuum_item_st
*)t2
;
119 WATCHPOINT_ASSERT(ct1
->value
!= 153);
120 if (ct1
->value
== ct2
->value
)
122 else if (ct1
->value
> ct2
->value
)
128 static uint32_t internal_generate_ketama_md5(char *key
, size_t key_length
)
130 unsigned char results
[16];
132 md5_signature((unsigned char*)key
, (unsigned int)key_length
, results
);
134 return ( (results
[3] ) << 24)
135 | ( (results
[2] ) << 16)
136 | ( (results
[1] ) << 8)
140 memcached_return
update_continuum(memcached_st
*ptr
)
144 uint32_t continuum_index
= 0;
146 memcached_server_st
*list
;
148 if (ptr
->number_of_hosts
> ptr
->continuum_count
)
150 memcached_continuum_item_st
*new_ptr
;
152 if (ptr
->call_realloc
)
153 new_ptr
= (memcached_continuum_item_st
*)ptr
->call_realloc(ptr
, ptr
->continuum
, sizeof(memcached_continuum_item_st
) * (ptr
->number_of_hosts
+ MEMCACHED_CONTINUUM_ADDITION
) * MEMCACHED_POINTS_PER_SERVER
);
155 new_ptr
= (memcached_continuum_item_st
*)realloc(ptr
->continuum
, sizeof(memcached_continuum_item_st
) * (ptr
->number_of_hosts
+ MEMCACHED_CONTINUUM_ADDITION
) * MEMCACHED_POINTS_PER_SERVER
);
158 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
160 ptr
->continuum
= new_ptr
;
161 ptr
->continuum_count
= ptr
->number_of_hosts
+ MEMCACHED_CONTINUUM_ADDITION
;
165 for (host_index
= 0; host_index
< ptr
->number_of_hosts
; ++host_index
)
167 for(index
= 1; index
<= MEMCACHED_POINTS_PER_SERVER
; ++index
)
169 char sort_host
[MEMCACHED_MAX_HOST_SORT_LENGTH
]= "";
170 size_t sort_host_length
;
172 sort_host_length
= snprintf(sort_host
, MEMCACHED_MAX_HOST_SORT_LENGTH
, "%s:%d-%d",
173 list
[host_index
].hostname
, list
[host_index
].port
, index
);
174 WATCHPOINT_ASSERT(sort_host_length
);
175 value
= internal_generate_ketama_md5(sort_host
, sort_host_length
);
176 ptr
->continuum
[continuum_index
].index
= host_index
;
177 ptr
->continuum
[continuum_index
++].value
= value
;
181 WATCHPOINT_ASSERT(ptr
->number_of_hosts
* MEMCACHED_POINTS_PER_SERVER
<= MEMCACHED_CONTINUUM_SIZE
);
182 qsort(ptr
->continuum
, ptr
->number_of_hosts
* MEMCACHED_POINTS_PER_SERVER
, sizeof(memcached_continuum_item_st
), continuum_item_cmp
);
184 for (index
= 0; index
< ((ptr
->number_of_hosts
* MEMCACHED_POINTS_PER_SERVER
) - 1); index
++)
186 WATCHPOINT_ASSERT(ptr
->continuum
[index
].value
<= ptr
->continuum
[index
+ 1].value
);
190 return MEMCACHED_SUCCESS
;
194 memcached_return
memcached_server_push(memcached_st
*ptr
, memcached_server_st
*list
)
198 memcached_server_st
*new_host_list
;
201 return MEMCACHED_SUCCESS
;
203 count
= list
[0].count
;
205 if (ptr
->call_realloc
)
207 (memcached_server_st
*)ptr
->call_realloc(ptr
, ptr
->hosts
,
208 sizeof(memcached_server_st
) * (count
+ ptr
->number_of_hosts
));
211 (memcached_server_st
*)realloc(ptr
->hosts
,
212 sizeof(memcached_server_st
) * (count
+ ptr
->number_of_hosts
));
215 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
217 ptr
->hosts
= new_host_list
;
219 for (x
= 0; x
< count
; x
++)
221 WATCHPOINT_ASSERT(list
[x
].hostname
[0] != 0);
222 host_reset(ptr
, &ptr
->hosts
[ptr
->number_of_hosts
], list
[x
].hostname
,
223 list
[x
].port
, list
[x
].type
);
224 ptr
->number_of_hosts
++;
226 ptr
->hosts
[0].count
= ptr
->number_of_hosts
;
228 return run_distribution(ptr
);
231 memcached_return
memcached_server_add_unix_socket(memcached_st
*ptr
, char *filename
)
234 return MEMCACHED_FAILURE
;
236 return server_add(ptr
, filename
, 0, MEMCACHED_CONNECTION_UNIX_SOCKET
);
239 memcached_return
memcached_server_add_udp(memcached_st
*ptr
,
244 port
= MEMCACHED_DEFAULT_PORT
;
247 hostname
= "localhost";
249 return server_add(ptr
, hostname
, port
, MEMCACHED_CONNECTION_UDP
);
252 memcached_return
memcached_server_add(memcached_st
*ptr
,
257 port
= MEMCACHED_DEFAULT_PORT
;
260 hostname
= "localhost";
262 return server_add(ptr
, hostname
, port
, MEMCACHED_CONNECTION_TCP
);
265 static memcached_return
server_add(memcached_st
*ptr
, char *hostname
,
267 memcached_connection type
)
269 memcached_server_st
*new_host_list
;
271 if (ptr
->call_realloc
)
272 new_host_list
= (memcached_server_st
*)ptr
->call_realloc(ptr
, ptr
->hosts
,
273 sizeof(memcached_server_st
) * (ptr
->number_of_hosts
+1));
275 new_host_list
= (memcached_server_st
*)realloc(ptr
->hosts
,
276 sizeof(memcached_server_st
) * (ptr
->number_of_hosts
+1));
277 if (new_host_list
== NULL
)
278 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
280 ptr
->hosts
= new_host_list
;
282 host_reset(ptr
, &ptr
->hosts
[ptr
->number_of_hosts
], hostname
, port
, type
);
283 ptr
->number_of_hosts
++;
284 ptr
->hosts
[0].count
= ptr
->number_of_hosts
;
286 return run_distribution(ptr
);
289 memcached_server_st
*memcached_server_list_append(memcached_server_st
*ptr
,
290 char *hostname
, unsigned int port
,
291 memcached_return
*error
)
294 memcached_server_st
*new_host_list
;
296 if (hostname
== NULL
|| error
== NULL
)
300 port
= MEMCACHED_DEFAULT_PORT
;
302 /* Increment count for hosts */
306 count
+= ptr
[0].count
;
309 new_host_list
= (memcached_server_st
*)realloc(ptr
, sizeof(memcached_server_st
) * count
);
312 *error
= MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
316 host_reset(NULL
, &new_host_list
[count
-1], hostname
, port
, MEMCACHED_CONNECTION_TCP
);
318 /* Backwards compatibility hack */
319 new_host_list
[0].count
= count
;
321 *error
= MEMCACHED_SUCCESS
;
322 return new_host_list
;
325 unsigned int memcached_server_list_count(memcached_server_st
*ptr
)
333 void memcached_server_list_free(memcached_server_st
*ptr
)
335 server_list_free(NULL
, ptr
);