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>
39 #include "libmemcached/assert.hpp"
44 /* Protoypes (static) */
45 static memcached_return_t
update_continuum(Memcached
*ptr
);
47 static int compare_servers(const void *p1
, const void *p2
)
49 const memcached_instance_st
* a
= (const memcached_instance_st
*)p1
;
50 const memcached_instance_st
* b
= (const memcached_instance_st
*)p2
;
52 int return_value
= strcmp(a
->_hostname
, b
->_hostname
);
54 if (return_value
== 0)
56 return_value
= int(a
->port() - b
->port());
62 static void sort_hosts(Memcached
*ptr
)
64 if (memcached_server_count(ptr
))
66 qsort(memcached_instance_list(ptr
), memcached_server_count(ptr
), sizeof(memcached_instance_st
), compare_servers
);
71 memcached_return_t
run_distribution(Memcached
*ptr
)
73 if (ptr
->flags
.use_sort_hosts
)
78 switch (ptr
->distribution
)
80 case MEMCACHED_DISTRIBUTION_CONSISTENT
:
81 case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA
:
82 case MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA_SPY
:
83 case MEMCACHED_DISTRIBUTION_CONSISTENT_WEIGHTED
:
84 return update_continuum(ptr
);
86 case MEMCACHED_DISTRIBUTION_VIRTUAL_BUCKET
:
87 case MEMCACHED_DISTRIBUTION_MODULA
:
90 case MEMCACHED_DISTRIBUTION_RANDOM
:
91 srandom((uint32_t) time(NULL
));
94 case MEMCACHED_DISTRIBUTION_CONSISTENT_MAX
:
96 assert_msg(0, "Invalid distribution type passed to run_distribution()");
99 return MEMCACHED_SUCCESS
;
102 static uint32_t ketama_server_hash(const char *key
, size_t key_length
, uint32_t alignment
)
104 unsigned char results
[16];
106 libhashkit_md5_signature((unsigned char*)key
, key_length
, results
);
108 return ((uint32_t) (results
[3 + alignment
* 4] & 0xFF) << 24)
109 | ((uint32_t) (results
[2 + alignment
* 4] & 0xFF) << 16)
110 | ((uint32_t) (results
[1 + alignment
* 4] & 0xFF) << 8)
111 | (results
[0 + alignment
* 4] & 0xFF);
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 /* Why 153? Hmmm... */
120 WATCHPOINT_ASSERT(ct1
->value
!= 153);
121 if (ct1
->value
== ct2
->value
)
125 else if (ct1
->value
> ct2
->value
)
135 static memcached_return_t
update_continuum(Memcached
*ptr
)
137 uint32_t continuum_index
= 0;
138 uint32_t pointer_counter
= 0;
139 uint32_t pointer_per_server
= MEMCACHED_POINTS_PER_SERVER
;
140 uint32_t pointer_per_hash
= 1;
141 uint32_t live_servers
= 0;
144 if (gettimeofday(&now
, NULL
))
146 return memcached_set_errno(*ptr
, errno
, MEMCACHED_AT
);
149 memcached_instance_st
* list
= memcached_instance_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
)
165 if (ptr
->ketama
.next_distribution_rebuild
== 0 or list
[host_index
].next_retry
< ptr
->ketama
.next_distribution_rebuild
)
167 ptr
->ketama
.next_distribution_rebuild
= list
[host_index
].next_retry
;
174 live_servers
= memcached_server_count(ptr
);
177 if (live_servers
== 0)
179 return MEMCACHED_SUCCESS
;
182 uint32_t points_per_server
= (uint32_t) (memcached_is_weighted_ketama(ptr
) ? MEMCACHED_POINTS_PER_SERVER_KETAMA
: MEMCACHED_POINTS_PER_SERVER
);
183 uint32_t continuum_limit
= live_servers
* points_per_server
;
184 uint32_t continuum_extra
= MEMCACHED_CONTINUUM_ADDITION
* points_per_server
;
186 if (continuum_limit
> ptr
->ketama
.continuum_count
)
188 memcached_continuum_item_st
*new_ptr
;
190 new_ptr
= libmemcached_xrealloc(ptr
, ptr
->ketama
.continuum
, continuum_limit
+ continuum_extra
, memcached_continuum_item_st
);
194 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
197 ptr
->ketama
.continuum
= new_ptr
;
198 ptr
->ketama
.continuum_count
= continuum_limit
+ continuum_extra
;
200 assert_msg(ptr
->ketama
.continuum
, "Programmer Error, empty ketama continuum");
202 uint64_t total_weight
= 0;
203 if (memcached_is_weighted_ketama(ptr
))
205 for (uint32_t host_index
= 0; host_index
< memcached_server_count(ptr
); ++host_index
)
207 if (is_auto_ejecting
== false or list
[host_index
].next_retry
<= now
.tv_sec
)
209 total_weight
+= list
[host_index
].weight
;
214 for (uint32_t host_index
= 0; host_index
< memcached_server_count(ptr
); ++host_index
)
216 if (is_auto_ejecting
and list
[host_index
].next_retry
> now
.tv_sec
)
221 if (memcached_is_weighted_ketama(ptr
))
223 float pct
= (float)list
[host_index
].weight
/ (float)total_weight
;
224 pointer_per_server
= (uint32_t) ((::floor((float) (pct
* MEMCACHED_POINTS_PER_SERVER_KETAMA
/ 4 * (float)live_servers
+ 0.0000000001))) * 4);
228 printf("ketama_weighted:%s|%d|%llu|%u\n",
229 list
[host_index
]._hostname
,
230 list
[host_index
].port(),
231 (unsigned long long)list
[host_index
].weight
,
237 if (ptr
->distribution
== MEMCACHED_DISTRIBUTION_CONSISTENT_KETAMA_SPY
)
239 for (uint32_t pointer_index
= 0;
240 pointer_index
< pointer_per_server
/ pointer_per_hash
;
243 char sort_host
[1 +MEMCACHED_NI_MAXHOST
+1 +MEMCACHED_NI_MAXSERV
+1 + MEMCACHED_NI_MAXSERV
]= "";
244 int sort_host_length
;
246 // Spymemcached ketema key format is: hostname/ip:port-index
247 // If hostname is not available then: /ip:port-index
248 sort_host_length
= snprintf(sort_host
, sizeof(sort_host
),
250 list
[host_index
]._hostname
,
251 (uint32_t)list
[host_index
].port(),
254 if (size_t(sort_host_length
) >= sizeof(sort_host
) or sort_host_length
< 0)
256 return memcached_set_error(*ptr
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
,
257 memcached_literal_param("snprintf(sizeof(sort_host))"));
262 fprintf(stdout
, "update_continuum: key is %s\n", sort_host
);
265 if (memcached_is_weighted_ketama(ptr
))
267 for (uint32_t x
= 0; x
< pointer_per_hash
; x
++)
269 uint32_t value
= ketama_server_hash(sort_host
, (size_t)sort_host_length
, x
);
270 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
271 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
276 uint32_t value
= hashkit_digest(&ptr
->hashkit
, sort_host
, (size_t)sort_host_length
);
277 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
278 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
284 for (uint32_t pointer_index
= 1;
285 pointer_index
<= pointer_per_server
/ pointer_per_hash
;
288 char sort_host
[MEMCACHED_NI_MAXHOST
+1 +MEMCACHED_NI_MAXSERV
+1 +MEMCACHED_NI_MAXSERV
]= "";
289 int sort_host_length
;
291 if (list
[host_index
].port() == MEMCACHED_DEFAULT_PORT
)
293 sort_host_length
= snprintf(sort_host
, sizeof(sort_host
),
295 list
[host_index
]._hostname
,
300 sort_host_length
= snprintf(sort_host
, sizeof(sort_host
),
302 list
[host_index
]._hostname
,
303 (uint32_t)list
[host_index
].port(),
307 if (size_t(sort_host_length
) >= sizeof(sort_host
) or sort_host_length
< 0)
309 return memcached_set_error(*ptr
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
,
310 memcached_literal_param("snprintf(sizeof(sort_host)))"));
313 if (memcached_is_weighted_ketama(ptr
))
315 for (uint32_t x
= 0; x
< pointer_per_hash
; x
++)
317 uint32_t value
= ketama_server_hash(sort_host
, (size_t)sort_host_length
, x
);
318 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
319 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
324 uint32_t value
= hashkit_digest(&ptr
->hashkit
, sort_host
, (size_t)sort_host_length
);
325 ptr
->ketama
.continuum
[continuum_index
].index
= host_index
;
326 ptr
->ketama
.continuum
[continuum_index
++].value
= value
;
331 pointer_counter
+= pointer_per_server
;
334 assert_msg(ptr
, "Programmer Error, no valid ptr");
335 assert_msg(ptr
->ketama
.continuum
, "Programmer Error, empty ketama continuum");
336 assert_msg(memcached_server_count(ptr
) * MEMCACHED_POINTS_PER_SERVER
<= MEMCACHED_CONTINUUM_SIZE
, "invalid size information being given to qsort()");
337 ptr
->ketama
.continuum_points_counter
= pointer_counter
;
338 qsort(ptr
->ketama
.continuum
, ptr
->ketama
.continuum_points_counter
, sizeof(memcached_continuum_item_st
), continuum_item_cmp
);
342 for (uint32_t pointer_index
= 0; memcached_server_count(ptr
) && pointer_index
< ((live_servers
* MEMCACHED_POINTS_PER_SERVER
) - 1); pointer_index
++)
344 WATCHPOINT_ASSERT(ptr
->ketama
.continuum
[pointer_index
].value
<= ptr
->ketama
.continuum
[pointer_index
+ 1].value
);
348 return MEMCACHED_SUCCESS
;
351 static memcached_return_t
server_add(Memcached
*memc
,
352 const memcached_string_t
& hostname
,
355 memcached_connection_t type
)
357 assert_msg(memc
, "Programmer mistake, somehow server_add() was passed a NULL memcached_st");
359 if (memc
->number_of_hosts
)
361 assert(memcached_instance_list(memc
));
364 if (memcached_instance_list(memc
))
366 assert(memc
->number_of_hosts
);
369 uint32_t host_list_size
= memc
->number_of_hosts
+1;
370 memcached_instance_st
* new_host_list
= libmemcached_xrealloc(memc
, memcached_instance_list(memc
), host_list_size
, memcached_instance_st
);
372 if (new_host_list
== NULL
)
374 return memcached_set_error(*memc
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
);
377 memcached_instance_set(memc
, new_host_list
, host_list_size
);
378 assert(memc
->number_of_hosts
== host_list_size
);
380 /* TODO: Check return type */
381 memcached_instance_st
* instance
= memcached_instance_fetch(memc
, memcached_server_count(memc
) -1);
383 if (__instance_create_with(memc
, instance
, hostname
, port
, weight
, type
) == NULL
)
385 return memcached_set_error(*memc
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
);
390 if (memcached_is_consistent_distribution(memc
))
392 memcached_set_weighted_ketama(memc
, true);
396 return run_distribution(memc
);
400 memcached_return_t
memcached_server_push(memcached_st
*shell
, const memcached_server_list_st list
)
404 return MEMCACHED_SUCCESS
;
407 Memcached
* ptr
= memcached2Memcached(shell
);
410 uint32_t original_host_size
= memcached_server_count(ptr
);
411 uint32_t count
= memcached_server_list_count(list
);
412 uint32_t host_list_size
= count
+original_host_size
;
414 memcached_instance_st
* new_host_list
= libmemcached_xrealloc(ptr
, memcached_instance_list(ptr
), host_list_size
, memcached_instance_st
);
416 if (new_host_list
== NULL
)
418 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
421 memcached_instance_set(ptr
, new_host_list
, host_list_size
);
423 ptr
->state
.is_parsing
= true;
424 for (uint32_t x
= 0; x
< count
; ++x
, ++original_host_size
)
426 WATCHPOINT_ASSERT(list
[x
].hostname
[0] != 0);
428 // We have extended the array, and now we will find it, and use it.
429 memcached_instance_st
* instance
= memcached_instance_fetch(ptr
, original_host_size
);
430 WATCHPOINT_ASSERT(instance
);
432 memcached_string_t hostname
= { memcached_string_make_from_cstr(list
[x
].hostname
) };
433 if (__instance_create_with(ptr
, instance
,
435 list
[x
].port
, list
[x
].weight
, list
[x
].type
) == NULL
)
437 ptr
->state
.is_parsing
= false;
438 return memcached_set_error(*ptr
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
);
441 if (list
[x
].weight
> 1)
443 memcached_set_weighted_ketama(ptr
, true);
446 ptr
->state
.is_parsing
= false;
448 return run_distribution(ptr
);
451 return MEMCACHED_INVALID_ARGUMENTS
;
454 memcached_return_t
memcached_instance_push(memcached_st
*ptr
, const struct memcached_instance_st
* list
, uint32_t number_of_hosts
)
458 return MEMCACHED_SUCCESS
;
461 uint32_t original_host_size
= memcached_server_count(ptr
);
462 uint32_t host_list_size
= number_of_hosts
+original_host_size
;
463 memcached_instance_st
* new_host_list
= libmemcached_xrealloc(ptr
, memcached_instance_list(ptr
), host_list_size
, memcached_instance_st
);
465 if (new_host_list
== NULL
)
467 return MEMCACHED_MEMORY_ALLOCATION_FAILURE
;
470 memcached_instance_set(ptr
, new_host_list
, host_list_size
);
472 // We don't bother with lookups for this operation
473 ptr
->state
.is_parsing
= true;
475 // We use original_host_size since size will now point to the first new
476 // instance allocated.
477 for (uint32_t x
= 0; x
< number_of_hosts
; ++x
, ++original_host_size
)
479 WATCHPOINT_ASSERT(list
[x
]._hostname
[0] != 0);
481 // We have extended the array, and now we will find it, and use it.
482 memcached_instance_st
* instance
= memcached_instance_fetch(ptr
, original_host_size
);
483 WATCHPOINT_ASSERT(instance
);
485 memcached_string_t hostname
= { memcached_string_make_from_cstr(list
[x
]._hostname
) };
486 if (__instance_create_with(ptr
, instance
,
488 list
[x
].port(), list
[x
].weight
, list
[x
].type
) == NULL
)
490 ptr
->state
.is_parsing
= false;
491 return memcached_set_error(*ptr
, MEMCACHED_MEMORY_ALLOCATION_FAILURE
, MEMCACHED_AT
);
494 if (list
[x
].weight
> 1)
496 memcached_set_weighted_ketama(ptr
, true);
499 ptr
->state
.is_parsing
= false;
501 return run_distribution(ptr
);
504 memcached_return_t
memcached_server_add_unix_socket(memcached_st
*ptr
,
505 const char *filename
)
507 return memcached_server_add_unix_socket_with_weight(ptr
, filename
, 0);
510 memcached_return_t
memcached_server_add_unix_socket_with_weight(memcached_st
*shell
,
511 const char *filename
,
514 Memcached
* ptr
= memcached2Memcached(shell
);
517 memcached_string_t _filename
= { memcached_string_make_from_cstr(filename
) };
518 if (memcached_is_valid_filename(_filename
) == false)
520 return memcached_set_error(*ptr
, MEMCACHED_INVALID_ARGUMENTS
, MEMCACHED_AT
, memcached_literal_param("Invalid filename for socket provided"));
523 return server_add(ptr
, _filename
, 0, weight
, MEMCACHED_CONNECTION_UNIX_SOCKET
);
526 return MEMCACHED_FAILURE
;
529 memcached_return_t
memcached_server_add_udp(memcached_st
*ptr
,
530 const char *hostname
,
533 return memcached_server_add_udp_with_weight(ptr
, hostname
, port
, 0);
536 memcached_return_t
memcached_server_add_udp_with_weight(memcached_st
*shell
,
541 Memcached
* self
= memcached2Memcached(shell
);
544 return memcached_set_error(*self
, MEMCACHED_DEPRECATED
, MEMCACHED_AT
);
547 return MEMCACHED_INVALID_ARGUMENTS
;
550 memcached_return_t
memcached_server_add(memcached_st
*shell
,
551 const char *hostname
,
554 return memcached_server_add_with_weight(shell
, hostname
, port
, 0);
557 memcached_return_t
memcached_server_add_with_weight(memcached_st
*shell
,
558 const char *hostname
,
562 Memcached
* ptr
= memcached2Memcached(shell
);
565 return MEMCACHED_INVALID_ARGUMENTS
;
570 port
= MEMCACHED_DEFAULT_PORT
;
573 size_t hostname_length
= hostname
? strlen(hostname
) : 0;
574 if (hostname_length
== 0)
576 hostname
= "localhost";
577 hostname_length
= memcached_literal_param_size("localhost");
580 memcached_string_t _hostname
= { hostname
, hostname_length
};
582 if (memcached_is_valid_servername(_hostname
) == false)
584 return memcached_set_error(*ptr
, MEMCACHED_INVALID_ARGUMENTS
, MEMCACHED_AT
, memcached_literal_param("Invalid hostname provided"));
587 return server_add(ptr
, _hostname
, port
, weight
, _hostname
.c_str
[0] == '/' ? MEMCACHED_CONNECTION_UNIX_SOCKET
: MEMCACHED_CONNECTION_TCP
);
590 memcached_return_t
memcached_server_add_parsed(memcached_st
*ptr
,
591 const char *hostname
,
592 size_t hostname_length
,
596 char buffer
[MEMCACHED_NI_MAXHOST
]= { 0 };
598 memcpy(buffer
, hostname
, hostname_length
);
599 buffer
[hostname_length
]= 0;
601 memcached_string_t _hostname
= { buffer
, hostname_length
};
603 return server_add(ptr
, _hostname
,
606 MEMCACHED_CONNECTION_TCP
);