1 /* vim:expandtab:shiftwidth=2:tabstop=2:smarttab:
5 * Copyright (C) 2011 Data Differential, http://datadifferential.com/
6 * Copyright (C) 2006-2010 Brian Aker All rights reserved.
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions are
12 * * Redistributions of source code must retain the above copyright
13 * notice, this list of conditions and the following disclaimer.
15 * * Redistributions in binary form must reproduce the above
16 * copyright notice, this list of conditions and the following disclaimer
17 * in the documentation and/or other materials provided with the
20 * * The names of its contributors may not be used to endorse or
21 * promote products derived from this software without specific prior
24 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
25 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
26 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
27 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
28 * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
29 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
30 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
31 * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
32 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
33 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
34 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
38 #include <libmemcached/common.h>
43 /* Protoypes (static) */
44 static memcached_return_t
server_add(memcached_st
*ptr
, const char *hostname
,
47 memcached_connection_t type
);
49 static memcached_return_t
update_continuum(memcached_st
*ptr
);
51 static int compare_servers(const void *p1
, const void *p2
)
54 memcached_server_instance_st a
= (memcached_server_instance_st
)p1
;
55 memcached_server_instance_st b
= (memcached_server_instance_st
)p2
;
57 return_value
= strcmp(a
->hostname
, b
->hostname
);
59 if (return_value
== 0)
61 return_value
= (int) (a
->port
- b
->port
);
67 static void sort_hosts(memcached_st
*ptr
)
69 if (memcached_server_count(ptr
))
71 memcached_server_write_instance_st instance
;
73 qsort(memcached_server_list(ptr
), memcached_server_count(ptr
), sizeof(memcached_server_st
), compare_servers
);
74 instance
= memcached_server_instance_fetch(ptr
, 0);
75 instance
->number_of_hosts
= memcached_server_count(ptr
);
80 memcached_return_t
run_distribution(memcached_st
*ptr
)
82 if (ptr
->flags
.use_sort_hosts
)
85 switch (ptr
->distribution
)
87 case MEMCACHED_DISTRIBUTION_CONSISTENT
:
88 case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA
:
89 case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA_SPY
:
90 case MEMCACHED_DISTRIBUTION_CONSISTENT_WEIGHTED
:
91 return update_continuum(ptr
);
92 case MEMCACHED_DISTRIBUTION_VIRTUAL_BUCKET
:
93 case MEMCACHED_DISTRIBUTION_MODULA
:
95 case MEMCACHED_DISTRIBUTION_RANDOM
:
96 srandom((uint32_t) time(NULL
));
98 case MEMCACHED_DISTRIBUTION_CONSISTENT_MAX
:
100 WATCHPOINT_ASSERT(0); /* We have added a distribution without extending the logic */
103 return MEMCACHED_SUCCESS
;
106 static uint32_t ketama_server_hash(const char *key
, size_t key_length
, uint32_t alignment
)
108 unsigned char results
[16];
110 libhashkit_md5_signature((unsigned char*)key
, key_length
, results
);
112 return ((uint32_t) (results
[3 + alignment
* 4] & 0xFF) << 24)
113 | ((uint32_t) (results
[2 + alignment
* 4] & 0xFF) << 16)
114 | ((uint32_t) (results
[1 + alignment
* 4] & 0xFF) << 8)
115 | (results
[0 + alignment
* 4] & 0xFF);
118 static int continuum_item_cmp(const void *t1
, const void *t2
)
120 memcached_continuum_item_st
*ct1
= (memcached_continuum_item_st
*)t1
;
121 memcached_continuum_item_st
*ct2
= (memcached_continuum_item_st
*)t2
;
123 /* Why 153? Hmmm... */
124 WATCHPOINT_ASSERT(ct1
->value
!= 153);
125 if (ct1
->value
== ct2
->value
)
127 else if (ct1
->value
> ct2
->value
)
133 static memcached_return_t
update_continuum(memcached_st
*ptr
)
135 uint32_t continuum_index
= 0;
136 memcached_server_st
*list
;
137 uint32_t pointer_counter
= 0;
138 uint32_t pointer_per_server
= MEMCACHED_POINTS_PER_SERVER
;
139 uint32_t pointer_per_hash
= 1;
140 uint32_t live_servers
= 0;
143 if (gettimeofday(&now
, NULL
) != 0)
145 memcached_set_errno(ptr
, errno
, NULL
);
146 return MEMCACHED_ERRNO
;
149 list
= memcached_server_list(ptr
);
151 /* count live servers (those without a retry delay set) */
152 bool is_auto_ejecting
= _is_auto_eject_host(ptr
);
153 if (is_auto_ejecting
)
156 ptr
->ketama
.next_distribution_rebuild
= 0;
157 for (uint32_t host_index
= 0; host_index
< memcached_server_count(ptr
); ++host_index
)
159 if (list
[host_index
].next_retry
<= now
.tv_sec
)
163 if (ptr
->ketama
.next_distribution_rebuild
== 0 || list
[host_index
].next_retry
< ptr
->ketama
.next_distribution_rebuild
)
164 ptr
->ketama
.next_distribution_rebuild
= list
[host_index
].next_retry
;
170 live_servers
= memcached_server_count(ptr
);
173 uint64_t is_ketama_weighted
= memcached_behavior_get(ptr
, MEMCACHED_BEHAVIOR_KETAMA_WEIGHTED
);
174 uint32_t points_per_server
= (uint32_t) (is_ketama_weighted
? MEMCACHED_POINTS_PER_SERVER_KETAMA
: MEMCACHED_POINTS_PER_SERVER
);
176 if (live_servers
== 0)
177 return MEMCACHED_SUCCESS
;
179 if (live_servers
> ptr
->ketama
.continuum_count
)
181 memcached_continuum_item_st
*new_ptr
;
183 new_ptr
= static_cast<memcached_continuum_item_st
*>(libmemcached_realloc(ptr
, ptr
->ketama
.continuum
,
184 sizeof(memcached_continuum_item_st
) * (live_servers
+ MEMCACHED_CONTINUUM_ADDITION
) * points_per_server
));
187 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
189 ptr
->ketama
.continuum
= new_ptr
;
190 ptr
->ketama
.continuum_count
= live_servers
+ MEMCACHED_CONTINUUM_ADDITION
;
193 uint64_t total_weight
= 0;
194 if (is_ketama_weighted
)
196 for (uint32_t host_index
= 0; host_index
< memcached_server_count(ptr
); ++host_index
)
198 if (! is_auto_ejecting
|| list
[host_index
].next_retry
<= now
.tv_sec
)
200 total_weight
+= list
[host_index
].weight
;
205 for (uint32_t host_index
= 0; host_index
< memcached_server_count(ptr
); ++host_index
)
207 if (is_auto_ejecting
&& list
[host_index
].next_retry
> now
.tv_sec
)
210 if (is_ketama_weighted
)
212 float pct
= (float)list
[host_index
].weight
/ (float)total_weight
;
213 pointer_per_server
= (uint32_t) ((floorf((float) (pct
* MEMCACHED_POINTS_PER_SERVER_KETAMA
/ 4 * (float)live_servers
+ 0.0000000001))) * 4);
216 printf("ketama_weighted:%s|%d|%llu|%u\n",
217 list
[host_index
].hostname
,
218 list
[host_index
].port
,
219 (unsigned long long)list
[host_index
].weight
,
225 if (ptr
->distribution
== MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA_SPY
)
227 for (uint32_t pointer_index
= 0;
228 pointer_index
< pointer_per_server
/ pointer_per_hash
;
231 char sort_host
[MEMCACHED_MAX_HOST_SORT_LENGTH
]= "";
232 int sort_host_length
;
234 // Spymemcached ketema key format is: hostname/ip:port-index
235 // If hostname is not available then: /ip:port-index
236 sort_host_length
= snprintf(sort_host
, MEMCACHED_MAX_HOST_SORT_LENGTH
,
238 list
[host_index
].hostname
,
239 (uint32_t)list
[host_index
].port
,
242 if (sort_host_length
>= MEMCACHED_MAX_HOST_SORT_LENGTH
|| sort_host_length
< 0)
244 return MEMCACHED_FAILURE
;
247 printf("update_continuum: key is %s\n", sort_host
);
250 WATCHPOINT_ASSERT(sort_host_length
);
252 if (is_ketama_weighted
)
254 for (uint32_t x
= 0; x
< pointer_per_hash
; x
++)
256 uint32_t value
= ketama_server_hash(sort_host
, (size_t)sort_host_length
, x
);
257 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
258 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
263 uint32_t value
= hashkit_digest(&ptr
->hashkit
, sort_host
, (size_t)sort_host_length
);
264 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
265 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
271 for (uint32_t pointer_index
= 1;
272 pointer_index
<= pointer_per_server
/ pointer_per_hash
;
275 char sort_host
[MEMCACHED_MAX_HOST_SORT_LENGTH
]= "";
276 int sort_host_length
;
278 if (list
[host_index
].port
== MEMCACHED_DEFAULT_PORT
)
280 sort_host_length
= snprintf(sort_host
, MEMCACHED_MAX_HOST_SORT_LENGTH
,
282 list
[host_index
].hostname
,
287 sort_host_length
= snprintf(sort_host
, MEMCACHED_MAX_HOST_SORT_LENGTH
,
289 list
[host_index
].hostname
,
290 (uint32_t)list
[host_index
].port
,
294 if (sort_host_length
>= MEMCACHED_MAX_HOST_SORT_LENGTH
|| sort_host_length
< 0)
296 return MEMCACHED_FAILURE
;
299 WATCHPOINT_ASSERT(sort_host_length
);
301 if (is_ketama_weighted
)
303 for (uint32_t x
= 0; x
< pointer_per_hash
; x
++)
305 uint32_t value
= ketama_server_hash(sort_host
, (size_t)sort_host_length
, x
);
306 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
307 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
312 uint32_t value
= hashkit_digest(&ptr
->hashkit
, sort_host
, (size_t)sort_host_length
);
313 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
314 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
319 pointer_counter
+= pointer_per_server
;
322 WATCHPOINT_ASSERT(ptr
);
323 WATCHPOINT_ASSERT(ptr
->continuum
);
324 WATCHPOINT_ASSERT(memcached_server_count(ptr
) * MEMCACHED_POINTS_PER_SERVER
<= MEMCACHED_CONTINUUM_SIZE
);
325 ptr
->ketama
.continuum_points_counter
= pointer_counter
;
326 qsort(ptr
->ketama
.continuum
, ptr
->ketama
.continuum_points_counter
, sizeof(memcached_continuum_item_st
), continuum_item_cmp
);
329 for (uint32_t pointer_index
= 0; memcached_server_count(ptr
) && pointer_index
< ((live_servers
* MEMCACHED_POINTS_PER_SERVER
) - 1); pointer_index
++)
331 WATCHPOINT_ASSERT(ptr
->continuum
[pointer_index
].value
<= ptr
->continuum
[pointer_index
+ 1].value
);
335 return MEMCACHED_SUCCESS
;
339 memcached_return_t
memcached_server_push(memcached_st
*ptr
, const memcached_server_list_st list
)
342 return MEMCACHED_SUCCESS
;
344 uint32_t count
= memcached_server_list_count(list
);
346 memcached_server_st
*new_host_list
;
347 new_host_list
= static_cast<memcached_server_st
*>(libmemcached_realloc(ptr
, memcached_server_list(ptr
),
348 sizeof(memcached_server_st
) * (count
+ memcached_server_count(ptr
))));
350 if (not new_host_list
)
351 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
353 memcached_server_list_set(ptr
, new_host_list
);
355 for (uint32_t x
= 0; x
< count
; x
++)
357 memcached_server_write_instance_st instance
;
359 if ((ptr
->flags
.use_udp
&& list
[x
].type
!= MEMCACHED_CONNECTION_UDP
)
360 || ((list
[x
].type
== MEMCACHED_CONNECTION_UDP
)
361 && ! (ptr
->flags
.use_udp
)) )
363 return MEMCACHED_INVALID_HOST_PROTOCOL
;
366 WATCHPOINT_ASSERT(list
[x
].hostname
[0] != 0);
368 instance
= memcached_server_instance_fetch(ptr
, memcached_server_count(ptr
));
369 WATCHPOINT_ASSERT(instance
);
371 /* TODO check return type */
372 instance
= memcached_server_create_with(ptr
, instance
, list
[x
].hostname
,
373 list
[x
].port
, list
[x
].weight
, list
[x
].type
);
376 return memcached_set_error(ptr
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
);
379 if (list
[x
].weight
> 1)
381 ptr
->ketama
.weighted
= true;
384 ptr
->number_of_hosts
++;
387 // Provides backwards compatibility with server list.
389 memcached_server_write_instance_st instance
;
390 instance
= memcached_server_instance_fetch(ptr
, 0);
391 instance
->number_of_hosts
= memcached_server_count(ptr
);
394 return run_distribution(ptr
);
397 memcached_return_t
memcached_server_add_unix_socket(memcached_st
*ptr
,
398 const char *filename
)
400 return memcached_server_add_unix_socket_with_weight(ptr
, filename
, 0);
403 memcached_return_t
memcached_server_add_unix_socket_with_weight(memcached_st
*ptr
,
404 const char *filename
,
408 return MEMCACHED_FAILURE
;
410 return server_add(ptr
, filename
, 0, weight
, MEMCACHED_CONNECTION_UNIX_SOCKET
);
413 memcached_return_t
memcached_server_add_udp(memcached_st
*ptr
,
414 const char *hostname
,
417 return memcached_server_add_udp_with_weight(ptr
, hostname
, port
, 0);
420 memcached_return_t
memcached_server_add_udp_with_weight(memcached_st
*ptr
,
421 const char *hostname
,
426 port
= MEMCACHED_DEFAULT_PORT
;
429 hostname
= "localhost";
431 return server_add(ptr
, hostname
, port
, weight
, MEMCACHED_CONNECTION_UDP
);
434 memcached_return_t
memcached_server_add(memcached_st
*ptr
,
435 const char *hostname
,
438 return memcached_server_add_with_weight(ptr
, hostname
, port
, 0);
441 memcached_return_t
memcached_server_add_with_weight(memcached_st
*ptr
,
442 const char *hostname
,
447 port
= MEMCACHED_DEFAULT_PORT
;
450 hostname
= "localhost";
452 return server_add(ptr
, hostname
, port
, weight
, MEMCACHED_CONNECTION_TCP
);
455 static memcached_return_t
server_add(memcached_st
*ptr
, const char *hostname
,
458 memcached_connection_t type
)
460 memcached_server_st
*new_host_list
;
461 memcached_server_write_instance_st instance
;
463 if ( (ptr
->flags
.use_udp
&& type
!= MEMCACHED_CONNECTION_UDP
)
464 || ( (type
== MEMCACHED_CONNECTION_UDP
) && (! ptr
->flags
.use_udp
) ) )
465 return MEMCACHED_INVALID_HOST_PROTOCOL
;
467 new_host_list
= static_cast<memcached_server_st
*>(libmemcached_realloc(ptr
, memcached_server_list(ptr
),
468 sizeof(memcached_server_st
) * (ptr
->number_of_hosts
+ 1)));
470 if (new_host_list
== NULL
)
471 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
473 memcached_server_list_set(ptr
, new_host_list
);
475 /* TODO: Check return type */
476 instance
= memcached_server_instance_fetch(ptr
, memcached_server_count(ptr
));
478 (void)memcached_server_create_with(ptr
, instance
, hostname
, port
, weight
, type
);
482 ptr
->ketama
.weighted
= true;
485 ptr
->number_of_hosts
++;
487 instance
= memcached_server_instance_fetch(ptr
, 0);
488 memcached_servers_set_count(instance
, memcached_server_count(ptr
));
490 return run_distribution(ptr
);