2 +--------------------------------------------------------------------+
3 | libmemcached-awesome - C/C++ Client Library for memcached |
4 +--------------------------------------------------------------------+
5 | Redistribution and use in source and binary forms, with or without |
6 | modification, are permitted under the terms of the BSD license. |
7 | You should have received a copy of the license in a bundled file |
8 | named LICENSE; in case you did not receive a copy you can review |
9 | the terms online at: https://opensource.org/licenses/BSD-3-Clause |
10 +--------------------------------------------------------------------+
11 | Copyright (c) 2006-2014 Brian Aker https://datadifferential.com/ |
12 | Copyright (c) 2020-2021 Michael Wallner https://awesome.co/ |
13 +--------------------------------------------------------------------+
16 #include "libmemcached/common.h"
17 #include "libmemcached/assert.hpp"
18 #include "p9y/gettimeofday.hpp"
19 #include "p9y/random.hpp"
23 /* Protoypes (static) */
24 static memcached_return_t
update_continuum(Memcached
*ptr
);
26 static int compare_servers(const void *p1
, const void *p2
) {
27 const memcached_instance_st
*a
= (const memcached_instance_st
*) p1
;
28 const memcached_instance_st
*b
= (const memcached_instance_st
*) p2
;
30 int return_value
= strcmp(a
->_hostname
, b
->_hostname
);
32 if (return_value
== 0) {
33 return_value
= int(a
->port() - b
->port());
39 static void sort_hosts(Memcached
*ptr
) {
40 if (memcached_server_count(ptr
)) {
41 qsort(memcached_instance_list(ptr
), memcached_server_count(ptr
), sizeof(memcached_instance_st
),
46 memcached_return_t
run_distribution(Memcached
*ptr
) {
47 if (ptr
->flags
.use_sort_hosts
) {
51 switch (ptr
->distribution
) {
52 case MEMCACHED_DISTRIBUTION_CONSISTENT
:
53 case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA
:
54 case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA_SPY
:
55 case MEMCACHED_DISTRIBUTION_CONSISTENT_WEIGHTED
:
56 return update_continuum(ptr
);
58 case MEMCACHED_DISTRIBUTION_VIRTUAL_BUCKET
:
59 case MEMCACHED_DISTRIBUTION_MODULA
:
62 case MEMCACHED_DISTRIBUTION_RANDOM
:
63 srandom((uint32_t) time(NULL
));
66 case MEMCACHED_DISTRIBUTION_CONSISTENT_MAX
:
68 assert_msg(0, "Invalid distribution type passed to run_distribution()");
71 return MEMCACHED_SUCCESS
;
74 static uint32_t ketama_server_hash(const char *key
, size_t key_length
, uint32_t alignment
) {
75 unsigned char results
[16];
77 libhashkit_md5_signature((unsigned char *) key
, key_length
, results
);
79 return ((uint32_t)(results
[3 + alignment
* 4] & 0xFF) << 24)
80 | ((uint32_t)(results
[2 + alignment
* 4] & 0xFF) << 16)
81 | ((uint32_t)(results
[1 + alignment
* 4] & 0xFF) << 8) | (results
[0 + alignment
* 4] & 0xFF);
84 static int continuum_item_cmp(const void *t1
, const void *t2
) {
85 memcached_continuum_item_st
*ct1
= (memcached_continuum_item_st
*) t1
;
86 memcached_continuum_item_st
*ct2
= (memcached_continuum_item_st
*) t2
;
88 /* Why 153? Hmmm... */
89 WATCHPOINT_ASSERT(ct1
->value
!= 153);
90 if (ct1
->value
== ct2
->value
) {
91 if (ct1
->index
== ct2
->index
) {
93 } else if (ct1
->index
> ct2
->index
) {
98 } else if (ct1
->value
> ct2
->value
) {
105 static memcached_return_t
update_continuum(Memcached
*ptr
) {
106 uint32_t continuum_index
= 0;
107 uint32_t pointer_counter
= 0;
108 uint32_t pointer_per_server
= MEMCACHED_POINTS_PER_SERVER
;
109 uint32_t pointer_per_hash
= 1;
110 uint32_t live_servers
= 0;
113 if (gettimeofday(&now
, NULL
)) {
114 return memcached_set_errno(*ptr
, errno
, MEMCACHED_AT
);
117 memcached_instance_st
*list
= memcached_instance_list(ptr
);
119 /* count live servers (those without a retry delay set) */
120 bool is_auto_ejecting
= _is_auto_eject_host(ptr
);
121 if (is_auto_ejecting
) {
123 ptr
->ketama
.next_distribution_rebuild
= 0;
124 for (uint32_t host_index
= 0; host_index
< memcached_server_count(ptr
); ++host_index
) {
125 if (list
[host_index
].next_retry
<= now
.tv_sec
) {
128 if (ptr
->ketama
.next_distribution_rebuild
== 0
129 or list
[host_index
].next_retry
< ptr
->ketama
.next_distribution_rebuild
)
131 ptr
->ketama
.next_distribution_rebuild
= list
[host_index
].next_retry
;
136 live_servers
= memcached_server_count(ptr
);
139 if (live_servers
== 0) {
140 return MEMCACHED_SUCCESS
;
143 uint32_t points_per_server
=
144 (uint32_t)(memcached_is_weighted_ketama(ptr
) ? MEMCACHED_POINTS_PER_SERVER_KETAMA
145 : MEMCACHED_POINTS_PER_SERVER
);
146 uint32_t continuum_limit
= live_servers
* points_per_server
;
147 uint32_t continuum_extra
= MEMCACHED_CONTINUUM_ADDITION
* points_per_server
;
149 if (continuum_limit
> ptr
->ketama
.continuum_count
) {
150 memcached_continuum_item_st
*new_ptr
;
152 new_ptr
= libmemcached_xrealloc(ptr
, ptr
->ketama
.continuum
, continuum_limit
+ continuum_extra
,
153 memcached_continuum_item_st
);
156 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
159 ptr
->ketama
.continuum
= new_ptr
;
160 ptr
->ketama
.continuum_count
= continuum_limit
+ continuum_extra
;
162 assert_msg(ptr
->ketama
.continuum
, "Programmer Error, empty ketama continuum");
164 uint64_t total_weight
= 0;
165 if (memcached_is_weighted_ketama(ptr
)) {
166 for (uint32_t host_index
= 0; host_index
< memcached_server_count(ptr
); ++host_index
) {
167 if (is_auto_ejecting
== false or list
[host_index
].next_retry
<= now
.tv_sec
) {
168 total_weight
+= list
[host_index
].weight
;
173 for (uint32_t host_index
= 0; host_index
< memcached_server_count(ptr
); ++host_index
) {
174 if (is_auto_ejecting
and list
[host_index
].next_retry
> now
.tv_sec
) {
178 if (memcached_is_weighted_ketama(ptr
)) {
179 float pct
= (float) list
[host_index
].weight
/ (float) total_weight
;
180 pointer_per_server
= (uint32_t)(
181 (::floor((float) (pct
* MEMCACHED_POINTS_PER_SERVER_KETAMA
/ 4 * (float) live_servers
184 pointer_per_hash
= 4;
186 printf("ketama_weighted:%s|%d|%llu|%u\n", list
[host_index
]._hostname
,
187 list
[host_index
].port(), (unsigned long long) list
[host_index
].weight
,
192 if (ptr
->distribution
== MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA_SPY
) {
193 for (uint32_t pointer_index
= 0; pointer_index
< pointer_per_server
/ pointer_per_hash
;
196 char sort_host
[1 + MEMCACHED_NI_MAXHOST
+ 1 + MEMCACHED_NI_MAXSERV
+ 1
197 + MEMCACHED_NI_MAXSERV
] = "";
198 int sort_host_length
;
200 // Spymemcached ketema key format is: hostname/ip:port-index
201 // If hostname is not available then: /ip:port-index
203 snprintf(sort_host
, sizeof(sort_host
), "/%s:%u-%u", list
[host_index
]._hostname
,
204 (uint32_t) list
[host_index
].port(), pointer_index
);
206 if (size_t(sort_host_length
) >= sizeof(sort_host
) or sort_host_length
< 0) {
207 return memcached_set_error(*ptr
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
,
208 memcached_literal_param("snprintf(sizeof(sort_host))"));
212 fprintf(stdout
, "update_continuum: key is %s\n", sort_host
);
215 if (memcached_is_weighted_ketama(ptr
)) {
216 for (uint32_t x
= 0; x
< pointer_per_hash
; x
++) {
217 uint32_t value
= ketama_server_hash(sort_host
, (size_t) sort_host_length
, x
);
218 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
219 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
222 uint32_t value
= hashkit_digest(&ptr
->hashkit
, sort_host
, (size_t) sort_host_length
);
223 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
224 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
228 for (uint32_t pointer_index
= 1; pointer_index
<= pointer_per_server
/ pointer_per_hash
;
231 char sort_host
[MEMCACHED_NI_MAXHOST
+ 1 + MEMCACHED_NI_MAXSERV
+ 1 + MEMCACHED_NI_MAXSERV
] =
233 int sort_host_length
;
235 if (list
[host_index
].port() == MEMCACHED_DEFAULT_PORT
) {
236 sort_host_length
= snprintf(sort_host
, sizeof(sort_host
), "%s-%u",
237 list
[host_index
]._hostname
, pointer_index
- 1);
240 snprintf(sort_host
, sizeof(sort_host
), "%s:%u-%u", list
[host_index
]._hostname
,
241 (uint32_t) list
[host_index
].port(), pointer_index
- 1);
244 if (size_t(sort_host_length
) >= sizeof(sort_host
) or sort_host_length
< 0) {
245 return memcached_set_error(*ptr
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
,
246 memcached_literal_param("snprintf(sizeof(sort_host)))"));
249 if (memcached_is_weighted_ketama(ptr
)) {
250 for (uint32_t x
= 0; x
< pointer_per_hash
; x
++) {
251 uint32_t value
= ketama_server_hash(sort_host
, (size_t) sort_host_length
, x
);
252 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
253 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
256 uint32_t value
= hashkit_digest(&ptr
->hashkit
, sort_host
, (size_t) sort_host_length
);
257 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
258 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
263 pointer_counter
+= pointer_per_server
;
266 assert_msg(ptr
, "Programmer Error, no valid ptr");
267 assert_msg(ptr
->ketama
.continuum
, "Programmer Error, empty ketama continuum");
268 assert_msg(memcached_server_count(ptr
) * MEMCACHED_POINTS_PER_SERVER
<= MEMCACHED_CONTINUUM_SIZE
,
269 "invalid size information being given to qsort()");
270 ptr
->ketama
.continuum_points_counter
= pointer_counter
;
271 qsort(ptr
->ketama
.continuum
, ptr
->ketama
.continuum_points_counter
,
272 sizeof(memcached_continuum_item_st
), continuum_item_cmp
);
275 for (uint32_t pointer_index
= 0; memcached_server_count(ptr
)
276 && pointer_index
< ((live_servers
* MEMCACHED_POINTS_PER_SERVER
) - 1);
279 WATCHPOINT_ASSERT(ptr
->ketama
.continuum
[pointer_index
].value
280 <= ptr
->ketama
.continuum
[pointer_index
+ 1].value
);
284 return MEMCACHED_SUCCESS
;
287 static memcached_return_t
server_add(Memcached
*memc
, const memcached_string_t
&hostname
,
288 in_port_t port
, uint32_t weight
, memcached_connection_t type
) {
289 assert_msg(memc
, "Programmer mistake, somehow server_add() was passed a NULL memcached_st");
291 if (memc
->number_of_hosts
) {
292 assert(memcached_instance_list(memc
));
295 if (memcached_instance_list(memc
)) {
296 assert(memc
->number_of_hosts
);
299 uint32_t host_list_size
= memc
->number_of_hosts
+ 1;
300 memcached_instance_st
*new_host_list
= libmemcached_xrealloc(
301 memc
, memcached_instance_list(memc
), host_list_size
, memcached_instance_st
);
303 if (new_host_list
== NULL
) {
304 return memcached_set_error(*memc
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
);
307 memcached_instance_set(memc
, new_host_list
, host_list_size
);
308 assert(memc
->number_of_hosts
== host_list_size
);
310 /* TODO: Check return type */
311 memcached_instance_st
*instance
=
312 memcached_instance_fetch(memc
, memcached_server_count(memc
) - 1);
314 if (instance_create_with(memc
, instance
, hostname
, port
, weight
, type
) == NULL
) {
315 return memcached_set_error(*memc
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
);
319 if (memcached_is_consistent_distribution(memc
)) {
320 memcached_set_weighted_ketama(memc
, true);
324 return run_distribution(memc
);
327 memcached_return_t
memcached_server_push(memcached_st
*shell
, const memcached_server_list_st list
) {
329 return MEMCACHED_SUCCESS
;
332 Memcached
*ptr
= memcached2Memcached(shell
);
334 uint32_t original_host_size
= memcached_server_count(ptr
);
335 uint32_t count
= memcached_server_list_count(list
);
336 uint32_t host_list_size
= count
+ original_host_size
;
338 memcached_instance_st
*new_host_list
= libmemcached_xrealloc(
339 ptr
, memcached_instance_list(ptr
), host_list_size
, memcached_instance_st
);
341 if (new_host_list
== NULL
) {
342 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
345 memcached_instance_set(ptr
, new_host_list
, host_list_size
);
347 ptr
->state
.is_parsing
= true;
348 for (uint32_t x
= 0; x
< count
; ++x
, ++original_host_size
) {
349 WATCHPOINT_ASSERT(list
[x
].hostname
[0]);
351 // We have extended the array, and now we will find it, and use it.
352 memcached_instance_st
*instance
= memcached_instance_fetch(ptr
, original_host_size
);
353 WATCHPOINT_ASSERT(instance
);
355 memcached_string_t hostname
= {memcached_string_make_from_cstr(list
[x
].hostname
)};
356 if (instance_create_with(ptr
, instance
, hostname
, list
[x
].port
, list
[x
].weight
, list
[x
].type
)
359 ptr
->state
.is_parsing
= false;
360 return memcached_set_error(*ptr
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
);
363 if (list
[x
].weight
> 1) {
364 memcached_set_weighted_ketama(ptr
, true);
367 ptr
->state
.is_parsing
= false;
369 return run_distribution(ptr
);
372 return MEMCACHED_INVALID_ARGUMENTS
;
375 memcached_return_t
memcached_instance_push(memcached_st
*ptr
,
376 const struct memcached_instance_st
*list
,
377 uint32_t number_of_hosts
) {
379 return MEMCACHED_SUCCESS
;
382 uint32_t original_host_size
= memcached_server_count(ptr
);
383 uint32_t host_list_size
= number_of_hosts
+ original_host_size
;
384 memcached_instance_st
*new_host_list
= libmemcached_xrealloc(
385 ptr
, memcached_instance_list(ptr
), host_list_size
, memcached_instance_st
);
387 if (new_host_list
== NULL
) {
388 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
391 memcached_instance_set(ptr
, new_host_list
, host_list_size
);
393 // We don't bother with lookups for this operation
394 ptr
->state
.is_parsing
= true;
396 // We use original_host_size since size will now point to the first new
397 // instance allocated.
398 for (uint32_t x
= 0; x
< number_of_hosts
; ++x
, ++original_host_size
) {
399 WATCHPOINT_ASSERT(list
[x
]._hostname
[0]);
401 // We have extended the array, and now we will find it, and use it.
402 memcached_instance_st
*instance
= memcached_instance_fetch(ptr
, original_host_size
);
403 WATCHPOINT_ASSERT(instance
);
405 memcached_string_t hostname
= {memcached_string_make_from_cstr(list
[x
]._hostname
)};
406 if (instance_create_with(ptr
, instance
, hostname
, list
[x
].port(), list
[x
].weight
, list
[x
].type
)
409 ptr
->state
.is_parsing
= false;
410 return memcached_set_error(*ptr
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
);
413 if (list
[x
].weight
> 1) {
414 memcached_set_weighted_ketama(ptr
, true);
417 ptr
->state
.is_parsing
= false;
419 return run_distribution(ptr
);
422 memcached_return_t
memcached_server_add_unix_socket(memcached_st
*ptr
, const char *filename
) {
423 return memcached_server_add_unix_socket_with_weight(ptr
, filename
, 0);
426 memcached_return_t
memcached_server_add_unix_socket_with_weight(memcached_st
*shell
,
427 const char *filename
,
429 Memcached
*ptr
= memcached2Memcached(shell
);
431 memcached_string_t _filename
= {memcached_string_make_from_cstr(filename
)};
432 if (memcached_is_valid_filename(_filename
) == false) {
433 return memcached_set_error(*ptr
, MEMCACHED_INVALID_ARGUMENTS
, MEMCACHED_AT
,
434 memcached_literal_param("Invalid filename for socket provided"));
437 return server_add(ptr
, _filename
, 0, weight
, MEMCACHED_CONNECTION_UNIX_SOCKET
);
440 return MEMCACHED_FAILURE
;
443 memcached_return_t
memcached_server_add_udp(memcached_st
*ptr
, const char *hostname
,
445 return memcached_server_add_udp_with_weight(ptr
, hostname
, port
, 0);
448 memcached_return_t
memcached_server_add_udp_with_weight(memcached_st
*shell
, const char *,
449 in_port_t
, uint32_t) {
450 Memcached
*self
= memcached2Memcached(shell
);
452 return memcached_set_error(*self
, MEMCACHED_DEPRECATED
, MEMCACHED_AT
);
455 return MEMCACHED_INVALID_ARGUMENTS
;
458 memcached_return_t
memcached_server_add(memcached_st
*shell
, const char *hostname
, in_port_t port
) {
459 return memcached_server_add_with_weight(shell
, hostname
, port
, 0);
462 memcached_return_t
memcached_server_add_with_weight(memcached_st
*shell
, const char *hostname
,
463 in_port_t port
, uint32_t weight
) {
464 Memcached
*ptr
= memcached2Memcached(shell
);
466 return MEMCACHED_INVALID_ARGUMENTS
;
470 port
= MEMCACHED_DEFAULT_PORT
;
473 size_t hostname_length
= hostname
? strlen(hostname
) : 0;
474 if (hostname_length
== 0) {
475 hostname
= "localhost";
476 hostname_length
= memcached_literal_param_size("localhost");
479 memcached_string_t _hostname
= {hostname
, hostname_length
};
481 if (memcached_is_valid_servername(_hostname
) == false) {
482 return memcached_set_error(*ptr
, MEMCACHED_INVALID_ARGUMENTS
, MEMCACHED_AT
,
483 memcached_literal_param("Invalid hostname provided"));
486 return server_add(ptr
, _hostname
, port
, weight
,
487 _hostname
.c_str
[0] == '/' ? MEMCACHED_CONNECTION_UNIX_SOCKET
488 : MEMCACHED_CONNECTION_TCP
);
491 memcached_return_t
memcached_server_add_parsed(memcached_st
*ptr
, const char *hostname
,
492 size_t hostname_length
, in_port_t port
,
494 char buffer
[MEMCACHED_NI_MAXHOST
] = {0};
496 memcpy(buffer
, hostname
, hostname_length
);
497 buffer
[hostname_length
] = 0;
499 memcached_string_t _hostname
= {buffer
, hostname_length
};
501 return server_add(ptr
, _hostname
, port
, weight
, MEMCACHED_CONNECTION_TCP
);