18e37d7ba8c450a46509c00771f2c02e902814a4
[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 memcached_return update_continuum(memcached_st *ptr)
147 {
148 uint32_t index;
149 uint32_t host_index;
150 uint32_t continuum_index= 0;
151 uint32_t value;
152 memcached_server_st *list;
153
154 if (ptr->number_of_hosts > ptr->continuum_count)
155 {
156 memcached_continuum_item_st *new_ptr;
157
158 if (ptr->call_realloc)
159 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);
160 else
161 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);
162
163 if (new_ptr == 0)
164 return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
165
166 ptr->continuum= new_ptr;
167 ptr->continuum_count= ptr->number_of_hosts + MEMCACHED_CONTINUUM_ADDITION;
168 }
169
170 list = ptr->hosts;
171 for (host_index = 0; host_index < ptr->number_of_hosts; ++host_index)
172 {
173 for(index= 1; index <= MEMCACHED_POINTS_PER_SERVER; ++index)
174 {
175 char sort_host[MEMCACHED_MAX_HOST_SORT_LENGTH]= "";
176 size_t sort_host_length;
177
178 sort_host_length= snprintf(sort_host, MEMCACHED_MAX_HOST_SORT_LENGTH, "%s:%d-%d",
179 list[host_index].hostname, list[host_index].port, index);
180 WATCHPOINT_ASSERT(sort_host_length);
181 value= generate_hash(ptr, sort_host, sort_host_length);
182 ptr->continuum[continuum_index].index= host_index;
183 ptr->continuum[continuum_index++].value= value;
184 }
185 }
186
187 WATCHPOINT_ASSERT(ptr->number_of_hosts * MEMCACHED_POINTS_PER_SERVER <= MEMCACHED_CONTINUUM_SIZE);
188 qsort(ptr->continuum, ptr->number_of_hosts * MEMCACHED_POINTS_PER_SERVER, sizeof(memcached_continuum_item_st), continuum_item_cmp);
189 #ifdef HAVE_DEBUG
190 for (index= 0; index < ((ptr->number_of_hosts * MEMCACHED_POINTS_PER_SERVER) - 1); index++)
191 {
192 WATCHPOINT_ASSERT(ptr->continuum[index].value <= ptr->continuum[index + 1].value);
193 }
194 #endif
195
196 return MEMCACHED_SUCCESS;
197 }
198
199
200 memcached_return memcached_server_push(memcached_st *ptr, memcached_server_st *list)
201 {
202 unsigned int x;
203 uint16_t count;
204 memcached_server_st *new_host_list;
205
206 if (!list)
207 return MEMCACHED_SUCCESS;
208
209 count= list[0].count;
210
211 if (ptr->call_realloc)
212 new_host_list=
213 (memcached_server_st *)ptr->call_realloc(ptr, ptr->hosts,
214 sizeof(memcached_server_st) * (count + ptr->number_of_hosts));
215 else
216 new_host_list=
217 (memcached_server_st *)realloc(ptr->hosts,
218 sizeof(memcached_server_st) * (count + ptr->number_of_hosts));
219
220 if (!new_host_list)
221 return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
222
223 ptr->hosts= new_host_list;
224
225 for (x= 0; x < count; x++)
226 {
227 WATCHPOINT_ASSERT(list[x].hostname[0] != 0);
228 host_reset(ptr, &ptr->hosts[ptr->number_of_hosts], list[x].hostname,
229 list[x].port, list[x].type);
230 ptr->number_of_hosts++;
231 }
232 ptr->hosts[0].count= ptr->number_of_hosts;
233
234 return run_distribution(ptr);
235 }
236
237 memcached_return memcached_server_add_unix_socket(memcached_st *ptr, char *filename)
238 {
239 if (!filename)
240 return MEMCACHED_FAILURE;
241
242 return server_add(ptr, filename, 0, MEMCACHED_CONNECTION_UNIX_SOCKET);
243 }
244
245 memcached_return memcached_server_add_udp(memcached_st *ptr,
246 char *hostname,
247 unsigned int port)
248 {
249 if (!port)
250 port= MEMCACHED_DEFAULT_PORT;
251
252 if (!hostname)
253 hostname= "localhost";
254
255 return server_add(ptr, hostname, port, MEMCACHED_CONNECTION_UDP);
256 }
257
258 memcached_return memcached_server_add(memcached_st *ptr,
259 char *hostname,
260 unsigned int port)
261 {
262 if (!port)
263 port= MEMCACHED_DEFAULT_PORT;
264
265 if (!hostname)
266 hostname= "localhost";
267
268 return server_add(ptr, hostname, port, MEMCACHED_CONNECTION_TCP);
269 }
270
271 static memcached_return server_add(memcached_st *ptr, char *hostname,
272 unsigned int port,
273 memcached_connection type)
274 {
275 memcached_server_st *new_host_list;
276
277 if (ptr->call_realloc)
278 new_host_list= (memcached_server_st *)ptr->call_realloc(ptr, ptr->hosts,
279 sizeof(memcached_server_st) * (ptr->number_of_hosts+1));
280 else
281 new_host_list= (memcached_server_st *)realloc(ptr->hosts,
282 sizeof(memcached_server_st) * (ptr->number_of_hosts+1));
283 if (new_host_list == NULL)
284 return MEMCACHED_MEMORY_ALLOCATION_FAILURE;
285
286 ptr->hosts= new_host_list;
287
288 host_reset(ptr, &ptr->hosts[ptr->number_of_hosts], hostname, port, type);
289 ptr->number_of_hosts++;
290 ptr->hosts[0].count= ptr->number_of_hosts;
291
292 return run_distribution(ptr);
293 }
294
295 memcached_server_st *memcached_server_list_append(memcached_server_st *ptr,
296 char *hostname, unsigned int port,
297 memcached_return *error)
298 {
299 unsigned int count;
300 memcached_server_st *new_host_list;
301
302 if (hostname == NULL || error == NULL)
303 return NULL;
304
305 if (!port)
306 port= MEMCACHED_DEFAULT_PORT;
307
308 /* Increment count for hosts */
309 count= 1;
310 if (ptr != NULL)
311 {
312 count+= ptr[0].count;
313 }
314
315 new_host_list= (memcached_server_st *)realloc(ptr, sizeof(memcached_server_st) * count);
316 if (!new_host_list)
317 {
318 *error= MEMCACHED_MEMORY_ALLOCATION_FAILURE;
319 return NULL;
320 }
321
322 host_reset(NULL, &new_host_list[count-1], hostname, port, MEMCACHED_CONNECTION_TCP);
323
324 /* Backwards compatibility hack */
325 new_host_list[0].count= count;
326
327 *error= MEMCACHED_SUCCESS;
328 return new_host_list;
329 }
330
331 unsigned int memcached_server_list_count(memcached_server_st *ptr)
332 {
333 if (ptr == NULL)
334 return 0;
335
336 return ptr[0].count;
337 }
338
339 void memcached_server_list_free(memcached_server_st *ptr)
340 {
341 server_list_free(NULL, ptr);
342 }