03ea81e37eb05fc9a71dc6a0d196719fb901a1fd
[awesomized/libmemcached] / libmemcached / memcached_hosts.c
1 #include "common.h"
2
3 /* Protoypes (static) */
4 static memcached_return server_add(memcached_st *ptr, char *hostname,
5 unsigned int port,
6 memcached_connection type);
7 memcached_return update_continuum(memcached_st *ptr);
8
9 #define MEMCACHED_WHEEL_SIZE 1024
10 #define MEMCACHED_STRIDE 4
11 static memcached_return rebalance_wheel(memcached_st *ptr)
12 {
13 unsigned int x;
14 unsigned int y;
15 unsigned int latch;
16
17 if (ptr->number_of_hosts > ptr->wheel_count)
18 {
19 uint32_t *new_ptr;
20
21 if (ptr->call_realloc)
22 new_ptr= (uint32_t *)ptr->call_realloc(ptr, ptr->wheel, sizeof(uint32_t) * (ptr->number_of_hosts + MEMCACHED_CONTINUUM_ADDITION) * MEMCACHED_STRIDE);
23 else
24 new_ptr= (uint32_t *)realloc(ptr->wheel, sizeof(uint32_t) * (ptr->number_of_hosts + MEMCACHED_CONTINUUM_ADDITION) * MEMCACHED_STRIDE);
25
26 if (new_ptr == 0)
27 return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
28
29 ptr->wheel= new_ptr;
30 ptr->wheel_count= ptr->number_of_hosts + MEMCACHED_CONTINUUM_ADDITION;
31 }
32
33 /* Seed the Wheel */
34 memset(ptr->wheel, 0, sizeof(uint32_t) * MEMCACHED_STRIDE * ptr->wheel_count);
35
36 for (latch= y= x= 0; x < MEMCACHED_STRIDE * ptr->wheel_count; x++, latch++)
37 {
38 if (latch == MEMCACHED_STRIDE)
39 {
40 y++;
41 if (y == ptr->number_of_hosts)
42 y= 0;
43 latch= 0;
44 }
45
46 ptr->wheel[x]= y;
47 }
48
49 return MEMCACHED_SUCCESS;
50 }
51
52 static int compare_servers(const void *p1, const void *p2)
53 {
54 int return_value;
55 memcached_server_st *a= (memcached_server_st *)p1;
56 memcached_server_st *b= (memcached_server_st *)p2;
57
58 return_value= strcmp(a->hostname, b->hostname);
59
60 if (return_value == 0)
61 {
62 return_value= (int) (a->port - b->port);
63 }
64
65 return return_value;
66 }
67
68 void sort_hosts(memcached_st *ptr)
69 {
70 if (ptr->number_of_hosts)
71 {
72 qsort(ptr->hosts, ptr->number_of_hosts, sizeof(memcached_server_st), compare_servers);
73 ptr->hosts[0].count= ptr->number_of_hosts;
74 }
75 }
76
77
78 memcached_return run_distribution(memcached_st *ptr)
79 {
80 switch (ptr->distribution)
81 {
82 case MEMCACHED_DISTRIBUTION_CONSISTENT:
83 case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA:
84 return update_continuum(ptr);
85 case MEMCACHED_DISTRIBUTION_CONSISTENT_WHEEL:
86 return rebalance_wheel(ptr);
87 break;
88 case MEMCACHED_DISTRIBUTION_MODULA:
89 if (ptr->flags & MEM_USE_SORT_HOSTS)
90 sort_hosts(ptr);
91 break;
92 default:
93 WATCHPOINT_ASSERT(0); /* We have added a distribution without extending the logic */
94 }
95
96 return MEMCACHED_SUCCESS;
97 }
98
99 static void host_reset(memcached_st *ptr, memcached_server_st *host,
100 char *hostname, unsigned int port,
101 memcached_connection type)
102 {
103 memset(host, 0, sizeof(memcached_server_st));
104 strncpy(host->hostname, hostname, MEMCACHED_MAX_HOST_LENGTH - 1);
105 host->root= ptr ? ptr : NULL;
106 host->port= port;
107 host->fd= -1;
108 host->type= type;
109 host->read_ptr= host->read_buffer;
110 if (ptr)
111 host->next_retry= ptr->retry_timeout;
112 host->sockaddr_inited= MEMCACHED_NOT_ALLOCATED;
113 }
114
115 void server_list_free(memcached_st *ptr, memcached_server_st *servers)
116 {
117 unsigned int x;
118
119 if (servers == NULL)
120 return;
121
122 for (x= 0; x < servers->count; x++)
123 if (servers[x].address_info)
124 freeaddrinfo(servers[x].address_info);
125
126 if (ptr && ptr->call_free)
127 ptr->call_free(ptr, servers);
128 else
129 free(servers);
130 }
131
132 static int continuum_item_cmp(const void *t1, const void *t2)
133 {
134 memcached_continuum_item_st *ct1 = (memcached_continuum_item_st *)t1;
135 memcached_continuum_item_st *ct2 = (memcached_continuum_item_st *)t2;
136
137 WATCHPOINT_ASSERT(ct1->value != 153);
138 if (ct1->value == ct2->value)
139 return 0;
140 else if (ct1->value > ct2->value)
141 return 1;
142 else
143 return -1;
144 }
145
146 static uint32_t internal_generate_ketama_md5(char *key, size_t key_length)
147 {
148 unsigned char results[16];
149
150 md5_signature((unsigned char*)key, (unsigned int)key_length, results);
151
152 return ( (results[3] ) << 24)
153 | ( (results[2] ) << 16)
154 | ( (results[1] ) << 8)
155 | ( results[0] );
156 }
157
158 memcached_return update_continuum(memcached_st *ptr)
159 {
160 uint32_t index;
161 uint32_t host_index;
162 uint32_t continuum_index= 0;
163 uint32_t value;
164 memcached_server_st *list;
165
166 if (ptr->number_of_hosts > ptr->continuum_count)
167 {
168 memcached_continuum_item_st *new_ptr;
169
170 if (ptr->call_realloc)
171 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);
172 else
173 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);
174
175 if (new_ptr == 0)
176 return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
177
178 ptr->continuum= new_ptr;
179 ptr->continuum_count= ptr->number_of_hosts + MEMCACHED_CONTINUUM_ADDITION;
180 }
181
182 list = ptr->hosts;
183 for (host_index = 0; host_index < ptr->number_of_hosts; ++host_index)
184 {
185 for(index= 1; index <= MEMCACHED_POINTS_PER_SERVER; ++index)
186 {
187 char sort_host[MEMCACHED_MAX_HOST_SORT_LENGTH]= "";
188 size_t sort_host_length;
189
190 sort_host_length= snprintf(sort_host, MEMCACHED_MAX_HOST_SORT_LENGTH, "%s:%d-%d",
191 list[host_index].hostname, list[host_index].port, index);
192 WATCHPOINT_ASSERT(sort_host_length);
193 value= internal_generate_ketama_md5(sort_host, sort_host_length);
194 ptr->continuum[continuum_index].index= host_index;
195 ptr->continuum[continuum_index++].value= value;
196 }
197 }
198
199 WATCHPOINT_ASSERT(ptr->number_of_hosts * MEMCACHED_POINTS_PER_SERVER <= MEMCACHED_CONTINUUM_SIZE);
200 qsort(ptr->continuum, ptr->number_of_hosts * MEMCACHED_POINTS_PER_SERVER, sizeof(memcached_continuum_item_st), continuum_item_cmp);
201 #ifdef HAVE_DEBUG
202 for (index= 0; index < ((ptr->number_of_hosts * MEMCACHED_POINTS_PER_SERVER) - 1); index++)
203 {
204 WATCHPOINT_ASSERT(ptr->continuum[index].value <= ptr->continuum[index + 1].value);
205 }
206 #endif
207
208 return MEMCACHED_SUCCESS;
209 }
210
211
212 memcached_return memcached_server_push(memcached_st *ptr, memcached_server_st *list)
213 {
214 unsigned int x;
215 uint16_t count;
216 memcached_server_st *new_host_list;
217
218 if (!list)
219 return MEMCACHED_SUCCESS;
220
221 count= list[0].count;
222
223 if (ptr->call_realloc)
224 new_host_list=
225 (memcached_server_st *)ptr->call_realloc(ptr, ptr->hosts,
226 sizeof(memcached_server_st) * (count + ptr->number_of_hosts));
227 else
228 new_host_list=
229 (memcached_server_st *)realloc(ptr->hosts,
230 sizeof(memcached_server_st) * (count + ptr->number_of_hosts));
231
232 if (!new_host_list)
233 return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
234
235 ptr->hosts= new_host_list;
236
237 for (x= 0; x < count; x++)
238 {
239 WATCHPOINT_ASSERT(list[x].hostname[0] != 0);
240 host_reset(ptr, &ptr->hosts[ptr->number_of_hosts], list[x].hostname,
241 list[x].port, list[x].type);
242 ptr->number_of_hosts++;
243 }
244 ptr->hosts[0].count= ptr->number_of_hosts;
245
246 return run_distribution(ptr);
247 }
248
249 memcached_return memcached_server_add_unix_socket(memcached_st *ptr, char *filename)
250 {
251 if (!filename)
252 return MEMCACHED_FAILURE;
253
254 return server_add(ptr, filename, 0, MEMCACHED_CONNECTION_UNIX_SOCKET);
255 }
256
257 memcached_return memcached_server_add_udp(memcached_st *ptr,
258 char *hostname,
259 unsigned int port)
260 {
261 if (!port)
262 port= MEMCACHED_DEFAULT_PORT;
263
264 if (!hostname)
265 hostname= "localhost";
266
267 return server_add(ptr, hostname, port, MEMCACHED_CONNECTION_UDP);
268 }
269
270 memcached_return memcached_server_add(memcached_st *ptr,
271 char *hostname,
272 unsigned int port)
273 {
274 if (!port)
275 port= MEMCACHED_DEFAULT_PORT;
276
277 if (!hostname)
278 hostname= "localhost";
279
280 return server_add(ptr, hostname, port, MEMCACHED_CONNECTION_TCP);
281 }
282
283 static memcached_return server_add(memcached_st *ptr, char *hostname,
284 unsigned int port,
285 memcached_connection type)
286 {
287 memcached_server_st *new_host_list;
288
289 if (ptr->call_realloc)
290 new_host_list= (memcached_server_st *)ptr->call_realloc(ptr, ptr->hosts,
291 sizeof(memcached_server_st) * (ptr->number_of_hosts+1));
292 else
293 new_host_list= (memcached_server_st *)realloc(ptr->hosts,
294 sizeof(memcached_server_st) * (ptr->number_of_hosts+1));
295 if (new_host_list == NULL)
296 return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
297
298 ptr->hosts= new_host_list;
299
300 host_reset(ptr, &ptr->hosts[ptr->number_of_hosts], hostname, port, type);
301 ptr->number_of_hosts++;
302 ptr->hosts[0].count= ptr->number_of_hosts;
303
304 return run_distribution(ptr);
305 }
306
307 memcached_server_st *memcached_server_list_append(memcached_server_st *ptr,
308 char *hostname, unsigned int port,
309 memcached_return *error)
310 {
311 unsigned int count;
312 memcached_server_st *new_host_list;
313
314 if (hostname == NULL || error == NULL)
315 return NULL;
316
317 if (!port)
318 port= MEMCACHED_DEFAULT_PORT;
319
320 /* Increment count for hosts */
321 count= 1;
322 if (ptr != NULL)
323 {
324 count+= ptr[0].count;
325 }
326
327 new_host_list= (memcached_server_st *)realloc(ptr, sizeof(memcached_server_st) * count);
328 if (!new_host_list)
329 {
330 *error= MEMCACHED_MEMORY_ALLOCATION_FAILURE;
331 return NULL;
332 }
333
334 host_reset(NULL, &new_host_list[count-1], hostname, port, MEMCACHED_CONNECTION_TCP);
335
336 /* Backwards compatibility hack */
337 new_host_list[0].count= count;
338
339 *error= MEMCACHED_SUCCESS;
340 return new_host_list;
341 }
342
343 unsigned int memcached_server_list_count(memcached_server_st *ptr)
344 {
345 if (ptr == NULL)
346 return 0;
347
348 return ptr[0].count;
349 }
350
351 void memcached_server_list_free(memcached_server_st *ptr)
352 {
353 server_list_free(NULL, ptr);
354 }