Index: /tags/1.2.5/stats.c =================================================================== --- /tags/1.2.5/stats.c (revision 591) +++ /tags/1.2.5/stats.c (revision 591) @@ -0,0 +1,365 @@ +/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */ +/* + * Detailed statistics management. For simple stats like total number of + * "get" requests, we use inline code in memcached.c and friends, but when + * stats detail mode is activated, the code here records more information. + * + * Author: + * Steven Grimm + * + * $Id$ + */ +#include "memcached.h" +#include +#include +#include +#include + +/* + * Stats are tracked on the basis of key prefixes. This is a simple + * fixed-size hash of prefixes; we run the prefixes through the same + * CRC function used by the cache hashtable. + */ +typedef struct _prefix_stats PREFIX_STATS; +struct _prefix_stats { + char *prefix; + size_t prefix_len; + uint64_t num_gets; + uint64_t num_sets; + uint64_t num_deletes; + uint64_t num_hits; + PREFIX_STATS *next; +}; + +#define PREFIX_HASH_SIZE 256 + +static PREFIX_STATS *prefix_stats[PREFIX_HASH_SIZE]; +static int num_prefixes = 0; +static int total_prefix_size = 0; + +void stats_prefix_init() { + memset(prefix_stats, 0, sizeof(prefix_stats)); +} + +/* + * Cleans up all our previously collected stats. NOTE: the stats lock is + * assumed to be held when this is called. + */ +void stats_prefix_clear() { + int i; + + for (i = 0; i < PREFIX_HASH_SIZE; i++) { + PREFIX_STATS *cur, *next; + for (cur = prefix_stats[i]; cur != NULL; cur = next) { + next = cur->next; + free(cur->prefix); + free(cur); + } + prefix_stats[i] = NULL; + } + num_prefixes = 0; + total_prefix_size = 0; +} + +/* + * Returns the stats structure for a prefix, creating it if it's not already + * in the list. + */ +/*@null@*/ +static PREFIX_STATS *stats_prefix_find(const char *key) { + PREFIX_STATS *pfs; + uint32_t hashval; + size_t length; + + assert(key != NULL); + + for (length = 0; key[length] != '\0'; length++) + if (key[length] == settings.prefix_delimiter) + break; + + hashval = hash(key, length, 0) % PREFIX_HASH_SIZE; + + for (pfs = prefix_stats[hashval]; NULL != pfs; pfs = pfs->next) { + if (strncmp(pfs->prefix, key, length) == 0) + return pfs; + } + + pfs = calloc(sizeof(PREFIX_STATS), 1); + if (NULL == pfs) { + perror("Can't allocate space for stats structure: calloc"); + return NULL; + } + + pfs->prefix = malloc(length + 1); + if (NULL == pfs->prefix) { + perror("Can't allocate space for copy of prefix: malloc"); + free(pfs); + return NULL; + } + + strncpy(pfs->prefix, key, length); + pfs->prefix[length] = '\0'; /* because strncpy() sucks */ + pfs->prefix_len = length; + + pfs->next = prefix_stats[hashval]; + prefix_stats[hashval] = pfs; + + num_prefixes++; + total_prefix_size += length; + + return pfs; +} + +/* + * Records a "get" of a key. + */ +void stats_prefix_record_get(const char *key, const bool is_hit) { + PREFIX_STATS *pfs; + + STATS_LOCK(); + pfs = stats_prefix_find(key); + if (NULL != pfs) { + pfs->num_gets++; + if (is_hit) { + pfs->num_hits++; + } + } + STATS_UNLOCK(); +} + +/* + * Records a "delete" of a key. + */ +void stats_prefix_record_delete(const char *key) { + PREFIX_STATS *pfs; + + STATS_LOCK(); + pfs = stats_prefix_find(key); + if (NULL != pfs) { + pfs->num_deletes++; + } + STATS_UNLOCK(); +} + +/* + * Records a "set" of a key. + */ +void stats_prefix_record_set(const char *key) { + PREFIX_STATS *pfs; + + STATS_LOCK(); + pfs = stats_prefix_find(key); + if (NULL != pfs) { + pfs->num_sets++; + } + STATS_UNLOCK(); +} + +/* + * Returns stats in textual form suitable for writing to a client. + */ +/*@null@*/ +char *stats_prefix_dump(int *length) { + const char *format = "PREFIX %s get %llu hit %llu set %llu del %llu\r\n"; + PREFIX_STATS *pfs; + char *buf; + int i, pos; + size_t size; + + /* + * Figure out how big the buffer needs to be. This is the sum of the + * lengths of the prefixes themselves, plus the size of one copy of + * the per-prefix output with 20-digit values for all the counts, + * plus space for the "END" at the end. + */ + STATS_LOCK(); + size = strlen(format) + total_prefix_size + + num_prefixes * (strlen(format) - 2 /* %s */ + + 4 * (20 - 4)) /* %llu replaced by 20-digit num */ + + sizeof("END\r\n"); + buf = malloc(size); + if (NULL == buf) { + perror("Can't allocate stats response: malloc"); + STATS_UNLOCK(); + return NULL; + } + + pos = 0; + for (i = 0; i < PREFIX_HASH_SIZE; i++) { + for (pfs = prefix_stats[i]; NULL != pfs; pfs = pfs->next) { + pos += snprintf(buf + pos, size-pos, format, + pfs->prefix, pfs->num_gets, pfs->num_hits, + pfs->num_sets, pfs->num_deletes); + } + } + + STATS_UNLOCK(); + memcpy(buf + pos, "END\r\n", 6); + + *length = pos + 5; + return buf; +} + + +#ifdef UNIT_TEST + +/**************************************************************************** + To run unit tests, compile with $(CC) -DUNIT_TEST stats.c assoc.o + (need assoc.o to get the hash() function). +****************************************************************************/ + +struct settings settings; + +static char *current_test = ""; +static int test_count = 0; +static int fail_count = 0; + +static void fail(char *what) { printf("\tFAIL: %s\n", what); fflush(stdout); fail_count++; } +static void test_equals_int(char *what, int a, int b) { test_count++; if (a != b) fail(what); } +static void test_equals_ptr(char *what, void *a, void *b) { test_count++; if (a != b) fail(what); } +static void test_equals_str(char *what, const char *a, const char *b) { test_count++; if (strcmp(a, b)) fail(what); } +static void test_equals_ull(char *what, uint64_t a, uint64_t b) { test_count++; if (a != b) fail(what); } +static void test_notequals_ptr(char *what, void *a, void *b) { test_count++; if (a == b) fail(what); } +static void test_notnull_ptr(char *what, void *a) { test_count++; if (NULL == a) fail(what); } + +static void test_prefix_find() { + PREFIX_STATS *pfs1, *pfs2; + + pfs1 = stats_prefix_find("abc"); + test_notnull_ptr("initial prefix find", pfs1); + test_equals_ull("request counts", 0ULL, + pfs1->num_gets + pfs1->num_sets + pfs1->num_deletes + pfs1->num_hits); + pfs2 = stats_prefix_find("abc"); + test_equals_ptr("find of same prefix", pfs1, pfs2); + pfs2 = stats_prefix_find("abc:"); + test_equals_ptr("find of same prefix, ignoring delimiter", pfs1, pfs2); + pfs2 = stats_prefix_find("abc:d"); + test_equals_ptr("find of same prefix, ignoring extra chars", pfs1, pfs2); + pfs2 = stats_prefix_find("xyz123"); + test_notequals_ptr("find of different prefix", pfs1, pfs2); + pfs2 = stats_prefix_find("ab:"); + test_notequals_ptr("find of shorter prefix", pfs1, pfs2); +} + +static void test_prefix_record_get() { + PREFIX_STATS *pfs; + + stats_prefix_record_get("abc:123", 0); + pfs = stats_prefix_find("abc:123"); + test_equals_ull("get count after get #1", 1, pfs->num_gets); + test_equals_ull("hit count after get #1", 0, pfs->num_hits); + stats_prefix_record_get("abc:456", 0); + test_equals_ull("get count after get #2", 2, pfs->num_gets); + test_equals_ull("hit count after get #2", 0, pfs->num_hits); + stats_prefix_record_get("abc:456", 1); + test_equals_ull("get count after get #3", 3, pfs->num_gets); + test_equals_ull("hit count after get #3", 1, pfs->num_hits); + stats_prefix_record_get("def:", 1); + test_equals_ull("get count after get #4", 3, pfs->num_gets); + test_equals_ull("hit count after get #4", 1, pfs->num_hits); +} + +static void test_prefix_record_delete() { + PREFIX_STATS *pfs; + + stats_prefix_record_delete("abc:123"); + pfs = stats_prefix_find("abc:123"); + test_equals_ull("get count after delete #1", 0, pfs->num_gets); + test_equals_ull("hit count after delete #1", 0, pfs->num_hits); + test_equals_ull("delete count after delete #1", 1, pfs->num_deletes); + test_equals_ull("set count after delete #1", 0, pfs->num_sets); + stats_prefix_record_delete("def:"); + test_equals_ull("delete count after delete #2", 1, pfs->num_deletes); +} + +static void test_prefix_record_set() { + PREFIX_STATS *pfs; + + stats_prefix_record_set("abc:123"); + pfs = stats_prefix_find("abc:123"); + test_equals_ull("get count after set #1", 0, pfs->num_gets); + test_equals_ull("hit count after set #1", 0, pfs->num_hits); + test_equals_ull("delete count after set #1", 0, pfs->num_deletes); + test_equals_ull("set count after set #1", 1, pfs->num_sets); + stats_prefix_record_delete("def:"); + test_equals_ull("set count after set #2", 1, pfs->num_sets); +} + +static void test_prefix_dump() { + int hashval = hash("abc", 3, 0) % PREFIX_HASH_SIZE; + char tmp[500]; + char *expected; + int keynum; + int length; + + test_equals_str("empty stats", "END\r\n", stats_prefix_dump(&length)); + test_equals_int("empty stats length", 5, length); + stats_prefix_record_set("abc:123"); + expected = "PREFIX abc get 0 hit 0 set 1 del 0\r\nEND\r\n"; + test_equals_str("stats after set", expected, stats_prefix_dump(&length)); + test_equals_int("stats length after set", strlen(expected), length); + stats_prefix_record_get("abc:123", 0); + expected = "PREFIX abc get 1 hit 0 set 1 del 0\r\nEND\r\n"; + test_equals_str("stats after get #1", expected, stats_prefix_dump(&length)); + test_equals_int("stats length after get #1", strlen(expected), length); + stats_prefix_record_get("abc:123", 1); + expected = "PREFIX abc get 2 hit 1 set 1 del 0\r\nEND\r\n"; + test_equals_str("stats after get #2", expected, stats_prefix_dump(&length)); + test_equals_int("stats length after get #2", strlen(expected), length); + stats_prefix_record_delete("abc:123"); + expected = "PREFIX abc get 2 hit 1 set 1 del 1\r\nEND\r\n"; + test_equals_str("stats after del #1", expected, stats_prefix_dump(&length)); + test_equals_int("stats length after del #1", strlen(expected), length); + + /* The order of results might change if we switch hash functions. */ + stats_prefix_record_delete("def:123"); + expected = "PREFIX abc get 2 hit 1 set 1 del 1\r\n" + "PREFIX def get 0 hit 0 set 0 del 1\r\n" + "END\r\n"; + test_equals_str("stats after del #2", expected, stats_prefix_dump(&length)); + test_equals_int("stats length after del #2", strlen(expected), length); + + /* Find a key that hashes to the same bucket as "abc" */ + for (keynum = 0; keynum < PREFIX_HASH_SIZE * 100; keynum++) { + sprintf(tmp, "%d", keynum); + if (hashval == hash(tmp, strlen(tmp), 0) % PREFIX_HASH_SIZE) { + break; + } + } + stats_prefix_record_set(tmp); + sprintf(tmp, "PREFIX %d get 0 hit 0 set 1 del 0\r\n" + "PREFIX abc get 2 hit 1 set 1 del 1\r\n" + "PREFIX def get 0 hit 0 set 0 del 1\r\n" + "END\r\n", keynum); + test_equals_str("stats with two stats in one bucket", + tmp, stats_prefix_dump(&length)); + test_equals_int("stats length with two stats in one bucket", + strlen(tmp), length); +} + +static void run_test(char *what, void (*func)(void)) { + current_test = what; + test_count = fail_count = 0; + puts(what); + fflush(stdout); + + stats_prefix_clear(); + (func)(); + printf("\t%d / %d pass\n", (test_count - fail_count), test_count); +} + +/* In case we're compiled in thread mode */ +void mt_stats_lock() { } +void mt_stats_unlock() { } + +main(int argc, char **argv) { + stats_prefix_init(); + settings.prefix_delimiter = ':'; + run_test("stats_prefix_find", test_prefix_find); + run_test("stats_prefix_record_get", test_prefix_record_get); + run_test("stats_prefix_record_delete", test_prefix_record_delete); + run_test("stats_prefix_record_set", test_prefix_record_set); + run_test("stats_prefix_dump", test_prefix_dump); +} + +#endif Index: /tags/1.2.5/LICENSE =================================================================== --- /tags/1.2.5/LICENSE (revision 257) +++ /tags/1.2.5/LICENSE (revision 257) @@ -0,0 +1,30 @@ +Copyright (c) 2003, Danga Interactive, Inc. +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are +met: + + * Redistributions of source code must retain the above copyright +notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above +copyright notice, this list of conditions and the following disclaimer +in the documentation and/or other materials provided with the +distribution. + + * Neither the name of the Danga Interactive nor the names of its +contributors may be used to endorse or promote products derived from +this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR +A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT +OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT +LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, +DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY +THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. Index: /tags/1.2.5/stats.h =================================================================== --- /tags/1.2.5/stats.h (revision 512) +++ /tags/1.2.5/stats.h (revision 512) @@ -0,0 +1,8 @@ +/* stats */ +void stats_prefix_init(void); +void stats_prefix_clear(void); +void stats_prefix_record_get(const char *key, const bool is_hit); +void stats_prefix_record_delete(const char *key); +void stats_prefix_record_set(const char *key); +/*@null@*/ +char *stats_prefix_dump(int *length); Index: /tags/1.2.5/daemon.c =================================================================== --- /tags/1.2.5/daemon.c (revision 587) +++ /tags/1.2.5/daemon.c (revision 587) @@ -0,0 +1,68 @@ +/* $Header: /cvsroot/wikipedia/willow/src/bin/willow/daemon.c,v 1.1 2005/05/02 19:15:21 kateturner Exp $ */ +/* $NetBSD: daemon.c,v 1.9 2003/08/07 16:42:46 agc Exp $ */ +/*- + * Copyright (c) 1990, 1993 + * The Regents of the University of California. All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions + * are met: + * 1. Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * 2. Redistributions in binary form must reproduce the above copyright + * notice, this list of conditions and the following disclaimer in the + * documentation and/or other materials provided with the distribution. + * 3. Neither the name of the University nor the names of its contributors + * may be used to endorse or promote products derived from this software + * without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND + * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE + * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE + * ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE + * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL + * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS + * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) + * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT + * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY + * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF + * SUCH DAMAGE. + */ + +#if defined __SUNPRO_C || defined __DECC || defined __HP_cc +# pragma ident "@(#)$Header: /cvsroot/wikipedia/willow/src/bin/willow/daemon.c,v 1.1 2005/05/02 19:15:21 kateturner Exp $" +# pragma ident "$NetBSD: daemon.c,v 1.9 2003/08/07 16:42:46 agc Exp $" +#endif + +#include +#include +#include + +int daemon(int nochdir, int noclose) +{ + int fd; + + switch (fork()) { + case -1: + return (-1); + case 0: + break; + default: + _exit(EXIT_SUCCESS); + } + + if (setsid() == -1) + return (-1); + + if (nochdir == 0) + (void)chdir("/"); + + if (noclose == 0 && (fd = open("/dev/null", O_RDWR, 0)) != -1) { + (void)dup2(fd, STDIN_FILENO); + (void)dup2(fd, STDOUT_FILENO); + (void)dup2(fd, STDERR_FILENO); + if (fd > STDERR_FILENO) + (void)close(fd); + } + return (0); +} Index: /tags/1.2.5/slabs.c =================================================================== --- /tags/1.2.5/slabs.c (revision 739) +++ /tags/1.2.5/slabs.c (revision 739) @@ -0,0 +1,421 @@ +/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */ +/* + * Slabs memory allocation, based on powers-of-N. Slabs are up to 1MB in size + * and are divided into chunks. The chunk sizes start off at the size of the + * "item" structure plus space for a small key and value. They increase by + * a multiplier factor from there, up to half the maximum slab size. The last + * slab size is always 1MB, since that's the maximum item size allowed by the + * memcached protocol. + * + * $Id$ + */ +#include "memcached.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#define POWER_SMALLEST 1 +#define POWER_LARGEST 200 +#define POWER_BLOCK 1048576 +#define CHUNK_ALIGN_BYTES 8 +#define DONT_PREALLOC_SLABS + +/* powers-of-N allocation structures */ + +typedef struct { + unsigned int size; /* sizes of items */ + unsigned int perslab; /* how many items per slab */ + + void **slots; /* list of item ptrs */ + unsigned int sl_total; /* size of previous array */ + unsigned int sl_curr; /* first free slot */ + + void *end_page_ptr; /* pointer to next free item at end of page, or 0 */ + unsigned int end_page_free; /* number of items remaining at end of last alloced page */ + + unsigned int slabs; /* how many slabs were allocated for this class */ + + void **slab_list; /* array of slab pointers */ + unsigned int list_size; /* size of prev array */ + + unsigned int killing; /* index+1 of dying slab, or zero if none */ +} slabclass_t; + +static slabclass_t slabclass[POWER_LARGEST + 1]; +static size_t mem_limit = 0; +static size_t mem_malloced = 0; +static int power_largest; + +static void *mem_base = NULL; +static void *mem_current = NULL; +static size_t mem_avail = 0; + +/* + * Forward Declarations + */ +static int do_slabs_newslab(const unsigned int id); +static void *memory_allocate(size_t size); + +#ifndef DONT_PREALLOC_SLABS +/* Preallocate as many slab pages as possible (called from slabs_init) + on start-up, so users don't get confused out-of-memory errors when + they do have free (in-slab) space, but no space to make new slabs. + if maxslabs is 18 (POWER_LARGEST - POWER_SMALLEST + 1), then all + slab types can be made. if max memory is less than 18 MB, only the + smaller ones will be made. */ +static void slabs_preallocate (const unsigned int maxslabs); +#endif + +/* + * Figures out which slab class (chunk size) is required to store an item of + * a given size. + * + * Given object size, return id to use when allocating/freeing memory for object + * 0 means error: can't store such a large object + */ + +unsigned int slabs_clsid(const size_t size) { + int res = POWER_SMALLEST; + + if (size == 0) + return 0; + while (size > slabclass[res].size) + if (res++ == power_largest) /* won't fit in the biggest slab */ + return 0; + return res; +} + +/** + * Determines the chunk sizes and initializes the slab class descriptors + * accordingly. + */ +void slabs_init(const size_t limit, const double factor, const bool prealloc) { + int i = POWER_SMALLEST - 1; + unsigned int size = sizeof(item) + settings.chunk_size; + + /* Factor of 2.0 means use the default memcached behavior */ + if (factor == 2.0 && size < 128) + size = 128; + + mem_limit = limit; + + if (prealloc) { + /* Allocate everything in a big chunk with malloc */ + mem_base = malloc(mem_limit); + if (mem_base != NULL) { + mem_current = mem_base; + mem_avail = mem_limit; + } else { + fprintf(stderr, "Warning: Failed to allocate requested memory in" + " one large chunk.\nWill allocate in smaller chunks\n"); + } + } + + memset(slabclass, 0, sizeof(slabclass)); + + while (++i < POWER_LARGEST && size <= POWER_BLOCK / 2) { + /* Make sure items are always n-byte aligned */ + if (size % CHUNK_ALIGN_BYTES) + size += CHUNK_ALIGN_BYTES - (size % CHUNK_ALIGN_BYTES); + + slabclass[i].size = size; + slabclass[i].perslab = POWER_BLOCK / slabclass[i].size; + size *= factor; + if (settings.verbose > 1) { + fprintf(stderr, "slab class %3d: chunk size %6u perslab %5u\n", + i, slabclass[i].size, slabclass[i].perslab); + } + } + + power_largest = i; + slabclass[power_largest].size = POWER_BLOCK; + slabclass[power_largest].perslab = 1; + + /* for the test suite: faking of how much we've already malloc'd */ + { + char *t_initial_malloc = getenv("T_MEMD_INITIAL_MALLOC"); + if (t_initial_malloc) { + mem_malloced = (size_t)atol(t_initial_malloc); + } + + } + +#ifndef DONT_PREALLOC_SLABS + { + char *pre_alloc = getenv("T_MEMD_SLABS_ALLOC"); + + if (pre_alloc == NULL || atoi(pre_alloc) != 0) { + slabs_preallocate(power_largest); + } + } +#endif +} + +#ifndef DONT_PREALLOC_SLABS +static void slabs_preallocate (const unsigned int maxslabs) { + int i; + unsigned int prealloc = 0; + + /* pre-allocate a 1MB slab in every size class so people don't get + confused by non-intuitive "SERVER_ERROR out of memory" + messages. this is the most common question on the mailing + list. if you really don't want this, you can rebuild without + these three lines. */ + + for (i = POWER_SMALLEST; i <= POWER_LARGEST; i++) { + if (++prealloc > maxslabs) + return; + do_slabs_newslab(i); + } + +} +#endif + +static int grow_slab_list (const unsigned int id) { + slabclass_t *p = &slabclass[id]; + if (p->slabs == p->list_size) { + size_t new_size = (p->list_size != 0) ? p->list_size * 2 : 16; + void *new_list = realloc(p->slab_list, new_size * sizeof(void *)); + if (new_list == 0) return 0; + p->list_size = new_size; + p->slab_list = new_list; + } + return 1; +} + +static int do_slabs_newslab(const unsigned int id) { + slabclass_t *p = &slabclass[id]; +#ifdef ALLOW_SLABS_REASSIGN + int len = POWER_BLOCK; +#else + int len = p->size * p->perslab; +#endif + char *ptr; + + if (mem_limit && mem_malloced + len > mem_limit && p->slabs > 0) + return 0; + + if (grow_slab_list(id) == 0) return 0; + + ptr = memory_allocate((size_t)len); + if (ptr == 0) return 0; + + memset(ptr, 0, (size_t)len); + p->end_page_ptr = ptr; + p->end_page_free = p->perslab; + + p->slab_list[p->slabs++] = ptr; + mem_malloced += len; + return 1; +} + +/*@null@*/ +void *do_slabs_alloc(const size_t size, unsigned int id) { + slabclass_t *p; + + if (id < POWER_SMALLEST || id > power_largest) + return NULL; + + p = &slabclass[id]; + assert(p->sl_curr == 0 || ((item *)p->slots[p->sl_curr - 1])->slabs_clsid == 0); + +#ifdef USE_SYSTEM_MALLOC + if (mem_limit && mem_malloced + size > mem_limit) + return 0; + mem_malloced += size; + return malloc(size); +#endif + + /* fail unless we have space at the end of a recently allocated page, + we have something on our freelist, or we could allocate a new page */ + if (! (p->end_page_ptr != 0 || p->sl_curr != 0 || do_slabs_newslab(id) != 0)) + return 0; + + /* return off our freelist, if we have one */ + if (p->sl_curr != 0) + return p->slots[--p->sl_curr]; + + /* if we recently allocated a whole page, return from that */ + if (p->end_page_ptr) { + void *ptr = p->end_page_ptr; + if (--p->end_page_free != 0) { + p->end_page_ptr += p->size; + } else { + p->end_page_ptr = 0; + } + return ptr; + } + + return NULL; /* shouldn't ever get here */ +} + +void do_slabs_free(void *ptr, const size_t size, unsigned int id) { + slabclass_t *p; + + assert(((item *)ptr)->slabs_clsid == 0); + assert(id >= POWER_SMALLEST && id <= power_largest); + if (id < POWER_SMALLEST || id > power_largest) + return; + + p = &slabclass[id]; + +#ifdef USE_SYSTEM_MALLOC + mem_malloced -= size; + free(ptr); + return; +#endif + + if (p->sl_curr == p->sl_total) { /* need more space on the free list */ + int new_size = (p->sl_total != 0) ? p->sl_total * 2 : 16; /* 16 is arbitrary */ + void **new_slots = realloc(p->slots, new_size * sizeof(void *)); + if (new_slots == 0) + return; + p->slots = new_slots; + p->sl_total = new_size; + } + p->slots[p->sl_curr++] = ptr; + return; +} + +/*@null@*/ +char* do_slabs_stats(int *buflen) { + int i, total; + char *buf = (char *)malloc(power_largest * 200 + 100); + char *bufcurr = buf; + + *buflen = 0; + if (buf == NULL) return NULL; + + total = 0; + for(i = POWER_SMALLEST; i <= power_largest; i++) { + slabclass_t *p = &slabclass[i]; + if (p->slabs != 0) { + unsigned int perslab, slabs; + + slabs = p->slabs; + perslab = p->perslab; + + bufcurr += sprintf(bufcurr, "STAT %d:chunk_size %u\r\n", i, p->size); + bufcurr += sprintf(bufcurr, "STAT %d:chunks_per_page %u\r\n", i, perslab); + bufcurr += sprintf(bufcurr, "STAT %d:total_pages %u\r\n", i, slabs); + bufcurr += sprintf(bufcurr, "STAT %d:total_chunks %u\r\n", i, slabs*perslab); + bufcurr += sprintf(bufcurr, "STAT %d:used_chunks %u\r\n", i, slabs*perslab - p->sl_curr); + bufcurr += sprintf(bufcurr, "STAT %d:free_chunks %u\r\n", i, p->sl_curr); + bufcurr += sprintf(bufcurr, "STAT %d:free_chunks_end %u\r\n", i, p->end_page_free); + total++; + } + } + bufcurr += sprintf(bufcurr, "STAT active_slabs %d\r\nSTAT total_malloced %llu\r\n", total, (unsigned long long)mem_malloced); + bufcurr += sprintf(bufcurr, "END\r\n"); + *buflen = bufcurr - buf; + return buf; +} + +#ifdef ALLOW_SLABS_REASSIGN +/* Blows away all the items in a slab class and moves its slabs to another + class. This is only used by the "slabs reassign" command, for manual tweaking + of memory allocation. It's disabled by default since it requires that all + slabs be the same size (which can waste space for chunk size mantissas of + other than 2.0). + 1 = success + 0 = fail + -1 = tried. busy. send again shortly. */ +int do_slabs_reassign(unsigned char srcid, unsigned char dstid) { + void *slab, *slab_end; + slabclass_t *p, *dp; + void *iter; + bool was_busy = false; + + if (srcid < POWER_SMALLEST || srcid > power_largest || + dstid < POWER_SMALLEST || dstid > power_largest) + return 0; + + p = &slabclass[srcid]; + dp = &slabclass[dstid]; + + /* fail if src still populating, or no slab to give up in src */ + if (p->end_page_ptr || ! p->slabs) + return 0; + + /* fail if dst is still growing or we can't make room to hold its new one */ + if (dp->end_page_ptr || ! grow_slab_list(dstid)) + return 0; + + if (p->killing == 0) p->killing = 1; + + slab = p->slab_list[p->killing - 1]; + slab_end = (char*)slab + POWER_BLOCK; + + for (iter = slab; iter < slab_end; (char*)iter += p->size) { + item *it = (item *)iter; + if (it->slabs_clsid) { + if (it->refcount) was_busy = true; + item_unlink(it); + } + } + + /* go through free list and discard items that are no longer part of this slab */ + { + int fi; + for (fi = p->sl_curr - 1; fi >= 0; fi--) { + if (p->slots[fi] >= slab && p->slots[fi] < slab_end) { + p->sl_curr--; + if (p->sl_curr > fi) p->slots[fi] = p->slots[p->sl_curr]; + } + } + } + + if (was_busy) return -1; + + /* if good, now move it to the dst slab class */ + p->slab_list[p->killing - 1] = p->slab_list[p->slabs - 1]; + p->slabs--; + p->killing = 0; + dp->slab_list[dp->slabs++] = slab; + dp->end_page_ptr = slab; + dp->end_page_free = dp->perslab; + /* this isn't too critical, but other parts of the code do asserts to + make sure this field is always 0. */ + for (iter = slab; iter < slab_end; (char*)iter += dp->size) { + ((item *)iter)->slabs_clsid = 0; + } + return 1; +} +#endif + +static void *memory_allocate(size_t size) { + void *ret; + + if (mem_base == NULL) { + /* We are not using a preallocated large memory chunk */ + ret = malloc(size); + } else { + ret = mem_current; + + if (size > mem_avail) { + return NULL; + } + + /* mem_current pointer _must_ be aligned!!! */ + if (size % CHUNK_ALIGN_BYTES) { + size += CHUNK_ALIGN_BYTES - (size % CHUNK_ALIGN_BYTES); + } + + mem_current += size; + if (size < mem_avail) { + mem_avail -= size; + } else { + mem_avail = 0; + } + } + + return ret; +} Index: /tags/1.2.5/AUTHORS =================================================================== --- /tags/1.2.5/AUTHORS (revision 257) +++ /tags/1.2.5/AUTHORS (revision 257) @@ -0,0 +1,2 @@ +Anatoly Vorobey +Brad Fitzpatrick Index: /tags/1.2.5/scripts/memcached-tool =================================================================== --- /tags/1.2.5/scripts/memcached-tool (revision 628) +++ /tags/1.2.5/scripts/memcached-tool (revision 628) @@ -0,0 +1,186 @@ +#!/usr/bin/perl +# +# memcached-tool: +# stats/management tool for memcached. +# +# Author: +# Brad Fitzpatrick +# +# License: +# public domain. I give up all rights to this +# tool. modify and copy at will. +# + +use strict; +use IO::Socket::INET; + +my $host = shift; +my $mode = shift || "display"; +my ($from, $to); + +if ($mode eq "display") { + undef $mode if @ARGV; +} elsif ($mode eq "move") { + $from = shift; + $to = shift; + undef $mode if $from < 6 || $from > 17; + undef $mode if $to < 6 || $to > 17; + print STDERR "ERROR: parameters out of range\n\n" unless $mode; +} elsif ($mode eq 'dump') { + ; +} elsif ($mode eq 'stats') { + ; +} else { + undef $mode; +} + +undef $mode if @ARGV; + +die +"Usage: memcached-tool [mode]\n + memcached-tool 10.0.0.5:11211 display # shows slabs + memcached-tool 10.0.0.5:11211 # same. (default is display) + memcached-tool 10.0.0.5:11211 stats # shows general stats + memcached-tool 10.0.0.5:11211 move 7 9 # takes 1MB slab from class #7 + # to class #9. + +You can only move slabs around once memory is totally allocated, and only +once the target class is full. (So you can't move from #6 to #9 and #7 +to #9 at the same itme, since you'd have to wait for #9 to fill from +the first reassigned page) +" unless $host && $mode; + +$host .= ":11211" unless $host =~ /:\d+/; + +my $sock = IO::Socket::INET->new(PeerAddr => $host, + Proto => 'tcp'); +die "Couldn't connect to $host\n" unless $sock; + + +if ($mode eq "move") { + my $tries = 0; + while (1) { + print $sock "slabs reassign $from $to\r\n"; + my $res = <$sock>; + $res =~ s/\s+//; + if ($res eq "DONE") { + print "Success.\n"; + exit 0; + } elsif ($res eq "CANT") { + print "Error: can't move from $from to $to. Destination not yet full? See usage docs.\n"; + exit; + } elsif ($res eq "BUSY") { + if (++$tries == 3) { + print "Failed to move after 3 tries. Try again later.\n"; + exit; + } + + print "Page busy, retrying...\n"; + sleep 1; + } + } + + exit; +} + +if ($mode eq 'dump') { + my %items; + my $totalitems; + + print $sock "stats items\r\n"; + + while (<$sock>) { + last if /^END/; + if (/^STAT items:(\d*):number (\d*)/) { + $items{$1} = $2; + $totalitems += $2; + } + } + print STDERR "Dumping memcache contents\n"; + print STDERR " Number of buckets: " . scalar(keys(%items)) . "\n"; + print STDERR " Number of items : $totalitems\n"; + + foreach my $bucket (sort(keys(%items))) { + print STDERR "Dumping bucket $bucket - " . $items{$bucket} . " total items\n"; + print $sock "stats cachedump $bucket $items{$bucket} 1\r\n"; + my %keyexp; + while (<$sock>) { + last if /^END/; + # return format looks like this + # ITEM foo [6 b; 1176415152 s] + if (/^ITEM (\S+) \[.* (\d+) s\]/) { + $keyexp{$1} = $2; + } + } + + foreach my $k (keys(%keyexp)) { + my $val; + print $sock "get $k\r\n"; + my $response = <$sock>; + $response =~ /VALUE (\S+) (\d+) (\d+)/; + my $flags = $2; + my $len = $3; + read $sock, $val , $len; + # get the END + $_ = <$sock>; + $_ = <$sock>; + print "add $k $flags $keyexp{$k} $len\r\n$val\r\n"; + } + } + exit; +} + +if ($mode eq 'stats') { + my %items; + + print $sock "stats\r\n"; + + while (<$sock>) { + last if /^END/; + chomp; + if (/^STAT\s+(\S*)\s+(.*)/) { + $items{$1} = $2; + } + } + printf ("#%-17s %5s %11s\n", $host, "Field", "Value"); + foreach my $name (sort(keys(%items))) { + printf ("%24s %12s\n", $name, $items{$name}); + + } + exit; +} + +# display mode: + +my %items; # class -> { number, age, chunk_size, chunks_per_page, + # total_pages, total_chunks, used_chunks, + # free_chunks, free_chunks_end } + +print $sock "stats items\r\n"; +while (<$sock>) { + last if /^END/; + if (/^STAT items:(\d+):(\w+) (\d+)/) { + $items{$1}{$2} = $3; + } +} + +print $sock "stats slabs\r\n"; +while (<$sock>) { + last if /^END/; + if (/^STAT (\d+):(\w+) (\d+)/) { + $items{$1}{$2} = $3; + } +} + +print " # Item_Size Max_age 1MB_pages Count Full?\n"; +foreach my $n (1..40) { + my $it = $items{$n}; + next if (0 == $it->{total_pages}); + my $size = $it->{chunk_size} < 1024 ? "$it->{chunk_size} B " : + sprintf("%.1f kB", $it->{chunk_size} / 1024.0); + my $full = $it->{free_chunks_end} == 0 ? "yes" : " no"; + printf "%3d %8s %7d s %7d %7d %7s\n", + $n, $size, $it->{age}, $it->{total_pages}, + $it->{number}, $full; +} + Index: /tags/1.2.5/scripts/start-memcached =================================================================== --- /tags/1.2.5/scripts/start-memcached (revision 201) +++ /tags/1.2.5/scripts/start-memcached (revision 201) @@ -0,0 +1,117 @@ +#!/usr/bin/perl -w + +# start-memcached +# 2003/2004 - Jay Bonci +# This script handles the parsing of the /etc/memcached.conf file +# and was originally created for the Debian distribution. +# Anyone may use this little script under the same terms as +# memcached itself. + +use strict; + +if($> != 0 and $< != 0) +{ + print STDERR "Only root wants to run start-memcached.\n"; + exit; +} + +my $params; my $etchandle; my $etcfile = "/etc/memcached.conf"; + +# This script assumes that memcached is located at /usr/bin/memcached, and +# that the pidfile is writable at /var/run/memcached.pid + +my $memcached = "/usr/bin/memcached"; +my $pidfile = "/var/run/memcached.pid"; + +# If we don't get a valid logfile parameter in the /etc/memcached.conf file, +# we'll just throw away all of our in-daemon output. We need to re-tie it so +# that non-bash shells will not hang on logout. Thanks to Michael Renner for +# the tip +my $fd_reopened = "/dev/null"; + + sub handle_logfile + { + my ($logfile) = @_; + $fd_reopened = $logfile; + } + + sub reopen_logfile + { + my ($logfile) = @_; + + open *STDERR, ">>$logfile"; + open *STDOUT, ">>$logfile"; + open *STDIN, ">>/dev/null"; + $fd_reopened = $logfile; + } + +# This is set up in place here to support other non -[a-z] directives + +my $conf_directives = { + "logfile" => \&handle_logfile, +}; + +if(open $etchandle, $etcfile) +{ + foreach my $line (<$etchandle>) + { + $line ||= ""; + $line =~ s/\#.*//g; + $line =~ s/\s+$//g; + $line =~ s/^\s+//g; + next unless $line; + next if $line =~ /^\-[dh]/; + + if($line =~ /^[^\-]/) + { + my ($directive, $arg) = $line =~ /^(.*?)\s+(.*)/; + $conf_directives->{$directive}->($arg); + next; + } + + push @$params, $line; + } + +}else{ + $params = []; +} + + push @$params, "-u root" unless(grep "-u", @$params); + $params = join " ", @$params; + +if(-e $pidfile) +{ + open PIDHANDLE, "$pidfile"; + my $localpid = ; + close PIDHANDLE; + + chomp $localpid; + if(-d "/proc/$localpid") + { + print STDERR "memcached is already running.\n"; + exit; + }else{ + `rm -f $localpid`; + } + +} + +my $pid = fork(); + +if($pid == 0) +{ + reopen_logfile($fd_reopened); + exec "$memcached $params"; + exit(0); + +}else{ + if(open PIDHANDLE,">$pidfile") + { + print PIDHANDLE $pid; + close PIDHANDLE; + }else{ + + print STDERR "Can't write pidfile to $pidfile.\n"; + } +} + Index: /tags/1.2.5/scripts/memcached-init =================================================================== --- /tags/1.2.5/scripts/memcached-init (revision 176) +++ /tags/1.2.5/scripts/memcached-init (revision 176) @@ -0,0 +1,59 @@ +#! /bin/sh +# +# skeleton example file to build /etc/init.d/ scripts. +# This file should be used to construct scripts for /etc/init.d. +# +# Written by Miquel van Smoorenburg . +# Modified for Debian +# by Ian Murdock . +# +# Version: @(#)skeleton 1.9 26-Feb-2001 miquels@cistron.nl +# + +PATH=/usr/local/sbin:/usr/local/bin:/sbin:/bin:/usr/sbin:/usr/bin +DAEMON=/usr/bin/memcached +DAEMONBOOTSTRAP=/usr/share/memcached/scripts/start-memcached +NAME=memcached +DESC=memcached +PIDFILE=/var/run/$NAME.pid + +test -x $DAEMON || exit 0 +test -x $DAEMONBOOTSTRAP || exit 0 + +set -e + +case "$1" in + start) + echo -n "Starting $DESC: " + start-stop-daemon --start --quiet --exec $DAEMONBOOTSTRAP + echo "$NAME." + ;; + stop) + echo -n "Stopping $DESC: " + start-stop-daemon --stop --quiet --oknodo --pidfile $PIDFILE --exec $DAEMON + echo "$NAME." + rm -f $PIDFILE + ;; + + restart|force-reload) + # + # If the "reload" option is implemented, move the "force-reload" + # option to the "reload" entry above. If not, "force-reload" is + # just the same as "restart". + # + echo -n "Restarting $DESC: " + start-stop-daemon --stop --quiet --oknodo --pidfile $PIDFILE + rm -f $PIDFILE + sleep 1 + start-stop-daemon --start --quiet --exec $DAEMONBOOTSTRAP + echo "$NAME." + ;; + *) + N=/etc/init.d/$NAME + # echo "Usage: $N {start|stop|restart|reload|force-reload}" >&2 + echo "Usage: $N {start|stop|restart|force-reload}" >&2 + exit 1 + ;; +esac + +exit 0 Index: /tags/1.2.5/scripts/memcached.sysv =================================================================== --- /tags/1.2.5/scripts/memcached.sysv (revision 578) +++ /tags/1.2.5/scripts/memcached.sysv (revision 578) @@ -0,0 +1,78 @@ +#! /bin/sh +# +# chkconfig: - 55 45 +# description: The memcached daemon is a network memory cache service. +# processname: memcached +# config: /etc/sysconfig/memcached + +# Source function library. +. /etc/rc.d/init.d/functions + +PORT=11211 +USER=nobody +MAXCONN=1024 +CACHESIZE=64 +OPTIONS="" + +if [ -f /etc/sysconfig/memcached ];then + . /etc/sysconfig/memcached +fi + +# Check that networking is up. +if [ "$NETWORKING" = "no" ] +then + exit 0 +fi + +RETVAL=0 +prog="memcached" + +start () { + echo -n $"Starting $prog: " + # insure that /var/run/memcached has proper permissions + chown $USER /var/run/memcached + daemon memcached -d -p $PORT -u $USER -m $CACHESIZE -c $MAXCONN -P /var/run/memcached/memcached.pid $OPTIONS + RETVAL=$? + echo + [ $RETVAL -eq 0 ] && touch /var/lock/subsys/memcached +} +stop () { + echo -n $"Stopping $prog: " + killproc memcached + RETVAL=$? + echo + if [ $RETVAL -eq 0 ] ; then + rm -f /var/lock/subsys/memcached + rm -f /var/run/memcached.pid + fi +} + +restart () { + stop + start +} + + +# See how we were called. +case "$1" in + start) + start + ;; + stop) + stop + ;; + status) + status memcached + ;; + restart|reload) + restart + ;; + condrestart) + [ -f /var/lock/subsys/memcached ] && restart || : + ;; + *) + echo $"Usage: $0 {start|stop|status|restart|reload|condrestart}" + exit 1 +esac + +exit $? Index: /tags/1.2.5/memcached.c =================================================================== --- /tags/1.2.5/memcached.c (revision 736) +++ /tags/1.2.5/memcached.c (revision 736) @@ -0,0 +1,3143 @@ +/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */ +/* + * memcached - memory caching daemon + * + * http://www.danga.com/memcached/ + * + * Copyright 2003 Danga Interactive, Inc. All rights reserved. + * + * Use and distribution licensed under the BSD license. See + * the LICENSE file for full text. + * + * Authors: + * Anatoly Vorobey + * Brad Fitzpatrick +std * + * $Id$ + */ +#include "memcached.h" +#include +#include +#include +#include +#include +#include + +/* some POSIX systems need the following definition + * to get mlockall flags out of sys/mman.h. */ +#ifndef _P1003_1B_VISIBLE +#define _P1003_1B_VISIBLE +#endif +/* need this to get IOV_MAX on some platforms. */ +#ifndef __need_IOV_MAX +#define __need_IOV_MAX +#endif +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#ifdef HAVE_MALLOC_H +/* OpenBSD has a malloc.h, but warns to use stdlib.h instead */ +#ifndef __OpenBSD__ +#include +#endif +#endif + +/* FreeBSD 4.x doesn't have IOV_MAX exposed. */ +#ifndef IOV_MAX +#if defined(__FreeBSD__) || defined(__APPLE__) +# define IOV_MAX 1024 +#endif +#endif + +/* + * forward declarations + */ +static void drive_machine(conn *c); +static int new_socket(struct addrinfo *ai); +static int server_socket(const int port, const bool is_udp); +static int try_read_command(conn *c); +static int try_read_network(conn *c); +static int try_read_udp(conn *c); + +/* stats */ +static void stats_reset(void); +static void stats_init(void); + +/* defaults */ +static void settings_init(void); + +/* event handling, network IO */ +static void event_handler(const int fd, const short which, void *arg); +static void conn_close(conn *c); +static void conn_init(void); +static void accept_new_conns(const bool do_accept); +static bool update_event(conn *c, const int new_flags); +static void complete_nread(conn *c); +static void process_command(conn *c, char *command); +static int transmit(conn *c); +static int ensure_iov_space(conn *c); +static int add_iov(conn *c, const void *buf, int len); +static int add_msghdr(conn *c); + +/* time handling */ +static void set_current_time(void); /* update the global variable holding + global 32-bit seconds-since-start time + (to avoid 64 bit time_t) */ + +static void conn_free(conn *c); + +/** exported globals **/ +struct stats stats; +struct settings settings; + +/** file scope variables **/ +static item **todelete = NULL; +static int delcurr; +static int deltotal; +static conn *listen_conn = NULL; +static struct event_base *main_base; + +#define TRANSMIT_COMPLETE 0 +#define TRANSMIT_INCOMPLETE 1 +#define TRANSMIT_SOFT_ERROR 2 +#define TRANSMIT_HARD_ERROR 3 + +static int *buckets = 0; /* bucket->generation array for a managed instance */ + +#define REALTIME_MAXDELTA 60*60*24*30 +/* + * given time value that's either unix time or delta from current unix time, return + * unix time. Use the fact that delta can't exceed one month (and real time value can't + * be that low). + */ +static rel_time_t realtime(const time_t exptime) { + /* no. of seconds in 30 days - largest possible delta exptime */ + + if (exptime == 0) return 0; /* 0 means never expire */ + + if (exptime > REALTIME_MAXDELTA) { + /* if item expiration is at/before the server started, give it an + expiration time of 1 second after the server started. + (because 0 means don't expire). without this, we'd + underflow and wrap around to some large value way in the + future, effectively making items expiring in the past + really expiring never */ + if (exptime <= stats.started) + return (rel_time_t)1; + return (rel_time_t)(exptime - stats.started); + } else { + return (rel_time_t)(exptime + current_time); + } +} + +static void stats_init(void) { + stats.curr_items = stats.total_items = stats.curr_conns = stats.total_conns = stats.conn_structs = 0; + stats.get_cmds = stats.set_cmds = stats.get_hits = stats.get_misses = stats.evictions = 0; + stats.curr_bytes = stats.bytes_read = stats.bytes_written = 0; + + /* make the time we started always be 2 seconds before we really + did, so time(0) - time.started is never zero. if so, things + like 'settings.oldest_live' which act as booleans as well as + values are now false in boolean context... */ + stats.started = time(0) - 2; + stats_prefix_init(); +} + +static void stats_reset(void) { + STATS_LOCK(); + stats.total_items = stats.total_conns = 0; + stats.get_cmds = stats.set_cmds = stats.get_hits = stats.get_misses = stats.evictions = 0; + stats.bytes_read = stats.bytes_written = 0; + stats_prefix_clear(); + STATS_UNLOCK(); +} + +static void settings_init(void) { + settings.access=0700; + settings.port = 11211; + settings.udpport = 0; + /* By default this string should be NULL for getaddrinfo() */ + settings.inter = NULL; + settings.maxbytes = 64 * 1024 * 1024; /* default is 64MB */ + settings.maxconns = 1024; /* to limit connections-related memory to about 5MB */ + settings.verbose = 0; + settings.oldest_live = 0; + settings.evict_to_free = 1; /* push old items out of cache when memory runs out */ + settings.socketpath = NULL; /* by default, not using a unix socket */ + settings.managed = false; + settings.factor = 1.25; + settings.chunk_size = 48; /* space for a modest key and value */ +#ifdef USE_THREADS + settings.num_threads = 4; +#else + settings.num_threads = 1; +#endif + settings.prefix_delimiter = ':'; + settings.detail_enabled = 0; +} + +/* returns true if a deleted item's delete-locked-time is over, and it + should be removed from the namespace */ +static bool item_delete_lock_over (item *it) { + assert(it->it_flags & ITEM_DELETED); + return (current_time >= it->exptime); +} + +/* + * Adds a message header to a connection. + * + * Returns 0 on success, -1 on out-of-memory. + */ +static int add_msghdr(conn *c) +{ + struct msghdr *msg; + + assert(c != NULL); + + if (c->msgsize == c->msgused) { + msg = realloc(c->msglist, c->msgsize * 2 * sizeof(struct msghdr)); + if (! msg) + return -1; + c->msglist = msg; + c->msgsize *= 2; + } + + msg = c->msglist + c->msgused; + + /* this wipes msg_iovlen, msg_control, msg_controllen, and + msg_flags, the last 3 of which aren't defined on solaris: */ + memset(msg, 0, sizeof(struct msghdr)); + + msg->msg_iov = &c->iov[c->iovused]; + + if (c->request_addr_size > 0) { + msg->msg_name = &c->request_addr; + msg->msg_namelen = c->request_addr_size; + } + + c->msgbytes = 0; + c->msgused++; + + if (c->udp) { + /* Leave room for the UDP header, which we'll fill in later. */ + return add_iov(c, NULL, UDP_HEADER_SIZE); + } + + return 0; +} + + +/* + * Free list management for connections. + */ + +static conn **freeconns; +static int freetotal; +static int freecurr; + + +static void conn_init(void) { + freetotal = 200; + freecurr = 0; + if ((freeconns = (conn **)malloc(sizeof(conn *) * freetotal)) == NULL) { + fprintf(stderr, "malloc()\n"); + } + return; +} + +/* + * Returns a connection from the freelist, if any. Should call this using + * conn_from_freelist() for thread safety. + */ +conn *do_conn_from_freelist() { + conn *c; + + if (freecurr > 0) { + c = freeconns[--freecurr]; + } else { + c = NULL; + } + + return c; +} + +/* + * Adds a connection to the freelist. 0 = success. Should call this using + * conn_add_to_freelist() for thread safety. + */ +bool do_conn_add_to_freelist(conn *c) { + if (freecurr < freetotal) { + freeconns[freecurr++] = c; + return false; + } else { + /* try to enlarge free connections array */ + conn **new_freeconns = realloc(freeconns, sizeof(conn *) * freetotal * 2); + if (new_freeconns) { + freetotal *= 2; + freeconns = new_freeconns; + freeconns[freecurr++] = c; + return false; + } + } + return true; +} + +conn *conn_new(const int sfd, const int init_state, const int event_flags, + const int read_buffer_size, const bool is_udp, struct event_base *base) { + conn *c = conn_from_freelist(); + + if (NULL == c) { + if (!(c = (conn *)malloc(sizeof(conn)))) { + fprintf(stderr, "malloc()\n"); + return NULL; + } + c->rbuf = c->wbuf = 0; + c->ilist = 0; + c->suffixlist = 0; + c->iov = 0; + c->msglist = 0; + c->hdrbuf = 0; + + c->rsize = read_buffer_size; + c->wsize = DATA_BUFFER_SIZE; + c->isize = ITEM_LIST_INITIAL; + c->suffixsize = SUFFIX_LIST_INITIAL; + c->iovsize = IOV_LIST_INITIAL; + c->msgsize = MSG_LIST_INITIAL; + c->hdrsize = 0; + + c->rbuf = (char *)malloc((size_t)c->rsize); + c->wbuf = (char *)malloc((size_t)c->wsize); + c->ilist = (item **)malloc(sizeof(item *) * c->isize); + c->suffixlist = (char **)malloc(sizeof(char *) * c->suffixsize); + c->iov = (struct iovec *)malloc(sizeof(struct iovec) * c->iovsize); + c->msglist = (struct msghdr *)malloc(sizeof(struct msghdr) * c->msgsize); + + if (c->rbuf == 0 || c->wbuf == 0 || c->ilist == 0 || c->iov == 0 || + c->msglist == 0 || c->suffixlist == 0) { + if (c->rbuf != 0) free(c->rbuf); + if (c->wbuf != 0) free(c->wbuf); + if (c->ilist !=0) free(c->ilist); + if (c->suffixlist != 0) free(c->suffixlist); + if (c->iov != 0) free(c->iov); + if (c->msglist != 0) free(c->msglist); + free(c); + fprintf(stderr, "malloc()\n"); + return NULL; + } + + STATS_LOCK(); + stats.conn_structs++; + STATS_UNLOCK(); + } + + if (settings.verbose > 1) { + if (init_state == conn_listening) + fprintf(stderr, "<%d server listening\n", sfd); + else if (is_udp) + fprintf(stderr, "<%d server listening (udp)\n", sfd); + else + fprintf(stderr, "<%d new client connection\n", sfd); + } + + c->sfd = sfd; + c->udp = is_udp; + c->state = init_state; + c->rlbytes = 0; + c->rbytes = c->wbytes = 0; + c->wcurr = c->wbuf; + c->rcurr = c->rbuf; + c->ritem = 0; + c->icurr = c->ilist; + c->suffixcurr = c->suffixlist; + c->ileft = 0; + c->suffixleft = 0; + c->iovused = 0; + c->msgcurr = 0; + c->msgused = 0; + + c->write_and_go = conn_read; + c->write_and_free = 0; + c->item = 0; + c->bucket = -1; + c->gen = 0; + + c->noreply = false; + + event_set(&c->event, sfd, event_flags, event_handler, (void *)c); + event_base_set(base, &c->event); + c->ev_flags = event_flags; + + if (event_add(&c->event, 0) == -1) { + if (conn_add_to_freelist(c)) { + conn_free(c); + } + perror("event_add"); + return NULL; + } + + STATS_LOCK(); + stats.curr_conns++; + stats.total_conns++; + STATS_UNLOCK(); + + return c; +} + +static void conn_cleanup(conn *c) { + assert(c != NULL); + + if (c->item) { + item_remove(c->item); + c->item = 0; + } + + if (c->ileft != 0) { + for (; c->ileft > 0; c->ileft--,c->icurr++) { + item_remove(*(c->icurr)); + } + } + + if (c->suffixleft != 0) { + for (; c->suffixleft > 0; c->suffixleft--, c->suffixcurr++) { + if(suffix_add_to_freelist(*(c->suffixcurr))) { + free(*(c->suffixcurr)); + } + } + } + + if (c->write_and_free) { + free(c->write_and_free); + c->write_and_free = 0; + } +} + +/* + * Frees a connection. + */ +void conn_free(conn *c) { + if (c) { + if (c->hdrbuf) + free(c->hdrbuf); + if (c->msglist) + free(c->msglist); + if (c->rbuf) + free(c->rbuf); + if (c->wbuf) + free(c->wbuf); + if (c->ilist) + free(c->ilist); + if (c->suffixlist) + free(c->suffixlist); + if (c->iov) + free(c->iov); + free(c); + } +} + +static void conn_close(conn *c) { + assert(c != NULL); + + /* delete the event, the socket and the conn */ + event_del(&c->event); + + if (settings.verbose > 1) + fprintf(stderr, "<%d connection closed.\n", c->sfd); + + close(c->sfd); + accept_new_conns(true); + conn_cleanup(c); + + /* if the connection has big buffers, just free it */ + if (c->rsize > READ_BUFFER_HIGHWAT || conn_add_to_freelist(c)) { + conn_free(c); + } + + STATS_LOCK(); + stats.curr_conns--; + STATS_UNLOCK(); + + return; +} + + +/* + * Shrinks a connection's buffers if they're too big. This prevents + * periodic large "get" requests from permanently chewing lots of server + * memory. + * + * This should only be called in between requests since it can wipe output + * buffers! + */ +static void conn_shrink(conn *c) { + assert(c != NULL); + + if (c->udp) + return; + + if (c->rsize > READ_BUFFER_HIGHWAT && c->rbytes < DATA_BUFFER_SIZE) { + char *newbuf; + + if (c->rcurr != c->rbuf) + memmove(c->rbuf, c->rcurr, (size_t)c->rbytes); + + newbuf = (char *)realloc((void *)c->rbuf, DATA_BUFFER_SIZE); + + if (newbuf) { + c->rbuf = newbuf; + c->rsize = DATA_BUFFER_SIZE; + } + /* TODO check other branch... */ + c->rcurr = c->rbuf; + } + + if (c->isize > ITEM_LIST_HIGHWAT) { + item **newbuf = (item**) realloc((void *)c->ilist, ITEM_LIST_INITIAL * sizeof(c->ilist[0])); + if (newbuf) { + c->ilist = newbuf; + c->isize = ITEM_LIST_INITIAL; + } + /* TODO check error condition? */ + } + + if (c->msgsize > MSG_LIST_HIGHWAT) { + struct msghdr *newbuf = (struct msghdr *) realloc((void *)c->msglist, MSG_LIST_INITIAL * sizeof(c->msglist[0])); + if (newbuf) { + c->msglist = newbuf; + c->msgsize = MSG_LIST_INITIAL; + } + /* TODO check error condition? */ + } + + if (c->iovsize > IOV_LIST_HIGHWAT) { + struct iovec *newbuf = (struct iovec *) realloc((void *)c->iov, IOV_LIST_INITIAL * sizeof(c->iov[0])); + if (newbuf) { + c->iov = newbuf; + c->iovsize = IOV_LIST_INITIAL; + } + /* TODO check return value */ + } +} + +/* + * Sets a connection's current state in the state machine. Any special + * processing that needs to happen on certain state transitions can + * happen here. + */ +static void conn_set_state(conn *c, int state) { + assert(c != NULL); + + if (state != c->state) { + if (state == conn_read) { + conn_shrink(c); + assoc_move_next_bucket(); + } + c->state = state; + } +} + +/* + * Free list management for suffix buffers. + */ + +static char **freesuffix; +static int freesuffixtotal; +static int freesuffixcurr; + +static void suffix_init(void) { + freesuffixtotal = 500; + freesuffixcurr = 0; + + freesuffix = (char **)malloc( sizeof(char *) * freesuffixtotal ); + if (freesuffix == NULL) { + fprintf(stderr, "malloc()\n"); + } + return; +} + +/* + * Returns a suffix buffer from the freelist, if any. Should call this using + * suffix_from_freelist() for thread safety. + */ +char *do_suffix_from_freelist() { + char *s; + + if (freesuffixcurr > 0) { + s = freesuffix[--freesuffixcurr]; + } else { + /* If malloc fails, let the logic fall through without spamming + * STDERR on the server. */ + s = malloc( SUFFIX_SIZE ); + } + + return s; +} + +/* + * Adds a connection to the freelist. 0 = success. Should call this using + * conn_add_to_freelist() for thread safety. + */ +bool do_suffix_add_to_freelist(char *s) { + if (freesuffixcurr < freesuffixtotal) { + freesuffix[freesuffixcurr++] = s; + return false; + } else { + /* try to enlarge free connections array */ + char **new_freesuffix = realloc(freesuffix, freesuffixtotal * 2); + if (new_freesuffix) { + freesuffixtotal *= 2; + freesuffix = new_freesuffix; + freesuffix[freesuffixcurr++] = s; + return false; + } + } + return true; +} + +/* + * Ensures that there is room for another struct iovec in a connection's + * iov list. + * + * Returns 0 on success, -1 on out-of-memory. + */ +static int ensure_iov_space(conn *c) { + assert(c != NULL); + + if (c->iovused >= c->iovsize) { + int i, iovnum; + struct iovec *new_iov = (struct iovec *)realloc(c->iov, + (c->iovsize * 2) * sizeof(struct iovec)); + if (! new_iov) + return -1; + c->iov = new_iov; + c->iovsize *= 2; + + /* Point all the msghdr structures at the new list. */ + for (i = 0, iovnum = 0; i < c->msgused; i++) { + c->msglist[i].msg_iov = &c->iov[iovnum]; + iovnum += c->msglist[i].msg_iovlen; + } + } + + return 0; +} + + +/* + * Adds data to the list of pending data that will be written out to a + * connection. + * + * Returns 0 on success, -1 on out-of-memory. + */ + +static int add_iov(conn *c, const void *buf, int len) { + struct msghdr *m; + int leftover; + bool limit_to_mtu; + + assert(c != NULL); + + do { + m = &c->msglist[c->msgused - 1]; + + /* + * Limit UDP packets, and the first payloads of TCP replies, to + * UDP_MAX_PAYLOAD_SIZE bytes. + */ + limit_to_mtu = c->udp || (1 == c->msgused); + + /* We may need to start a new msghdr if this one is full. */ + if (m->msg_iovlen == IOV_MAX || + (limit_to_mtu && c->msgbytes >= UDP_MAX_PAYLOAD_SIZE)) { + add_msghdr(c); + m = &c->msglist[c->msgused - 1]; + } + + if (ensure_iov_space(c) != 0) + return -1; + + /* If the fragment is too big to fit in the datagram, split it up */ + if (limit_to_mtu && len + c->msgbytes > UDP_MAX_PAYLOAD_SIZE) { + leftover = len + c->msgbytes - UDP_MAX_PAYLOAD_SIZE; + len -= leftover; + } else { + leftover = 0; + } + + m = &c->msglist[c->msgused - 1]; + m->msg_iov[m->msg_iovlen].iov_base = (void *)buf; + m->msg_iov[m->msg_iovlen].iov_len = len; + + c->msgbytes += len; + c->iovused++; + m->msg_iovlen++; + + buf = ((char *)buf) + len; + len = leftover; + } while (leftover > 0); + + return 0; +} + + +/* + * Constructs a set of UDP headers and attaches them to the outgoing messages. + */ +static int build_udp_headers(conn *c) { + int i; + unsigned char *hdr; + + assert(c != NULL); + + if (c->msgused > c->hdrsize) { + void *new_hdrbuf; + if (c->hdrbuf) + new_hdrbuf = realloc(c->hdrbuf, c->msgused * 2 * UDP_HEADER_SIZE); + else + new_hdrbuf = malloc(c->msgused * 2 * UDP_HEADER_SIZE); + if (! new_hdrbuf) + return -1; + c->hdrbuf = (unsigned char *)new_hdrbuf; + c->hdrsize = c->msgused * 2; + } + + hdr = c->hdrbuf; + for (i = 0; i < c->msgused; i++) { + c->msglist[i].msg_iov[0].iov_base = hdr; + c->msglist[i].msg_iov[0].iov_len = UDP_HEADER_SIZE; + *hdr++ = c->request_id / 256; + *hdr++ = c->request_id % 256; + *hdr++ = i / 256; + *hdr++ = i % 256; + *hdr++ = c->msgused / 256; + *hdr++ = c->msgused % 256; + *hdr++ = 0; + *hdr++ = 0; + assert((void *) hdr == (void *)c->msglist[i].msg_iov[0].iov_base + UDP_HEADER_SIZE); + } + + return 0; +} + + +static void out_string(conn *c, const char *str) { + size_t len; + + assert(c != NULL); + + if (c->noreply) { + if (settings.verbose > 1) + fprintf(stderr, ">%d NOREPLY %s\n", c->sfd, str); + c->noreply = false; + conn_set_state(c, conn_read); + return; + } + + if (settings.verbose > 1) + fprintf(stderr, ">%d %s\n", c->sfd, str); + + len = strlen(str); + if ((len + 2) > c->wsize) { + /* ought to be always enough. just fail for simplicity */ + str = "SERVER_ERROR output line too long"; + len = strlen(str); + } + + memcpy(c->wbuf, str, len); + memcpy(c->wbuf + len, "\r\n", 3); + c->wbytes = len + 2; + c->wcurr = c->wbuf; + + conn_set_state(c, conn_write); + c->write_and_go = conn_read; + return; +} + +/* + * we get here after reading the value in set/add/replace commands. The command + * has been stored in c->item_comm, and the item is ready in c->item. + */ + +static void complete_nread(conn *c) { + assert(c != NULL); + + item *it = c->item; + int comm = c->item_comm; + int ret; + + STATS_LOCK(); + stats.set_cmds++; + STATS_UNLOCK(); + + if (strncmp(ITEM_data(it) + it->nbytes - 2, "\r\n", 2) != 0) { + out_string(c, "CLIENT_ERROR bad data chunk"); + } else { + ret = store_item(it, comm); + if (ret == 1) + out_string(c, "STORED"); + else if(ret == 2) + out_string(c, "EXISTS"); + else if(ret == 3) + out_string(c, "NOT_FOUND"); + else + out_string(c, "NOT_STORED"); + } + + item_remove(c->item); /* release the c->item reference */ + c->item = 0; +} + +/* + * Stores an item in the cache according to the semantics of one of the set + * commands. In threaded mode, this is protected by the cache lock. + * + * Returns true if the item was stored. + */ +int do_store_item(item *it, int comm) { + char *key = ITEM_key(it); + bool delete_locked = false; + item *old_it = do_item_get_notedeleted(key, it->nkey, &delete_locked); + int stored = 0; + + item *new_it = NULL; + int flags; + + if (old_it != NULL && comm == NREAD_ADD) { + /* add only adds a nonexistent item, but promote to head of LRU */ + do_item_update(old_it); + } else if (!old_it && (comm == NREAD_REPLACE + || comm == NREAD_APPEND || comm == NREAD_PREPEND)) + { + /* replace only replaces an existing value; don't store */ + } else if (delete_locked && (comm == NREAD_REPLACE || comm == NREAD_ADD + || comm == NREAD_APPEND || comm == NREAD_PREPEND)) + { + /* replace and add can't override delete locks; don't store */ + } else if (comm == NREAD_CAS) { + /* validate cas operation */ + if (delete_locked) + old_it = do_item_get_nocheck(key, it->nkey); + + if(old_it == NULL) { + // LRU expired + stored = 3; + } + else if(it->cas_id == old_it->cas_id) { + // cas validates + do_item_replace(old_it, it); + stored = 1; + } + else + { + stored = 2; + } + } else { + /* + * Append - combine new and old record into single one. Here it's + * atomic and thread-safe. + */ + + if (comm == NREAD_APPEND || comm == NREAD_PREPEND) { + + /* we have it and old_it here - alloc memory to hold both */ + /* flags was already lost - so recover them from ITEM_suffix(it) */ + + flags = (int) strtol(ITEM_suffix(old_it), (char **) NULL, 10); + + new_it = do_item_alloc(key, it->nkey, flags, old_it->exptime, it->nbytes + old_it->nbytes - 2 /* CRLF */); + + if (new_it == NULL) { + /* SERVER_ERROR out of memory */ + return 0; + } + + /* copy data from it and old_it to new_it */ + + if (comm == NREAD_APPEND) { + memcpy(ITEM_data(new_it), ITEM_data(old_it), old_it->nbytes); + memcpy(ITEM_data(new_it) + old_it->nbytes - 2 /* CRLF */, ITEM_data(it), it->nbytes); + } else { + /* NREAD_PREPEND */ + memcpy(ITEM_data(new_it), ITEM_data(it), it->nbytes); + memcpy(ITEM_data(new_it) + it->nbytes - 2 /* CRLF */, ITEM_data(old_it), old_it->nbytes); + } + + it = new_it; + } + + /* "set" commands can override the delete lock + window... in which case we have to find the old hidden item + that's in the namespace/LRU but wasn't returned by + item_get.... because we need to replace it */ + if (delete_locked) + old_it = do_item_get_nocheck(key, it->nkey); + + if (old_it != NULL) + do_item_replace(old_it, it); + else + do_item_link(it); + + stored = 1; + } + + if (old_it != NULL) + do_item_remove(old_it); /* release our reference */ + if (new_it != NULL) + do_item_remove(new_it); + + return stored; +} + +typedef struct token_s { + char *value; + size_t length; +} token_t; + +#define COMMAND_TOKEN 0 +#define SUBCOMMAND_TOKEN 1 +#define KEY_TOKEN 1 +#define KEY_MAX_LENGTH 250 + +#define MAX_TOKENS 8 + +/* + * Tokenize the command string by replacing whitespace with '\0' and update + * the token array tokens with pointer to start of each token and length. + * Returns total number of tokens. The last valid token is the terminal + * token (value points to the first unprocessed character of the string and + * length zero). + * + * Usage example: + * + * while(tokenize_command(command, ncommand, tokens, max_tokens) > 0) { + * for(int ix = 0; tokens[ix].length != 0; ix++) { + * ... + * } + * ncommand = tokens[ix].value - command; + * command = tokens[ix].value; + * } + */ +static size_t tokenize_command(char *command, token_t *tokens, const size_t max_tokens) { + char *s, *e; + size_t ntokens = 0; + + assert(command != NULL && tokens != NULL && max_tokens > 1); + + for (s = e = command; ntokens < max_tokens - 1; ++e) { + if (*e == ' ') { + if (s != e) { + tokens[ntokens].value = s; + tokens[ntokens].length = e - s; + ntokens++; + *e = '\0'; + } + s = e + 1; + } + else if (*e == '\0') { + if (s != e) { + tokens[ntokens].value = s; + tokens[ntokens].length = e - s; + ntokens++; + } + + break; /* string end */ + } + } + + /* + * If we scanned the whole string, the terminal value pointer is null, + * otherwise it is the first unprocessed character. + */ + tokens[ntokens].value = *e == '\0' ? NULL : e; + tokens[ntokens].length = 0; + ntokens++; + + return ntokens; +} + +/* set up a connection to write a buffer then free it, used for stats */ +static void write_and_free(conn *c, char *buf, int bytes) { + if (buf) { + c->write_and_free = buf; + c->wcurr = buf; + c->wbytes = bytes; + conn_set_state(c, conn_write); + c->write_and_go = conn_read; + } else { + out_string(c, "SERVER_ERROR out of memory writing stats"); + } +} + +static inline void set_noreply_maybe(conn *c, token_t *tokens, size_t ntokens) +{ + int noreply_index = ntokens - 2; + + /* + NOTE: this function is not the first place where we are going to + send the reply. We could send it instead from process_command() + if the request line has wrong number of tokens. However parsing + malformed line for "noreply" option is not reliable anyway, so + it can't be helped. + */ + if (tokens[noreply_index].value + && strcmp(tokens[noreply_index].value, "noreply") == 0) { + c->noreply = true; + } +} + +inline static void process_stats_detail(conn *c, const char *command) { + assert(c != NULL); + + if (strcmp(command, "on") == 0) { + settings.detail_enabled = 1; + out_string(c, "OK"); + } + else if (strcmp(command, "off") == 0) { + settings.detail_enabled = 0; + out_string(c, "OK"); + } + else if (strcmp(command, "dump") == 0) { + int len; + char *stats = stats_prefix_dump(&len); + write_and_free(c, stats, len); + } + else { + out_string(c, "CLIENT_ERROR usage: stats detail on|off|dump"); + } +} + +static void process_stat(conn *c, token_t *tokens, const size_t ntokens) { + rel_time_t now = current_time; + char *command; + char *subcommand; + + assert(c != NULL); + + if(ntokens < 2) { + out_string(c, "CLIENT_ERROR bad command line"); + return; + } + + command = tokens[COMMAND_TOKEN].value; + + if (ntokens == 2 && strcmp(command, "stats") == 0) { + char temp[1024]; + pid_t pid = getpid(); + char *pos = temp; + +#ifndef WIN32 + struct rusage usage; + getrusage(RUSAGE_SELF, &usage); +#endif /* !WIN32 */ + + STATS_LOCK(); + pos += sprintf(pos, "STAT pid %u\r\n", pid); + pos += sprintf(pos, "STAT uptime %u\r\n", now); + pos += sprintf(pos, "STAT time %ld\r\n", now + stats.started); + pos += sprintf(pos, "STAT version " VERSION "\r\n"); + pos += sprintf(pos, "STAT pointer_size %d\r\n", 8 * sizeof(void *)); +#ifndef WIN32 + pos += sprintf(pos, "STAT rusage_user %ld.%06ld\r\n", usage.ru_utime.tv_sec, usage.ru_utime.tv_usec); + pos += sprintf(pos, "STAT rusage_system %ld.%06ld\r\n", usage.ru_stime.tv_sec, usage.ru_stime.tv_usec); +#endif /* !WIN32 */ + pos += sprintf(pos, "STAT curr_items %u\r\n", stats.curr_items); + pos += sprintf(pos, "STAT total_items %u\r\n", stats.total_items); + pos += sprintf(pos, "STAT bytes %llu\r\n", stats.curr_bytes); + pos += sprintf(pos, "STAT curr_connections %u\r\n", stats.curr_conns - 1); /* ignore listening conn */ + pos += sprintf(pos, "STAT total_connections %u\r\n", stats.total_conns); + pos += sprintf(pos, "STAT connection_structures %u\r\n", stats.conn_structs); + pos += sprintf(pos, "STAT cmd_get %llu\r\n", stats.get_cmds); + pos += sprintf(pos, "STAT cmd_set %llu\r\n", stats.set_cmds); + pos += sprintf(pos, "STAT get_hits %llu\r\n", stats.get_hits); + pos += sprintf(pos, "STAT get_misses %llu\r\n", stats.get_misses); + pos += sprintf(pos, "STAT evictions %llu\r\n", stats.evictions); + pos += sprintf(pos, "STAT bytes_read %llu\r\n", stats.bytes_read); + pos += sprintf(pos, "STAT bytes_written %llu\r\n", stats.bytes_written); + pos += sprintf(pos, "STAT limit_maxbytes %llu\r\n", (uint64_t) settings.maxbytes); + pos += sprintf(pos, "STAT threads %u\r\n", settings.num_threads); + pos += sprintf(pos, "END"); + STATS_UNLOCK(); + out_string(c, temp); + return; + } + + subcommand = tokens[SUBCOMMAND_TOKEN].value; + + if (strcmp(subcommand, "reset") == 0) { + stats_reset(); + out_string(c, "RESET"); + return; + } + +#ifdef HAVE_MALLOC_H +#ifdef HAVE_STRUCT_MALLINFO + if (strcmp(subcommand, "malloc") == 0) { + char temp[512]; + struct mallinfo info; + char *pos = temp; + + info = mallinfo(); + pos += sprintf(pos, "STAT arena_size %d\r\n", info.arena); + pos += sprintf(pos, "STAT free_chunks %d\r\n", info.ordblks); + pos += sprintf(pos, "STAT fastbin_blocks %d\r\n", info.smblks); + pos += sprintf(pos, "STAT mmapped_regions %d\r\n", info.hblks); + pos += sprintf(pos, "STAT mmapped_space %d\r\n", info.hblkhd); + pos += sprintf(pos, "STAT max_total_alloc %d\r\n", info.usmblks); + pos += sprintf(pos, "STAT fastbin_space %d\r\n", info.fsmblks); + pos += sprintf(pos, "STAT total_alloc %d\r\n", info.uordblks); + pos += sprintf(pos, "STAT total_free %d\r\n", info.fordblks); + pos += sprintf(pos, "STAT releasable_space %d\r\nEND", info.keepcost); + out_string(c, temp); + return; + } +#endif /* HAVE_STRUCT_MALLINFO */ +#endif /* HAVE_MALLOC_H */ + +#if !defined(WIN32) || !defined(__APPLE__) + if (strcmp(subcommand, "maps") == 0) { + char *wbuf; + int wsize = 8192; /* should be enough */ + int fd; + int res; + + if ((wbuf = (char *)malloc(wsize)) == NULL) { + out_string(c, "SERVER_ERROR out of memory writing stats maps"); + return; + } + + fd = open("/proc/self/maps", O_RDONLY); + if (fd == -1) { + out_string(c, "SERVER_ERROR cannot open the maps file"); + free(wbuf); + return; + } + + res = read(fd, wbuf, wsize - 6); /* 6 = END\r\n\0 */ + if (res == wsize - 6) { + out_string(c, "SERVER_ERROR buffer overflow"); + free(wbuf); close(fd); + return; + } + if (res == 0 || res == -1) { + out_string(c, "SERVER_ERROR can't read the maps file"); + free(wbuf); close(fd); + return; + } + memcpy(wbuf + res, "END\r\n", 5); + write_and_free(c, wbuf, res + 5); + close(fd); + return; + } +#endif + + if (strcmp(subcommand, "cachedump") == 0) { + + char *buf; + unsigned int bytes, id, limit = 0; + + if(ntokens < 5) { + out_string(c, "CLIENT_ERROR bad command line"); + return; + } + + id = strtoul(tokens[2].value, NULL, 10); + limit = strtoul(tokens[3].value, NULL, 10); + + if(errno == ERANGE) { + out_string(c, "CLIENT_ERROR bad command line format"); + return; + } + + buf = item_cachedump(id, limit, &bytes); + write_and_free(c, buf, bytes); + return; + } + + if (strcmp(subcommand, "slabs") == 0) { + int bytes = 0; + char *buf = slabs_stats(&bytes); + write_and_free(c, buf, bytes); + return; + } + + if (strcmp(subcommand, "items") == 0) { + int bytes = 0; + char *buf = item_stats(&bytes); + write_and_free(c, buf, bytes); + return; + } + + if (strcmp(subcommand, "detail") == 0) { + if (ntokens < 4) + process_stats_detail(c, ""); /* outputs the error message */ + else + process_stats_detail(c, tokens[2].value); + return; + } + + if (strcmp(subcommand, "sizes") == 0) { + int bytes = 0; + char *buf = item_stats_sizes(&bytes); + write_and_free(c, buf, bytes); + return; + } + + out_string(c, "ERROR"); +} + +/* ntokens is overwritten here... shrug.. */ +static inline void process_get_command(conn *c, token_t *tokens, size_t ntokens, bool return_cas) { + char *key; + size_t nkey; + int i = 0; + item *it; + token_t *key_token = &tokens[KEY_TOKEN]; + char *suffix; + int stats_get_cmds = 0; + int stats_get_hits = 0; + int stats_get_misses = 0; + assert(c != NULL); + + if (settings.managed) { + int bucket = c->bucket; + if (bucket == -1) { + out_string(c, "CLIENT_ERROR no BG data in managed mode"); + return; + } + c->bucket = -1; + if (buckets[bucket] != c->gen) { + out_string(c, "ERROR_NOT_OWNER"); + return; + } + } + + do { + while(key_token->length != 0) { + + key = key_token->value; + nkey = key_token->length; + + if(nkey > KEY_MAX_LENGTH) { + STATS_LOCK(); + stats.get_cmds += stats_get_cmds; + stats.get_hits += stats_get_hits; + stats.get_misses += stats_get_misses; + STATS_UNLOCK(); + out_string(c, "CLIENT_ERROR bad command line format"); + return; + } + + stats_get_cmds++; + it = item_get(key, nkey); + if (settings.detail_enabled) { + stats_prefix_record_get(key, NULL != it); + } + if (it) { + if (i >= c->isize) { + item **new_list = realloc(c->ilist, sizeof(item *) * c->isize * 2); + if (new_list) { + c->isize *= 2; + c->ilist = new_list; + } else break; + } + + /* + * Construct the response. Each hit adds three elements to the + * outgoing data list: + * "VALUE " + * key + * " " + flags + " " + data length + "\r\n" + data (with \r\n) + */ + + if(return_cas == true) + { + /* Goofy mid-flight realloc. */ + if (i >= c->suffixsize) { + char **new_suffix_list = realloc(c->suffixlist, + sizeof(char *) * c->suffixsize * 2); + if (new_suffix_list) { + c->suffixsize *= 2; + c->suffixlist = new_suffix_list; + } else break; + } + + suffix = suffix_from_freelist(); + if (suffix == NULL) { + STATS_LOCK(); + stats.get_cmds += stats_get_cmds; + stats.get_hits += stats_get_hits; + stats.get_misses += stats_get_misses; + STATS_UNLOCK(); + out_string(c, "SERVER_ERROR out of memory making CAS suffix"); + return; + } + *(c->suffixlist + i) = suffix; + sprintf(suffix, " %llu\r\n", it->cas_id); + if (add_iov(c, "VALUE ", 6) != 0 || + add_iov(c, ITEM_key(it), it->nkey) != 0 || + add_iov(c, ITEM_suffix(it), it->nsuffix - 2) != 0 || + add_iov(c, suffix, strlen(suffix)) != 0 || + add_iov(c, ITEM_data(it), it->nbytes) != 0) + { + break; + } + } + else + { + if (add_iov(c, "VALUE ", 6) != 0 || + add_iov(c, ITEM_key(it), it->nkey) != 0 || + add_iov(c, ITEM_suffix(it), it->nsuffix + it->nbytes) != 0) + { + break; + } + } + + + if (settings.verbose > 1) + fprintf(stderr, ">%d sending key %s\n", c->sfd, ITEM_key(it)); + + /* item_get() has incremented it->refcount for us */ + stats_get_hits++; + item_update(it); + *(c->ilist + i) = it; + i++; + + } else { + stats_get_misses++; + } + + key_token++; + } + + /* + * If the command string hasn't been fully processed, get the next set + * of tokens. + */ + if(key_token->value != NULL) { + ntokens = tokenize_command(key_token->value, tokens, MAX_TOKENS); + key_token = tokens; + } + + } while(key_token->value != NULL); + + c->icurr = c->ilist; + c->ileft = i; + if (return_cas) { + c->suffixcurr = c->suffixlist; + c->suffixleft = i; + } + + if (settings.verbose > 1) + fprintf(stderr, ">%d END\n", c->sfd); + + /* + If the loop was terminated because of out-of-memory, it is not + reliable to add END\r\n to the buffer, because it might not end + in \r\n. So we send SERVER_ERROR instead. + */ + if (key_token->value != NULL || add_iov(c, "END\r\n", 5) != 0 + || (c->udp && build_udp_headers(c) != 0)) { + out_string(c, "SERVER_ERROR out of memory writing get response"); + } + else { + conn_set_state(c, conn_mwrite); + c->msgcurr = 0; + } + + STATS_LOCK(); + stats.get_cmds += stats_get_cmds; + stats.get_hits += stats_get_hits; + stats.get_misses += stats_get_misses; + STATS_UNLOCK(); + + return; +} + +static void process_update_command(conn *c, token_t *tokens, const size_t ntokens, int comm, bool handle_cas) { + char *key; + size_t nkey; + int flags; + time_t exptime; + int vlen, old_vlen; + uint64_t req_cas_id; + item *it, *old_it; + + assert(c != NULL); + + set_noreply_maybe(c, tokens, ntokens); + + if (tokens[KEY_TOKEN].length > KEY_MAX_LENGTH) { + out_string(c, "CLIENT_ERROR bad command line format"); + return; + } + + key = tokens[KEY_TOKEN].value; + nkey = tokens[KEY_TOKEN].length; + + flags = strtoul(tokens[2].value, NULL, 10); + exptime = strtol(tokens[3].value, NULL, 10); + vlen = strtol(tokens[4].value, NULL, 10); + + // does cas value exist? + if(handle_cas) + { + req_cas_id = strtoull(tokens[5].value, NULL, 10); + } + + if(errno == ERANGE || ((flags == 0 || exptime == 0) && errno == EINVAL)) { + out_string(c, "CLIENT_ERROR bad command line format"); + return; + } + + if (settings.detail_enabled) { + stats_prefix_record_set(key); + } + + if (settings.managed) { + int bucket = c->bucket; + if (bucket == -1) { + out_string(c, "CLIENT_ERROR no BG data in managed mode"); + return; + } + c->bucket = -1; + if (buckets[bucket] != c->gen) { + out_string(c, "ERROR_NOT_OWNER"); + return; + } + } + + it = item_alloc(key, nkey, flags, realtime(exptime), vlen+2); + + if (it == 0) { + if (! item_size_ok(nkey, flags, vlen + 2)) + out_string(c, "SERVER_ERROR object too large for cache"); + else + out_string(c, "SERVER_ERROR out of memory storing object"); + /* swallow the data line */ + c->write_and_go = conn_swallow; + c->sbytes = vlen + 2; + return; + } + if(handle_cas) + it->cas_id = req_cas_id; + + c->item = it; + c->ritem = ITEM_data(it); + c->rlbytes = it->nbytes; + c->item_comm = comm; + conn_set_state(c, conn_nread); +} + +static void process_arithmetic_command(conn *c, token_t *tokens, const size_t ntokens, const bool incr) { + char temp[sizeof("18446744073709551615")]; + item *it; + int64_t delta; + char *key; + size_t nkey; + + assert(c != NULL); + + set_noreply_maybe(c, tokens, ntokens); + + if(tokens[KEY_TOKEN].length > KEY_MAX_LENGTH) { + out_string(c, "CLIENT_ERROR bad command line format"); + return; + } + + key = tokens[KEY_TOKEN].value; + nkey = tokens[KEY_TOKEN].length; + + if (settings.managed) { + int bucket = c->bucket; + if (bucket == -1) { + out_string(c, "CLIENT_ERROR no BG data in managed mode"); + return; + } + c->bucket = -1; + if (buckets[bucket] != c->gen) { + out_string(c, "ERROR_NOT_OWNER"); + return; + } + } + + delta = strtoll(tokens[2].value, NULL, 10); + + if(errno == ERANGE) { + out_string(c, "CLIENT_ERROR bad command line format"); + return; + } + + it = item_get(key, nkey); + if (!it) { + out_string(c, "NOT_FOUND"); + return; + } + + out_string(c, add_delta(it, incr, delta, temp)); + item_remove(it); /* release our reference */ +} + +/* + * adds a delta value to a numeric item. + * + * it item to adjust + * incr true to increment value, false to decrement + * delta amount to adjust value by + * buf buffer for response string + * + * returns a response string to send back to the client. + */ +char *do_add_delta(item *it, const bool incr, const int64_t delta, char *buf) { + char *ptr; + int64_t value; + int res; + + ptr = ITEM_data(it); + while ((*ptr != '\0') && (*ptr < '0' && *ptr > '9')) ptr++; // BUG: can't be true + + value = strtoull(ptr, NULL, 10); + + if(errno == ERANGE) { + return "CLIENT_ERROR cannot increment or decrement non-numeric value"; + } + + if (incr) + value += delta; + else { + if (delta >= value) value = 0; + else value -= delta; + } + sprintf(buf, "%llu", value); + res = strlen(buf); + if (res + 2 > it->nbytes) { /* need to realloc */ + item *new_it; + new_it = do_item_alloc(ITEM_key(it), it->nkey, atoi(ITEM_suffix(it) + 1), it->exptime, res + 2 ); + if (new_it == 0) { + return "SERVER_ERROR out of memory in incr/decr"; + } + memcpy(ITEM_data(new_it), buf, res); + memcpy(ITEM_data(new_it) + res, "\r\n", 3); + do_item_replace(it, new_it); + do_item_remove(new_it); /* release our reference */ + } else { /* replace in-place */ + memcpy(ITEM_data(it), buf, res); + memset(ITEM_data(it) + res, ' ', it->nbytes - res - 2); + } + + return buf; +} + +static void process_delete_command(conn *c, token_t *tokens, const size_t ntokens) { + char *key; + size_t nkey; + item *it; + time_t exptime = 0; + + assert(c != NULL); + + set_noreply_maybe(c, tokens, ntokens); + + if (settings.managed) { + int bucket = c->bucket; + if (bucket == -1) { + out_string(c, "CLIENT_ERROR no BG data in managed mode"); + return; + } + c->bucket = -1; + if (buckets[bucket] != c->gen) { + out_string(c, "ERROR_NOT_OWNER"); + return; + } + } + + key = tokens[KEY_TOKEN].value; + nkey = tokens[KEY_TOKEN].length; + + if(nkey > KEY_MAX_LENGTH) { + out_string(c, "CLIENT_ERROR bad command line format"); + return; + } + + if(ntokens == (c->noreply ? 5 : 4)) { + exptime = strtol(tokens[2].value, NULL, 10); + + if(errno == ERANGE) { + out_string(c, "CLIENT_ERROR bad command line format"); + return; + } + } + + if (settings.detail_enabled) { + stats_prefix_record_delete(key); + } + + it = item_get(key, nkey); + if (it) { + if (exptime == 0) { + item_unlink(it); + item_remove(it); /* release our reference */ + out_string(c, "DELETED"); + } else { + /* our reference will be transfered to the delete queue */ + out_string(c, defer_delete(it, exptime)); + } + } else { + out_string(c, "NOT_FOUND"); + } +} + +/* + * Adds an item to the deferred-delete list so it can be reaped later. + * + * Returns the result to send to the client. + */ +char *do_defer_delete(item *it, time_t exptime) +{ + if (delcurr >= deltotal) { + item **new_delete = realloc(todelete, sizeof(item *) * deltotal * 2); + if (new_delete) { + todelete = new_delete; + deltotal *= 2; + } else { + /* + * can't delete it immediately, user wants a delay, + * but we ran out of memory for the delete queue + */ + item_remove(it); /* release reference */ + return "SERVER_ERROR out of memory expanding delete queue"; + } + } + + /* use its expiration time as its deletion time now */ + it->exptime = realtime(exptime); + it->it_flags |= ITEM_DELETED; + todelete[delcurr++] = it; + + return "DELETED"; +} + +static void process_verbosity_command(conn *c, token_t *tokens, const size_t ntokens) { + unsigned int level; + + assert(c != NULL); + + set_noreply_maybe(c, tokens, ntokens); + + level = strtoul(tokens[1].value, NULL, 10); + settings.verbose = level > MAX_VERBOSITY_LEVEL ? MAX_VERBOSITY_LEVEL : level; + out_string(c, "OK"); + return; +} + +static void process_command(conn *c, char *command) { + + token_t tokens[MAX_TOKENS]; + size_t ntokens; + int comm; + + assert(c != NULL); + + if (settings.verbose > 1) + fprintf(stderr, "<%d %s\n", c->sfd, command); + + /* + * for commands set/add/replace, we build an item and read the data + * directly into it, then continue in nread_complete(). + */ + + c->msgcurr = 0; + c->msgused = 0; + c->iovused = 0; + if (add_msghdr(c) != 0) { + out_string(c, "SERVER_ERROR out of memory preparing response"); + return; + } + + ntokens = tokenize_command(command, tokens, MAX_TOKENS); + if (ntokens >= 3 && + ((strcmp(tokens[COMMAND_TOKEN].value, "get") == 0) || + (strcmp(tokens[COMMAND_TOKEN].value, "bget") == 0))) { + + process_get_command(c, tokens, ntokens, false); + + } else if ((ntokens == 6 || ntokens == 7) && + ((strcmp(tokens[COMMAND_TOKEN].value, "add") == 0 && (comm = NREAD_ADD)) || + (strcmp(tokens[COMMAND_TOKEN].value, "set") == 0 && (comm = NREAD_SET)) || + (strcmp(tokens[COMMAND_TOKEN].value, "replace") == 0 && (comm = NREAD_REPLACE)) || + (strcmp(tokens[COMMAND_TOKEN].value, "prepend") == 0 && (comm = NREAD_PREPEND)) || + (strcmp(tokens[COMMAND_TOKEN].value, "append") == 0 && (comm = NREAD_APPEND)) )) { + + process_update_command(c, tokens, ntokens, comm, false); + + } else if ((ntokens == 7 || ntokens == 8) && (strcmp(tokens[COMMAND_TOKEN].value, "cas") == 0 && (comm = NREAD_CAS))) { + + process_update_command(c, tokens, ntokens, comm, true); + + } else if ((ntokens == 4 || ntokens == 5) && (strcmp(tokens[COMMAND_TOKEN].value, "incr") == 0)) { + + process_arithmetic_command(c, tokens, ntokens, 1); + + } else if (ntokens >= 3 && (strcmp(tokens[COMMAND_TOKEN].value, "gets") == 0)) { + + process_get_command(c, tokens, ntokens, true); + + } else if ((ntokens == 4 || ntokens == 5) && (strcmp(tokens[COMMAND_TOKEN].value, "decr") == 0)) { + + process_arithmetic_command(c, tokens, ntokens, 0); + + } else if (ntokens >= 3 && ntokens <= 5 && (strcmp(tokens[COMMAND_TOKEN].value, "delete") == 0)) { + + process_delete_command(c, tokens, ntokens); + + } else if (ntokens == 3 && strcmp(tokens[COMMAND_TOKEN].value, "own") == 0) { + unsigned int bucket, gen; + if (!settings.managed) { + out_string(c, "CLIENT_ERROR not a managed instance"); + return; + } + + if (sscanf(tokens[1].value, "%u:%u", &bucket,&gen) == 2) { + if ((bucket < 0) || (bucket >= MAX_BUCKETS)) { + out_string(c, "CLIENT_ERROR bucket number out of range"); + return; + } + buckets[bucket] = gen; + out_string(c, "OWNED"); + return; + } else { + out_string(c, "CLIENT_ERROR bad format"); + return; + } + + } else if (ntokens == 3 && (strcmp(tokens[COMMAND_TOKEN].value, "disown")) == 0) { + + int bucket; + if (!settings.managed) { + out_string(c, "CLIENT_ERROR not a managed instance"); + return; + } + if (sscanf(tokens[1].value, "%u", &bucket) == 1) { + if ((bucket < 0) || (bucket >= MAX_BUCKETS)) { + out_string(c, "CLIENT_ERROR bucket number out of range"); + return; + } + buckets[bucket] = 0; + out_string(c, "DISOWNED"); + return; + } else { + out_string(c, "CLIENT_ERROR bad format"); + return; + } + + } else if (ntokens == 3 && (strcmp(tokens[COMMAND_TOKEN].value, "bg")) == 0) { + int bucket, gen; + if (!settings.managed) { + out_string(c, "CLIENT_ERROR not a managed instance"); + return; + } + if (sscanf(tokens[1].value, "%u:%u", &bucket, &gen) == 2) { + /* we never write anything back, even if input's wrong */ + if ((bucket < 0) || (bucket >= MAX_BUCKETS) || (gen <= 0)) { + /* do nothing, bad input */ + } else { + c->bucket = bucket; + c->gen = gen; + } + conn_set_state(c, conn_read); + return; + } else { + out_string(c, "CLIENT_ERROR bad format"); + return; + } + + } else if (ntokens >= 2 && (strcmp(tokens[COMMAND_TOKEN].value, "stats") == 0)) { + + process_stat(c, tokens, ntokens); + + } else if (ntokens >= 2 && ntokens <= 4 && (strcmp(tokens[COMMAND_TOKEN].value, "flush_all") == 0)) { + time_t exptime = 0; + set_current_time(); + + set_noreply_maybe(c, tokens, ntokens); + + if(ntokens == (c->noreply ? 3 : 2)) { + settings.oldest_live = current_time - 1; + item_flush_expired(); + out_string(c, "OK"); + return; + } + + exptime = strtol(tokens[1].value, NULL, 10); + if(errno == ERANGE) { + out_string(c, "CLIENT_ERROR bad command line format"); + return; + } + + /* + If exptime is zero realtime() would return zero too, and + realtime(exptime) - 1 would overflow to the max unsigned + value. So we process exptime == 0 the same way we do when + no delay is given at all. + */ + if (exptime > 0) + settings.oldest_live = realtime(exptime) - 1; + else /* exptime == 0 */ + settings.oldest_live = current_time - 1; + item_flush_expired(); + out_string(c, "OK"); + return; + + } else if (ntokens == 2 && (strcmp(tokens[COMMAND_TOKEN].value, "version") == 0)) { + + out_string(c, "VERSION " VERSION); + + } else if (ntokens == 2 && (strcmp(tokens[COMMAND_TOKEN].value, "quit") == 0)) { + + conn_set_state(c, conn_closing); + + } else if (ntokens == 5 && (strcmp(tokens[COMMAND_TOKEN].value, "slabs") == 0 && + strcmp(tokens[COMMAND_TOKEN + 1].value, "reassign") == 0)) { +#ifdef ALLOW_SLABS_REASSIGN + + int src, dst, rv; + + src = strtol(tokens[2].value, NULL, 10); + dst = strtol(tokens[3].value, NULL, 10); + + if(errno == ERANGE) { + out_string(c, "CLIENT_ERROR bad command line format"); + return; + } + + rv = slabs_reassign(src, dst); + if (rv == 1) { + out_string(c, "DONE"); + return; + } + if (rv == 0) { + out_string(c, "CANT"); + return; + } + if (rv == -1) { + out_string(c, "BUSY"); + return; + } +#else + out_string(c, "CLIENT_ERROR Slab reassignment not supported"); +#endif + } else if ((ntokens == 3 || ntokens == 4) && (strcmp(tokens[COMMAND_TOKEN].value, "verbosity") == 0)) { + process_verbosity_command(c, tokens, ntokens); + } else { + out_string(c, "ERROR"); + } + return; +} + +/* + * if we have a complete line in the buffer, process it. + */ +static int try_read_command(conn *c) { + char *el, *cont; + + assert(c != NULL); + assert(c->rcurr <= (c->rbuf + c->rsize)); + + if (c->rbytes == 0) + return 0; + el = memchr(c->rcurr, '\n', c->rbytes); + if (!el) + return 0; + cont = el + 1; + if ((el - c->rcurr) > 1 && *(el - 1) == '\r') { + el--; + } + *el = '\0'; + + assert(cont <= (c->rcurr + c->rbytes)); + + process_command(c, c->rcurr); + + c->rbytes -= (cont - c->rcurr); + c->rcurr = cont; + + assert(c->rcurr <= (c->rbuf + c->rsize)); + + return 1; +} + +/* + * read a UDP request. + * return 0 if there's nothing to read. + */ +static int try_read_udp(conn *c) { + int res; + + assert(c != NULL); + + c->request_addr_size = sizeof(c->request_addr); + res = recvfrom(c->sfd, c->rbuf, c->rsize, + 0, &c->request_addr, &c->request_addr_size); + if (res > 8) { + unsigned char *buf = (unsigned char *)c->rbuf; + STATS_LOCK(); + stats.bytes_read += res; + STATS_UNLOCK(); + + /* Beginning of UDP packet is the request ID; save it. */ + c->request_id = buf[0] * 256 + buf[1]; + + /* If this is a multi-packet request, drop it. */ + if (buf[4] != 0 || buf[5] != 1) { + out_string(c, "SERVER_ERROR multi-packet request not supported"); + return 0; + } + + /* Don't care about any of the rest of the header. */ + res -= 8; + memmove(c->rbuf, c->rbuf + 8, res); + + c->rbytes += res; + c->rcurr = c->rbuf; + return 1; + } + return 0; +} + +/* + * read from network as much as we can, handle buffer overflow and connection + * close. + * before reading, move the remaining incomplete fragment of a command + * (if any) to the beginning of the buffer. + * return 0 if there's nothing to read on the first read. + */ +static int try_read_network(conn *c) { + int gotdata = 0; + int res; + + assert(c != NULL); + + if (c->rcurr != c->rbuf) { + if (c->rbytes != 0) /* otherwise there's nothing to copy */ + memmove(c->rbuf, c->rcurr, c->rbytes); + c->rcurr = c->rbuf; + } + + while (1) { + if (c->rbytes >= c->rsize) { + char *new_rbuf = realloc(c->rbuf, c->rsize * 2); + if (!new_rbuf) { + if (settings.verbose > 0) + fprintf(stderr, "Couldn't realloc input buffer\n"); + c->rbytes = 0; /* ignore what we read */ + out_string(c, "SERVER_ERROR out of memory reading request"); + c->write_and_go = conn_closing; + return 1; + } + c->rcurr = c->rbuf = new_rbuf; + c->rsize *= 2; + } + + /* unix socket mode doesn't need this, so zeroed out. but why + * is this done for every command? presumably for UDP + * mode. */ + if (!settings.socketpath) { + c->request_addr_size = sizeof(c->request_addr); + } else { + c->request_addr_size = 0; + } + + int avail = c->rsize - c->rbytes; + res = read(c->sfd, c->rbuf + c->rbytes, avail); + if (res > 0) { + STATS_LOCK(); + stats.bytes_read += res; + STATS_UNLOCK(); + gotdata = 1; + c->rbytes += res; + if (res == avail) { + continue; + } else { + break; + } + } + if (res == 0) { + /* connection closed */ + conn_set_state(c, conn_closing); + return 1; + } + if (res == -1) { + if (errno == EAGAIN || errno == EWOULDBLOCK) break; + /* Should close on unhandled errors. */ + conn_set_state(c, conn_closing); + return 1; + } + } + return gotdata; +} + +static bool update_event(conn *c, const int new_flags) { + assert(c != NULL); + + struct event_base *base = c->event.ev_base; + if (c->ev_flags == new_flags) + return true; + if (event_del(&c->event) == -1) return false; + event_set(&c->event, c->sfd, new_flags, event_handler, (void *)c); + event_base_set(base, &c->event); + c->ev_flags = new_flags; + if (event_add(&c->event, 0) == -1) return false; + return true; +} + +/* + * Sets whether we are listening for new connections or not. + */ +void accept_new_conns(const bool do_accept) { + conn *next; + + if (! is_listen_thread()) + return; + + for (next = listen_conn; next; next = next->next) { + if (do_accept) { + update_event(next, EV_READ | EV_PERSIST); + if (listen(next->sfd, 1024) != 0) { + perror("listen"); + } + } + else { + update_event(next, 0); + if (listen(next->sfd, 0) != 0) { + perror("listen"); + } + } + } +} + + +/* + * Transmit the next chunk of data from our list of msgbuf structures. + * + * Returns: + * TRANSMIT_COMPLETE All done writing. + * TRANSMIT_INCOMPLETE More data remaining to write. + * TRANSMIT_SOFT_ERROR Can't write any more right now. + * TRANSMIT_HARD_ERROR Can't write (c->state is set to conn_closing) + */ +static int transmit(conn *c) { + assert(c != NULL); + + if (c->msgcurr < c->msgused && + c->msglist[c->msgcurr].msg_iovlen == 0) { + /* Finished writing the current msg; advance to the next. */ + c->msgcurr++; + } + if (c->msgcurr < c->msgused) { + ssize_t res; + struct msghdr *m = &c->msglist[c->msgcurr]; + + res = sendmsg(c->sfd, m, 0); + if (res > 0) { + STATS_LOCK(); + stats.bytes_written += res; + STATS_UNLOCK(); + + /* We've written some of the data. Remove the completed + iovec entries from the list of pending writes. */ + while (m->msg_iovlen > 0 && res >= m->msg_iov->iov_len) { + res -= m->msg_iov->iov_len; + m->msg_iovlen--; + m->msg_iov++; + } + + /* Might have written just part of the last iovec entry; + adjust it so the next write will do the rest. */ + if (res > 0) { + m->msg_iov->iov_base += res; + m->msg_iov->iov_len -= res; + } + return TRANSMIT_INCOMPLETE; + } + if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) { + if (!update_event(c, EV_WRITE | EV_PERSIST)) { + if (settings.verbose > 0) + fprintf(stderr, "Couldn't update event\n"); + conn_set_state(c, conn_closing); + return TRANSMIT_HARD_ERROR; + } + return TRANSMIT_SOFT_ERROR; + } + /* if res==0 or res==-1 and error is not EAGAIN or EWOULDBLOCK, + we have a real error, on which we close the connection */ + if (settings.verbose > 0) + perror("Failed to write, and not due to blocking"); + + if (c->udp) + conn_set_state(c, conn_read); + else + conn_set_state(c, conn_closing); + return TRANSMIT_HARD_ERROR; + } else { + return TRANSMIT_COMPLETE; + } +} + +static void drive_machine(conn *c) { + bool stop = false; + int sfd, flags = 1; + socklen_t addrlen; + struct sockaddr_storage addr; + int res; + + assert(c != NULL); + + while (!stop) { + + switch(c->state) { + case conn_listening: + addrlen = sizeof(addr); + if ((sfd = accept(c->sfd, (struct sockaddr *)&addr, &addrlen)) == -1) { + if (errno == EAGAIN || errno == EWOULDBLOCK) { + /* these are transient, so don't log anything */ + stop = true; + } else if (errno == EMFILE) { + if (settings.verbose > 0) + fprintf(stderr, "Too many open connections\n"); + accept_new_conns(false); + stop = true; + } else { + perror("accept()"); + stop = true; + } + break; + } + if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || + fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) { + perror("setting O_NONBLOCK"); + close(sfd); + break; + } + dispatch_conn_new(sfd, conn_read, EV_READ | EV_PERSIST, + DATA_BUFFER_SIZE, false); + break; + + case conn_read: + if (try_read_command(c) != 0) { + continue; + } + if ((c->udp ? try_read_udp(c) : try_read_network(c)) != 0) { + continue; + } + /* we have no command line and no data to read from network */ + if (!update_event(c, EV_READ | EV_PERSIST)) { + if (settings.verbose > 0) + fprintf(stderr, "Couldn't update event\n"); + conn_set_state(c, conn_closing); + break; + } + stop = true; + break; + + case conn_nread: + /* we are reading rlbytes into ritem; */ + if (c->rlbytes == 0) { + complete_nread(c); + break; + } + /* first check if we have leftovers in the conn_read buffer */ + if (c->rbytes > 0) { + int tocopy = c->rbytes > c->rlbytes ? c->rlbytes : c->rbytes; + memcpy(c->ritem, c->rcurr, tocopy); + c->ritem += tocopy; + c->rlbytes -= tocopy; + c->rcurr += tocopy; + c->rbytes -= tocopy; + break; + } + + /* now try reading from the socket */ + res = read(c->sfd, c->ritem, c->rlbytes); + if (res > 0) { + STATS_LOCK(); + stats.bytes_read += res; + STATS_UNLOCK(); + c->ritem += res; + c->rlbytes -= res; + break; + } + if (res == 0) { /* end of stream */ + conn_set_state(c, conn_closing); + break; + } + if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) { + if (!update_event(c, EV_READ | EV_PERSIST)) { + if (settings.verbose > 0) + fprintf(stderr, "Couldn't update event\n"); + conn_set_state(c, conn_closing); + break; + } + stop = true; + break; + } + /* otherwise we have a real error, on which we close the connection */ + if (settings.verbose > 0) + fprintf(stderr, "Failed to read, and not due to blocking\n"); + conn_set_state(c, conn_closing); + break; + + case conn_swallow: + /* we are reading sbytes and throwing them away */ + if (c->sbytes == 0) { + conn_set_state(c, conn_read); + break; + } + + /* first check if we have leftovers in the conn_read buffer */ + if (c->rbytes > 0) { + int tocopy = c->rbytes > c->sbytes ? c->sbytes : c->rbytes; + c->sbytes -= tocopy; + c->rcurr += tocopy; + c->rbytes -= tocopy; + break; + } + + /* now try reading from the socket */ + res = read(c->sfd, c->rbuf, c->rsize > c->sbytes ? c->sbytes : c->rsize); + if (res > 0) { + STATS_LOCK(); + stats.bytes_read += res; + STATS_UNLOCK(); + c->sbytes -= res; + break; + } + if (res == 0) { /* end of stream */ + conn_set_state(c, conn_closing); + break; + } + if (res == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) { + if (!update_event(c, EV_READ | EV_PERSIST)) { + if (settings.verbose > 0) + fprintf(stderr, "Couldn't update event\n"); + conn_set_state(c, conn_closing); + break; + } + stop = true; + break; + } + /* otherwise we have a real error, on which we close the connection */ + if (settings.verbose > 0) + fprintf(stderr, "Failed to read, and not due to blocking\n"); + conn_set_state(c, conn_closing); + break; + + case conn_write: + /* + * We want to write out a simple response. If we haven't already, + * assemble it into a msgbuf list (this will be a single-entry + * list for TCP or a two-entry list for UDP). + */ + if (c->iovused == 0 || (c->udp && c->iovused == 1)) { + if (add_iov(c, c->wcurr, c->wbytes) != 0 || + (c->udp && build_udp_headers(c) != 0)) { + if (settings.verbose > 0) + fprintf(stderr, "Couldn't build response\n"); + conn_set_state(c, conn_closing); + break; + } + } + + /* fall through... */ + + case conn_mwrite: + switch (transmit(c)) { + case TRANSMIT_COMPLETE: + if (c->state == conn_mwrite) { + while (c->ileft > 0) { + item *it = *(c->icurr); + assert((it->it_flags & ITEM_SLABBED) == 0); + item_remove(it); + c->icurr++; + c->ileft--; + } + while (c->suffixleft > 0) { + char *suffix = *(c->suffixcurr); + if(suffix_add_to_freelist(suffix)) { + /* Failed to add to freelist, don't leak */ + free(suffix); + } + c->suffixcurr++; + c->suffixleft--; + } + conn_set_state(c, conn_read); + } else if (c->state == conn_write) { + if (c->write_and_free) { + free(c->write_and_free); + c->write_and_free = 0; + } + conn_set_state(c, c->write_and_go); + } else { + if (settings.verbose > 0) + fprintf(stderr, "Unexpected state %d\n", c->state); + conn_set_state(c, conn_closing); + } + break; + + case TRANSMIT_INCOMPLETE: + case TRANSMIT_HARD_ERROR: + break; /* Continue in state machine. */ + + case TRANSMIT_SOFT_ERROR: + stop = true; + break; + } + break; + + case conn_closing: + if (c->udp) + conn_cleanup(c); + else + conn_close(c); + stop = true; + break; + } + } + + return; +} + +void event_handler(const int fd, const short which, void *arg) { + conn *c; + + c = (conn *)arg; + assert(c != NULL); + + c->which = which; + + /* sanity */ + if (fd != c->sfd) { + if (settings.verbose > 0) + fprintf(stderr, "Catastrophic: event fd doesn't match conn fd!\n"); + conn_close(c); + return; + } + + drive_machine(c); + + /* wait for next event */ + return; +} + +static int new_socket(struct addrinfo *ai) { + int sfd; + int flags; + + if ((sfd = socket(ai->ai_family, ai->ai_socktype, ai->ai_protocol)) == -1) { + perror("socket()"); + return -1; + } + + if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || + fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) { + perror("setting O_NONBLOCK"); + close(sfd); + return -1; + } + return sfd; +} + + +/* + * Sets a socket's send buffer size to the maximum allowed by the system. + */ +static void maximize_sndbuf(const int sfd) { + socklen_t intsize = sizeof(int); + int last_good = 0; + int min, max, avg; + int old_size; + + /* Start with the default size. */ + if (getsockopt(sfd, SOL_SOCKET, SO_SNDBUF, &old_size, &intsize) != 0) { + if (settings.verbose > 0) + perror("getsockopt(SO_SNDBUF)"); + return; + } + + /* Binary-search for the real maximum. */ + min = old_size; + max = MAX_SENDBUF_SIZE; + + while (min <= max) { + avg = ((unsigned int)(min + max)) / 2; + if (setsockopt(sfd, SOL_SOCKET, SO_SNDBUF, (void *)&avg, intsize) == 0) { + last_good = avg; + min = avg + 1; + } else { + max = avg - 1; + } + } + + if (settings.verbose > 1) + fprintf(stderr, "<%d send buffer was %d, now %d\n", sfd, old_size, last_good); +} + +static int server_socket(const int port, const bool is_udp) { + int sfd; + struct linger ling = {0, 0}; + struct addrinfo *ai; + struct addrinfo *next; + struct addrinfo hints; + char port_buf[NI_MAXSERV]; + int error; + int success = 0; + + int flags =1; + + /* + * the memset call clears nonstandard fields in some impementations + * that otherwise mess things up. + */ + memset(&hints, 0, sizeof (hints)); + hints.ai_flags = AI_PASSIVE|AI_ADDRCONFIG; + if (is_udp) + { + hints.ai_protocol = IPPROTO_UDP; + hints.ai_socktype = SOCK_DGRAM; + hints.ai_family = AF_INET; /* This left here because of issues with OSX 10.5 */ + } else { + hints.ai_family = AF_UNSPEC; + hints.ai_protocol = IPPROTO_TCP; + hints.ai_socktype = SOCK_STREAM; + } + + snprintf(port_buf, NI_MAXSERV, "%d", port); + error= getaddrinfo(settings.inter, port_buf, &hints, &ai); + if (error != 0) { + if (error != EAI_SYSTEM) + fprintf(stderr, "getaddrinfo(): %s\n", gai_strerror(error)); + else + perror("getaddrinfo()"); + + return 1; + } + + for (next= ai; next; next= next->ai_next) { + conn *listen_conn_add; + if ((sfd = new_socket(next)) == -1) { + freeaddrinfo(ai); + return 1; + } + + setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags)); + if (is_udp) { + maximize_sndbuf(sfd); + } else { + setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags)); + setsockopt(sfd, SOL_SOCKET, SO_LINGER, (void *)&ling, sizeof(ling)); + setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, (void *)&flags, sizeof(flags)); + } + + if (bind(sfd, next->ai_addr, next->ai_addrlen) == -1) { + if (errno != EADDRINUSE) { + perror("bind()"); + close(sfd); + freeaddrinfo(ai); + return 1; + } + close(sfd); + continue; + } else { + success++; + if (!is_udp && listen(sfd, 1024) == -1) { + perror("listen()"); + close(sfd); + freeaddrinfo(ai); + return 1; + } + } + + if (is_udp) + { + int c; + + for (c = 0; c < settings.num_threads; c++) { + /* this is guaranteed to hit all threads because we round-robin */ + dispatch_conn_new(sfd, conn_read, EV_READ | EV_PERSIST, + UDP_READ_BUFFER_SIZE, 1); + } + } else { + if (!(listen_conn_add = conn_new(sfd, conn_listening, + EV_READ | EV_PERSIST, 1, false, main_base))) { + fprintf(stderr, "failed to create listening connection\n"); + exit(EXIT_FAILURE); + } + + listen_conn_add->next = listen_conn; + listen_conn = listen_conn_add; + } + } + + freeaddrinfo(ai); + + /* Return zero iff we detected no errors in starting up connections */ + return success == 0; +} + +static int new_socket_unix(void) { + int sfd; + int flags; + + if ((sfd = socket(AF_UNIX, SOCK_STREAM, 0)) == -1) { + perror("socket()"); + return -1; + } + + if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || + fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) { + perror("setting O_NONBLOCK"); + close(sfd); + return -1; + } + return sfd; +} + +static int server_socket_unix(const char *path, int access_mask) { + int sfd; + struct linger ling = {0, 0}; + struct sockaddr_un addr; + struct stat tstat; + int flags =1; + int old_umask; + + if (!path) { + return 1; + } + + if ((sfd = new_socket_unix()) == -1) { + return 1; + } + + /* + * Clean up a previous socket file if we left it around + */ + if (lstat(path, &tstat) == 0) { + if (S_ISSOCK(tstat.st_mode)) + unlink(path); + } + + setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, (void *)&flags, sizeof(flags)); + setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, (void *)&flags, sizeof(flags)); + setsockopt(sfd, SOL_SOCKET, SO_LINGER, (void *)&ling, sizeof(ling)); + + /* + * the memset call clears nonstandard fields in some impementations + * that otherwise mess things up. + */ + memset(&addr, 0, sizeof(addr)); + + addr.sun_family = AF_UNIX; + strcpy(addr.sun_path, path); + old_umask=umask( ~(access_mask&0777)); + if (bind(sfd, (struct sockaddr *)&addr, sizeof(addr)) == -1) { + perror("bind()"); + close(sfd); + umask(old_umask); + return 1; + } + umask(old_umask); + if (listen(sfd, 1024) == -1) { + perror("listen()"); + close(sfd); + return 1; + } + if (!(listen_conn = conn_new(sfd, conn_listening, + EV_READ | EV_PERSIST, 1, false, main_base))) { + fprintf(stderr, "failed to create listening connection\n"); + exit(EXIT_FAILURE); + } + + return 0; +} + +/* + * We keep the current time of day in a global variable that's updated by a + * timer event. This saves us a bunch of time() system calls (we really only + * need to get the time once a second, whereas there can be tens of thousands + * of requests a second) and allows us to use server-start-relative timestamps + * rather than absolute UNIX timestamps, a space savings on systems where + * sizeof(time_t) > sizeof(unsigned int). + */ +volatile rel_time_t current_time; +static struct event clockevent; + +/* time-sensitive callers can call it by hand with this, outside the normal ever-1-second timer */ +static void set_current_time(void) { + struct timeval timer; + + gettimeofday(&timer, NULL); + current_time = (rel_time_t) (timer.tv_sec - stats.started); +} + +static void clock_handler(const int fd, const short which, void *arg) { + struct timeval t = {.tv_sec = 1, .tv_usec = 0}; + static bool initialized = false; + + if (initialized) { + /* only delete the event if it's actually there. */ + evtimer_del(&clockevent); + } else { + initialized = true; + } + + evtimer_set(&clockevent, clock_handler, 0); + event_base_set(main_base, &clockevent); + evtimer_add(&clockevent, &t); + + set_current_time(); +} + +static struct event deleteevent; + +static void delete_handler(const int fd, const short which, void *arg) { + struct timeval t = {.tv_sec = 5, .tv_usec = 0}; + static bool initialized = false; + + if (initialized) { + /* some versions of libevent don't like deleting events that don't exist, + so only delete once we know this event has been added. */ + evtimer_del(&deleteevent); + } else { + initialized = true; + } + + evtimer_set(&deleteevent, delete_handler, 0); + event_base_set(main_base, &deleteevent); + evtimer_add(&deleteevent, &t); + run_deferred_deletes(); +} + +/* Call run_deferred_deletes instead of this. */ +void do_run_deferred_deletes(void) +{ + int i, j = 0; + + for (i = 0; i < delcurr; i++) { + item *it = todelete[i]; + if (item_delete_lock_over(it)) { + assert(it->refcount > 0); + it->it_flags &= ~ITEM_DELETED; + do_item_unlink(it); + do_item_remove(it); + } else { + todelete[j++] = it; + } + } + delcurr = j; +} + +static void usage(void) { + printf(PACKAGE " " VERSION "\n"); + printf("-p TCP port number to listen on (default: 11211)\n" + "-U UDP port number to listen on (default: 0, off)\n" + "-s unix socket path to listen on (disables network support)\n" + "-a access mask for unix socket, in octal (default 0700)\n" + "-l interface to listen on, default is INDRR_ANY\n" + "-d run as a daemon\n" + "-r maximize core file limit\n" + "-u assume identity of (only when run as root)\n" + "-m max memory to use for items in megabytes, default is 64 MB\n" + "-M return error on memory exhausted (rather than removing items)\n" + "-c max simultaneous connections, default is 1024\n" + "-k lock down all paged memory. Note that there is a\n" + " limit on how much memory you may lock. Trying to\n" + " allocate more than that would fail, so be sure you\n" + " set the limit correctly for the user you started\n" + " the daemon with (not for -u user;\n" + " under sh this is done with 'ulimit -S -l NUM_KB').\n" + "-v verbose (print errors/warnings while in event loop)\n" + "-vv very verbose (also print client commands/reponses)\n" + "-h print this help and exit\n" + "-i print memcached and libevent license\n" + "-b run a managed instanced (mnemonic: buckets)\n" + "-P save PID in , only used with -d option\n" + "-f chunk size growth factor, default 1.25\n" + "-n minimum space allocated for key+value+flags, default 48\n" + +#if defined(HAVE_GETPAGESIZES) && defined(HAVE_MEMCNTL) + "-L Try to use large memory pages (if available). Increasing\n" + " the memory page size could reduce the number of TLB misses\n" + " and improve the performance. In order to get large pages\n" + " from the OS, memcached will allocate the total item-cache\n" + " in one large chunk.\n" +#endif + ); + +#ifdef USE_THREADS + printf("-t number of threads to use, default 4\n"); +#endif + return; +} + +static void usage_license(void) { + printf(PACKAGE " " VERSION "\n\n"); + printf( + "Copyright (c) 2003, Danga Interactive, Inc. \n" + "All rights reserved.\n" + "\n" + "Redistribution and use in source and binary forms, with or without\n" + "modification, are permitted provided that the following conditions are\n" + "met:\n" + "\n" + " * Redistributions of source code must retain the above copyright\n" + "notice, this list of conditions and the following disclaimer.\n" + "\n" + " * Redistributions in binary form must reproduce the above\n" + "copyright notice, this list of conditions and the following disclaimer\n" + "in the documentation and/or other materials provided with the\n" + "distribution.\n" + "\n" + " * Neither the name of the Danga Interactive nor the names of its\n" + "contributors may be used to endorse or promote products derived from\n" + "this software without specific prior written permission.\n" + "\n" + "THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS\n" + "\"AS IS\" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT\n" + "LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR\n" + "A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT\n" + "OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,\n" + "SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT\n" + "LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,\n" + "DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY\n" + "THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT\n" + "(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE\n" + "OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n" + "\n" + "\n" + "This product includes software developed by Niels Provos.\n" + "\n" + "[ libevent ]\n" + "\n" + "Copyright 2000-2003 Niels Provos \n" + "All rights reserved.\n" + "\n" + "Redistribution and use in source and binary forms, with or without\n" + "modification, are permitted provided that the following conditions\n" + "are met:\n" + "1. Redistributions of source code must retain the above copyright\n" + " notice, this list of conditions and the following disclaimer.\n" + "2. Redistributions in binary form must reproduce the above copyright\n" + " notice, this list of conditions and the following disclaimer in the\n" + " documentation and/or other materials provided with the distribution.\n" + "3. All advertising materials mentioning features or use of this software\n" + " must display the following acknowledgement:\n" + " This product includes software developed by Niels Provos.\n" + "4. The name of the author may not be used to endorse or promote products\n" + " derived from this software without specific prior written permission.\n" + "\n" + "THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR\n" + "IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES\n" + "OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.\n" + "IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,\n" + "INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT\n" + "NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,\n" + "DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY\n" + "THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT\n" + "(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF\n" + "THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.\n" + ); + + return; +} + +static void save_pid(const pid_t pid, const char *pid_file) { + FILE *fp; + if (pid_file == NULL) + return; + + if ((fp = fopen(pid_file, "w")) == NULL) { + fprintf(stderr, "Could not open the pid file %s for writing\n", pid_file); + return; + } + + fprintf(fp,"%ld\n", (long)pid); + if (fclose(fp) == -1) { + fprintf(stderr, "Could not close the pid file %s.\n", pid_file); + return; + } +} + +static void remove_pidfile(const char *pid_file) { + if (pid_file == NULL) + return; + + if (unlink(pid_file) != 0) { + fprintf(stderr, "Could not remove the pid file %s.\n", pid_file); + } + +} + + +static void sig_handler(const int sig) { + printf("SIGINT handled.\n"); + exit(EXIT_SUCCESS); +} + +#if defined(HAVE_GETPAGESIZES) && defined(HAVE_MEMCNTL) +/* + * On systems that supports multiple page sizes we may reduce the + * number of TLB-misses by using the biggest available page size + */ +int enable_large_pages(void) { + int ret = -1; + size_t sizes[32]; + int avail = getpagesizes(sizes, 32); + if (avail != -1) { + size_t max = sizes[0]; + struct memcntl_mha arg = {0}; + int ii; + + for (ii = 1; ii < avail; ++ii) { + if (max < sizes[ii]) { + max = sizes[ii]; + } + } + + arg.mha_flags = 0; + arg.mha_pagesize = max; + arg.mha_cmd = MHA_MAPSIZE_BSSBRK; + + if (memcntl(0, 0, MC_HAT_ADVISE, (caddr_t)&arg, 0, 0) == -1) { + fprintf(stderr, "Failed to set large pages: %s\n", + strerror(errno)); + fprintf(stderr, "Will use default page size\n"); + } else { + ret = 0; + } + } else { + fprintf(stderr, "Failed to get supported pagesizes: %s\n", + strerror(errno)); + fprintf(stderr, "Will use default page size\n"); + } + + return ret; +} +#endif + +int main (int argc, char **argv) { + int c; + int x; + bool lock_memory = false; + bool daemonize = false; + bool preallocate = false; + int maxcore = 0; + char *username = NULL; + char *pid_file = NULL; + struct passwd *pw; + struct sigaction sa; + struct rlimit rlim; + /* listening socket */ + static int *l_socket = NULL; + + /* udp socket */ + static int *u_socket = NULL; + static int u_socket_count = 0; + + /* handle SIGINT */ + signal(SIGINT, sig_handler); + + /* init settings */ + settings_init(); + + /* set stderr non-buffering (for running under, say, daemontools) */ + setbuf(stderr, NULL); + + /* process arguments */ + while ((c = getopt(argc, argv, "a:bp:s:U:m:Mc:khirvdl:u:P:f:s:n:t:D:L")) != -1) { + switch (c) { + case 'a': + /* access for unix domain socket, as octal mask (like chmod)*/ + settings.access= strtol(optarg,NULL,8); + break; + + case 'U': + settings.udpport = atoi(optarg); + break; + case 'b': + settings.managed = true; + break; + case 'p': + settings.port = atoi(optarg); + break; + case 's': + settings.socketpath = optarg; + break; + case 'm': + settings.maxbytes = ((size_t)atoi(optarg)) * 1024 * 1024; + break; + case 'M': + settings.evict_to_free = 0; + break; + case 'c': + settings.maxconns = atoi(optarg); + break; + case 'h': + usage(); + exit(EXIT_SUCCESS); + case 'i': + usage_license(); + exit(EXIT_SUCCESS); + case 'k': + lock_memory = true; + break; + case 'v': + settings.verbose++; + break; + case 'l': + settings.inter= strdup(optarg); + break; + case 'd': + daemonize = true; + break; + case 'r': + maxcore = 1; + break; + case 'u': + username = optarg; + break; + case 'P': + pid_file = optarg; + break; + case 'f': + settings.factor = atof(optarg); + if (settings.factor <= 1.0) { + fprintf(stderr, "Factor must be greater than 1\n"); + return 1; + } + break; + case 'n': + settings.chunk_size = atoi(optarg); + if (settings.chunk_size == 0) { + fprintf(stderr, "Chunk size must be greater than 0\n"); + return 1; + } + break; + case 't': + settings.num_threads = atoi(optarg); + if (settings.num_threads == 0) { + fprintf(stderr, "Number of threads must be greater than 0\n"); + return 1; + } + break; + case 'D': + if (! optarg || ! optarg[0]) { + fprintf(stderr, "No delimiter specified\n"); + return 1; + } + settings.prefix_delimiter = optarg[0]; + settings.detail_enabled = 1; + break; +#if defined(HAVE_GETPAGESIZES) && defined(HAVE_MEMCNTL) + case 'L' : + if (enable_large_pages() == 0) { + preallocate = true; + } + break; +#endif + default: + fprintf(stderr, "Illegal argument \"%c\"\n", c); + return 1; + } + } + + if (maxcore != 0) { + struct rlimit rlim_new; + /* + * First try raising to infinity; if that fails, try bringing + * the soft limit to the hard. + */ + if (getrlimit(RLIMIT_CORE, &rlim) == 0) { + rlim_new.rlim_cur = rlim_new.rlim_max = RLIM_INFINITY; + if (setrlimit(RLIMIT_CORE, &rlim_new)!= 0) { + /* failed. try raising just to the old max */ + rlim_new.rlim_cur = rlim_new.rlim_max = rlim.rlim_max; + (void)setrlimit(RLIMIT_CORE, &rlim_new); + } + } + /* + * getrlimit again to see what we ended up with. Only fail if + * the soft limit ends up 0, because then no core files will be + * created at all. + */ + + if ((getrlimit(RLIMIT_CORE, &rlim) != 0) || rlim.rlim_cur == 0) { + fprintf(stderr, "failed to ensure corefile creation\n"); + exit(EXIT_FAILURE); + } + } + + /* + * If needed, increase rlimits to allow as many connections + * as needed. + */ + + if (getrlimit(RLIMIT_NOFILE, &rlim) != 0) { + fprintf(stderr, "failed to getrlimit number of files\n"); + exit(EXIT_FAILURE); + } else { + int maxfiles = settings.maxconns; + if (rlim.rlim_cur < maxfiles) + rlim.rlim_cur = maxfiles + 3; + if (rlim.rlim_max < rlim.rlim_cur) + rlim.rlim_max = rlim.rlim_cur; + if (setrlimit(RLIMIT_NOFILE, &rlim) != 0) { + fprintf(stderr, "failed to set rlimit for open files. Try running as root or requesting smaller maxconns value.\n"); + exit(EXIT_FAILURE); + } + } + + /* lock paged memory if needed */ + if (lock_memory) { +#ifdef HAVE_MLOCKALL + int res = mlockall(MCL_CURRENT | MCL_FUTURE); + if (res != 0) { + fprintf(stderr, "warning: -k invalid, mlockall() failed: %s\n", + strerror(errno)); + } +#else + fprintf(stderr, "warning: -k invalid, mlockall() not supported on this platform. proceeding without.\n"); +#endif + } + + /* lose root privileges if we have them */ + if (getuid() == 0 || geteuid() == 0) { + if (username == 0 || *username == '\0') { + fprintf(stderr, "can't run as root without the -u switch\n"); + return 1; + } + if ((pw = getpwnam(username)) == 0) { + fprintf(stderr, "can't find the user %s to switch to\n", username); + return 1; + } + if (setgid(pw->pw_gid) < 0 || setuid(pw->pw_uid) < 0) { + fprintf(stderr, "failed to assume identity of user %s\n", username); + return 1; + } + } + + /* daemonize if requested */ + /* if we want to ensure our ability to dump core, don't chdir to / */ + if (daemonize) { + int res; + res = daemon(maxcore, settings.verbose); + if (res == -1) { + fprintf(stderr, "failed to daemon() in order to daemonize\n"); + return 1; + } + } + + + /* initialize main thread libevent instance */ + main_base = event_init(); + + /* initialize other stuff */ + item_init(); + stats_init(); + assoc_init(); + conn_init(); + /* Hacky suffix buffers. */ + suffix_init(); + slabs_init(settings.maxbytes, settings.factor, preallocate); + + /* managed instance? alloc and zero a bucket array */ + if (settings.managed) { + buckets = malloc(sizeof(int) * MAX_BUCKETS); + if (buckets == 0) { + fprintf(stderr, "failed to allocate the bucket array"); + exit(EXIT_FAILURE); + } + memset(buckets, 0, sizeof(int) * MAX_BUCKETS); + } + + /* + * ignore SIGPIPE signals; we can use errno==EPIPE if we + * need that information + */ + sa.sa_handler = SIG_IGN; + sa.sa_flags = 0; + if (sigemptyset(&sa.sa_mask) == -1 || + sigaction(SIGPIPE, &sa, 0) == -1) { + perror("failed to ignore SIGPIPE; sigaction"); + exit(EXIT_FAILURE); + } + /* start up worker threads if MT mode */ + thread_init(settings.num_threads, main_base); + /* save the PID in if we're a daemon, do this after thread_init due to + a file descriptor handling bug somewhere in libevent */ + if (daemonize) + save_pid(getpid(), pid_file); + /* initialise clock event */ + clock_handler(0, 0, 0); + /* initialise deletion array and timer event */ + deltotal = 200; + delcurr = 0; + if ((todelete = malloc(sizeof(item *) * deltotal)) == NULL) { + perror("failed to allocate memory for deletion array"); + exit(EXIT_FAILURE); + } + delete_handler(0, 0, 0); /* sets up the event */ + + /* create unix mode sockets after dropping privileges */ + if (settings.socketpath != NULL) { + if (server_socket_unix(settings.socketpath,settings.access)) { + fprintf(stderr, "failed to listen\n"); + exit(EXIT_FAILURE); + } + } + + /* create the listening socket, bind it, and init */ + if (settings.socketpath == NULL) { + int udp_port; + + if (server_socket(settings.port, 0)) { + fprintf(stderr, "failed to listen\n"); + exit(EXIT_FAILURE); + } + /* + * initialization order: first create the listening sockets + * (may need root on low ports), then drop root if needed, + * then daemonise if needed, then init libevent (in some cases + * descriptors created by libevent wouldn't survive forking). + */ + udp_port = settings.udpport ? settings.udpport : settings.port; + + /* create the UDP listening socket and bind it */ + if (server_socket(udp_port, 1)) { + fprintf(stderr, "failed to listen on UDP port %d\n", settings.udpport); + exit(EXIT_FAILURE); + } + } + + /* enter the event loop */ + event_base_loop(main_base, 0); + /* remove the PID file if we're a daemon */ + if (daemonize) + remove_pidfile(pid_file); + /* Clean up strdup() call for bind() address */ + if (settings.inter) + free(settings.inter); + if (l_socket) + free(l_socket); + if (u_socket) + free(u_socket); + + return 0; +} Index: /tags/1.2.5/ChangeLog =================================================================== --- /tags/1.2.5/ChangeLog (revision 741) +++ /tags/1.2.5/ChangeLog (revision 741) @@ -0,0 +1,517 @@ +2008-03-02 [Version 1.2.5-rc1 released] + + * Add per-item-class tracking of evictions and OOM errors (dormando) + + * Optimize item_alloc() a little (dormando) + + * Give 'SERVER_ERROR out of memory' errors more context (dormando) + + * Enable usage of large memory pages under solaris + (Trond.Norbye@Sun.COM) + + * Enable UDP by default, clean up server socket code + (brian@tangent.org) + + * 'noreply' support (Tomash Brechko) + + * IPv6 support, and IPv6 multi-interface support (brian@tangent.org) + + * Add compiler options for Sun Studio compilers with --enable-threads + (Trond.Norbye@Sun.COM) + + * Add --enable-64bit for mulitarget platforms (Trond.Norbye@Sun.COM) + + * Use gettimeofday(2) instead of time(2). + + * Make -k option work (Tomash Brechko) + + * Fix chunk slab alignment (Trond.Norbye@Sun.COM) + +2007-12-06 [Version 1.2.4 released] + +2007-12-05 + + * Fix compilation on panther (JS and Dormando) + + * More CAS tests (Chris Goffinet) + + * Final fixes for all 1.2.4 features are in, -rc2 sent out. + +2007-11-19 [Version 1.2.4-rc1 released] + +2007-11-19 Dormando + + * Patch series from Tomash Brechko : + Minor fixes and optimisations. + + * Patches from Chris, Dustin, and Dormando to fix CAS. + + * Prepping for 1.2.4 release. + +2007-11-13 Dormando + + * Adjusted patch from js : Compile on OS X Panther + and earlier. + +2007-11-12 Steven Grimm + + * Patch from Tomash Brechko : Always send + "SERVER_ERROR out of memory" when memory exhausted. + +2007-10-15 Paul Lindner + + * Patch from David Bremner that implements + a new option "-a" which takes an octal permission mask + (like chmod) sets the permissions on the unix domain socket + (specified by "-s"). + +2007-10-03 Paul Lindner + * Incorporate "cas" operation developed by Dustin + Sallings This change allows you + to do atomic changes to an existing key. + + * Fix for stats.evictions not incrementing + when exptime == 0 items are kicked off the cache. + from Jean-Francois BUSTARRET . + + * Fix for do_item_cachedump() which was returning + an incorrect timestamp. + + * Switch to unsigned 64-bit increment/decrement counters + from Evan Miller and Dustin Sallings. + + * Add append command support written by Filipe Laborde. + Thread safe version plus prepend command from Maxim Dounin + + + * The memcached-tool script can now display stats. Patch + provided by Dan Christian + + * Fix for Unix Domain sockets on FreeBSD + FreeBSD's sendmsg() requires msg_name in msghdr structure + to be NULL if not used, setting msg_namelen to 0 isn't enough. + Patch from Maxim Dounin + +2007-08-21 Paul Lindner + * Incorporate incrememnt patch from Evan Miller + to define increment overflow + behavior. + +2007-08-07 Leon Brocard + * Bring the memcached.1 manpage up to date + +2007-08-06 Paul Lindner + * Fix crash when using -P and -d flags on x86_64 + with latest libevent release. + +2007-07-08 Steven Grimm + + * Item stats commands weren't thread-safe; wrap them with locks + when compiled in multithreaded mode. + * The "stats items" command now works again; it broke with the + introduction of the powers-of-N chunk size change. + +2007-07-06 [Version 1.2.3 released] + +2007-06-19 Paul Lindner + + * Solaris portability fixes from Trond Norbye + +2007-05-29 Paul Lindner + + * Properly document evictions statistic value + +2007-05-10 Paul Lindner + + * Flesh out tests for unix domain sockets and binary data. + * Update rpm spec file to run tests + +2007-05-07 Paul Lindner + + * Fix compilation bug on freebsd 6.x (and maybe others) + * Update RPM spec file per redhat bugzilla #238994 + * Move unistd.h to memcached.h to get rid of warnings + * Add string.h to thread.c to get correctly prototyped strerror() + +2007-05-04 Paul Lindner + + * Add fedora/redhat style init script and RPM spec file + +2007-05-12 [Version 1.2.2 released] + +2007-04-16 Steven Grimm + + * Command tokenizer performance and cleanliness improvement. + Patch contributed by Paolo Borelli . + +2007-04-16 Paul Lindner + + * Add notes to README about MacOS, libevent and kqueue. + + * Windows Patch integration -- part 1, warnings elimination. + +2007-04-12 Paul Lindner + + * Allow changes to the verbosity level of the server with a new + "verbosity" command and some compiler cleanups. + Patch contributed by Paolo Borelli . + +2007-04-08 Paul Lindner + + * Add cleanup patch from "Tim Yardley" to + clean up source spacing issues, fix -Wall warnings, add some + null checks, adds asserts at the top of each function for any + use of conn *c without checking to see if c is NULL first. + + * Also adjust clean-whitespace.pl to clean *.ac files. Add + script to test-suite to test for tabs. + +2007-04-04 Paul Lindner + + * Add clarification of flush_all in the protocol docs + from Elizabeth Mattijsen + +2007-03-31 Paul Lindner + + * Add patch from Eli Bingham to + re-enable the -n switch to memcached. + +2007-03-20 Paul Lindner + * Add patch to collect eviction statistics from + Jean-Francois BUSTARRET . + + * Updated docs, added new test cases for t/stats.t + +2007-03-18 Paul Lindner + + * Add more test cases using larger buffer sizes up to and greater + than 1MB. + + * Remove unused parameter to item_size_ok() + + * Use a single printf() in usage() + + * Add a failing test for conforming with maximum connections. + +2007-03-17 + * crash fix from Thomas van Gulick in + conn_shrink(), passing &ptr, instead of ptr to realloc(). + +2007-03-05 Paul Lindner + * Fix a number of places where (s)printf calls were using unsigned + or signed formats that did not match their arguments. + + * Add support for stdbool.h and stdint.h to use the bool and + uint8_t types. + + * Major refactoring - move API calls for assoc/items/slabs to + their own individual header files. Add apropriate const and + static declarations as appropriate. + + * Avoid type-punning. Do a more efficient realloc inside the + conn_shrink routine. + + * Fix overflow bug where uninitialized access to slabclass caused + size-0 mallocs during slab preallocation. + + * Use EXIT_SUCCESS/EXIT_FAILURE constants. + + * Convert some sprintf calls to snprintf to protect against + buffer overflows. + + * Explicitly compare against NULL or zero in many places. + +2007-03-05 + * Steven Grimm : Per-object-type stats collection + support. Specify the object type delimiter with the -D command line + option. Turn stats gathering on and off with "stats detail on" and + "stats detail off". Dump the per-object-type details with + "stats detail dump". + +2007-03-01 + * Steven Grimm : Fix an off-by-one error in the + multithreaded version's message passing code. + +2006-12-23 + * fix expirations of items set with absolute expiration times in + the past, before the server's start time. bug was introduced in + 1.2.0 with rel_time_t. Thanks to Adam Dixon + for the bug report and test case! + +2006-11-26 + * Steven Grimm : Performance improvements: + + Dynamic sizing of hashtable to reduce collisions on very large + caches and conserve memory on small caches. + + Only reposition items in the LRU queue once a minute, to reduce + overhead of accessing extremely frequently-used items. + + Stop listening for new connections until an existing one closes + if we run out of available file descriptors. + + Command parser refactoring: Add a single-pass tokenizer to cut + down on string scanning. Split the command processing into + separate functions for easier profiling and better readability. + Pass key lengths along with the keys in all API functions that + need keys, to avoid needing to call strlen() repeatedly. + +2006-11-25 + * Steve Peters : OpenBSD has a malloc.h, + but warns to use stdlib.h instead + +2006-11-22 + * Steven Grimm : Add support for multithreaded + execution. Run configure with "--enable-threads" to enable. See + doc/threads.txt for details. + +2006-11-13 + * Iain Wade : Fix for UDP responses on non-"get" + commands. + +2006-10-15 + * Steven Grimm : Dynamic sizing of hashtable to + reduce collisions on very large caches and conserve memory on + small caches. + +2006-10-13 + * Steven Grimm : New faster hash function. + +2006-09-20 + + * don't listen on UDP by default; more clear message when UDP port in use + +2006-09-09 + * release 1.2.0 (along with 1.1.13, which is the more tested branch) + + nobody has run 1.2.0 in production, to my knowledge. facebook has run + their pre-merge-with-trunk version, but bugs were discovered (and fixed) + after the merge. there might be more. you've been warned. :) + +2006-09-04 + * improved autoconf libevent detection, from the Tor project. + +2006-09-03 + * test suite and lot of expiration, delete, flush_all, etc corner + case bugs fixed (Brad Fitzpatrick) + +2006-09-02 + * Nathan Neulinger : fix breakage in expiration code + causing expiration times to not be processed correctly. + +2006-08-21 + * Nathan Neulinger : fix incompatabilities with + unix domain socket support and the UDP code and clean up stale + sockets + +2006-08-20 + * Nathan Neulinger : unix domain socket support + +2006-05-03 + * Steven Grimm : big bunch of changes: + big CPU reduction work, UDP-based interface, increased memory + efficiency. (intertwined patch, committed all together) + + or see svn commit logs + +2006-04-30 + * River Tarnell: autoconf work for Solaris 10. Brad: + merge and verify it works on Nexenta. + +2006-03-04 + * avva: bucket/generation patch (old, but Brad's just finally + committing it) + +2006-01-01 + * Brad Fitzpatrick : allocate 1 slab per class + on start-up, to avoid confusing users with out-of-memory errors + later. this is 18 MB of allocation on start, unless max memory + allowed with -m is lower, in which case only the smaller slab + classes are allocated. + +2005-08-09 + * Elizabeth Mattijsen : needed a way to flush all + memcached backend servers, but not at exactly the same time (to + reduce load peaks), I've added some simple functionality to the + memcached protocol in the "flush_all" command that allows you to + specify a time at which the flush will actually occur (instead of + always at the moment the "flush_all" command is received). + +2005-05-25 + * patch from Peter van Dijk to make + stderr unbuffered, for running under daemontools + +2005-04-04 + * patch from Don MacAskill 'flush_all' doesn't + seem to work properly. Basically, if you try to add a key which + is present, but expired, the store fails but the old key is no + longer expired. + + * release 1.1.12 + +2005-01-14 + * Date: Thu, 18 Nov 2004 15:25:59 -0600 + From: David Phillips + Here is a patch to configure.ac and Makefile.am to put the man page in + the correct location. Trying to install the man page from a + subdirectory results in the subdirectory being used in the install + path (it tries to install to doc/memcached.1). This is the correct + thing to do: + + - create a Makefile.am in the doc directory that installs the man page + with man_MANS + - modify Makefile.am in the base directory to reference the doc + directory using SUBDIRS + - modify the AC_CONFIG_FILES macro in configure.ac to output the + Makefile in doc + + +2005-01-14 + * pidfile saving support from Lisa Seelye , sent + Jan 13, 2005 + +2005-01-14 + * don't delete libevent events that haven't been added (the deltimer) + patch from Ted Schundler + +2004-12-10 + * document -M and -r in manpage (Doug Porter ) + +2004-07-22 + * fix buffer overflow in items.c with 250 byte keys along with + other info on the same line going into a 256 byte char[]. + thanks to Andrei Nigmatulin + +2004-06-15 + * immediate deletes weren't being unlinked a few seconds, + preventing "add" commands to the same key in that time period. + thanks to Michael Alan Dorman for the + bug report and demo script. + +2004-04-30 + * released 1.1.11 + +2004-04-24 + * Avva: Add a new command line option: -r , to maximize core file + limit. + +2004-03-31 + * Avva: Use getrlimit and setrlimit to set limits for number of + simultaneously open file descriptors. Get the current limits and + try to raise them if they're not enough for the specified (or the + default) setting of max connections. + +2004-02-24 + * Adds a '-M' flag to turn off tossing items from the cache. + (Jason Titus ) + +2004-02-19 (Evan) + * Install manpage on "make install", etc. + +2003-12-30 (Brad) + * remove static build stuff. interferes with PAM setuid stuff + and was only included as a possible fix with the old memory + allocator. really shouldn't make a difference. + * add Jay Bonci's Debian scripts and manpage + * release version 1.1.10 + +2003-12-01 (Avva) + * New command: flush_all, causes all existing items to + be invalidated immediately (without deleting them from + memory, merely causing memcached to no longer return them). +2003-10-23 + * Shift init code around to fix daemon mode on FreeBSD, + * and drop root only after creating the server socket (to + * allow the use of privileged ports) + * version 1.1.10pre + +2003-10-09 + * BSD compile fixes from Ryan T. Dean + * version 1.1.9 + +2003-09-29 + * ignore SIGPIPE at start instead of crashing in rare cases it + comes up. no other code had to be modified, since everything + else is already dead-connection-aware. (avva) + +2003-09-09 (Avva, Lisa Marie Seelye ) + * setuid support + +2003-09-05 (Avva) + * accept all new connections in the same event (so we work with ET epoll) + * mark all items as clsid=0 after slab page reassignment to please future + asserts (on the road to making slab page reassignment work fully) + +2003-08-12 (Brad Fitzpatrick) + * use TCP_CORK on Linux or TCP_PUSH on BSD + * only use TCP_NODELAY when we don't have alternatives + +2003-08-10 + * disable Nagel's Algorithm (TCP_NODELAY) for better performance (avva) + +2003-08-10 + * support multiple levels of verbosity (-vv) + +2003-08-10 (Evan Martin) + * Makefile.am: debug, optimization, and static flags are controlled + by the configure script. + * configure.ac: + - allow specifying libevent directory with --with-libevent=DIR + - check for malloc.h (unavailable on BSDs) + - check for socklen_t (unavailable on OSX) + * assoc.c, items.c, slabs.c: Remove some unused headers. + * memcached.c: allow for nonexistence of malloc.h; #define a POSIX + macro to import mlockall flags. + +2003-07-29 + * version 1.1.7 + * big bug fix: item exptime 0 meant expire immediately, not never + * version 1.1.8 + +2003-07-22 + * make 'delete' take second arg, of time to refuse new add/replace + * set/add/replace/delete can all take abs or delta time (delta can't + be larger than a month) + +2003-07-21 + * added doc/protocol.txt + +2003-07-01 + * report CPU usage in stats + +2003-06-30 + * version 1.1.6 + * fix a number of obscure bugs + * more stats reporting + +2003-06-10 + * removing use of Judy; use a hash. (judy caused memory fragmentation) + * shrink some structures + * security improvements + * version 1.1.0 + +2003-06-18 + * changing maxsize back to an unsigned int + +2003-06-16 + * adding PHP support + * added CONTRIBUTORS file + * version 1.0.4 + +2003-06-15 + * forgot to distribute website/api (still learning auto*) + * version 1.0.3 + +2003-06-15 + * update to version 1.0.2 + * autoconf/automake fixes for older versions + * make stats report version number + * change license from GPL to BSD + +Fri, 13 Jun 2003 10:05:51 -0700 Evan Martin + + * configure.ac, autogen.sh, Makefile.am: Use autotools. + * items.c, memcached.c: #include for time(), + printf time_t as %lu (is this correct?), + minor warnings fixes. + Index: /tags/1.2.5/thread.c =================================================================== --- /tags/1.2.5/thread.c (revision 739) +++ /tags/1.2.5/thread.c (revision 739) @@ -0,0 +1,675 @@ +/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */ +/* + * Thread management for memcached. + * + * $Id$ + */ +#include "memcached.h" +#include +#include +#include +#include + +#ifdef HAVE_MALLOC_H +#include +#endif + +#ifdef HAVE_STRING_H +#include +#endif + +#ifdef USE_THREADS + +#include + +#define ITEMS_PER_ALLOC 64 + +/* An item in the connection queue. */ +typedef struct conn_queue_item CQ_ITEM; +struct conn_queue_item { + int sfd; + int init_state; + int event_flags; + int read_buffer_size; + int is_udp; + CQ_ITEM *next; +}; + +/* A connection queue. */ +typedef struct conn_queue CQ; +struct conn_queue { + CQ_ITEM *head; + CQ_ITEM *tail; + pthread_mutex_t lock; + pthread_cond_t cond; +}; + +/* Lock for connection freelist */ +static pthread_mutex_t conn_lock; + +/* Lock for alternative item suffix freelist */ +static pthread_mutex_t suffix_lock; + +/* Lock for cache operations (item_*, assoc_*) */ +static pthread_mutex_t cache_lock; + +/* Lock for slab allocator operations */ +static pthread_mutex_t slabs_lock; + +/* Lock for global stats */ +static pthread_mutex_t stats_lock; + +/* Free list of CQ_ITEM structs */ +static CQ_ITEM *cqi_freelist; +static pthread_mutex_t cqi_freelist_lock; + +/* + * Each libevent instance has a wakeup pipe, which other threads + * can use to signal that they've put a new connection on its queue. + */ +typedef struct { + pthread_t thread_id; /* unique ID of this thread */ + struct event_base *base; /* libevent handle this thread uses */ + struct event notify_event; /* listen event for notify pipe */ + int notify_receive_fd; /* receiving end of notify pipe */ + int notify_send_fd; /* sending end of notify pipe */ + CQ new_conn_queue; /* queue of new connections to handle */ +} LIBEVENT_THREAD; + +static LIBEVENT_THREAD *threads; + +/* + * Number of threads that have finished setting themselves up. + */ +static int init_count = 0; +static pthread_mutex_t init_lock; +static pthread_cond_t init_cond; + + +static void thread_libevent_process(int fd, short which, void *arg); + +/* + * Initializes a connection queue. + */ +static void cq_init(CQ *cq) { + pthread_mutex_init(&cq->lock, NULL); + pthread_cond_init(&cq->cond, NULL); + cq->head = NULL; + cq->tail = NULL; +} + +/* + * Waits for work on a connection queue. + */ +static CQ_ITEM *cq_pop(CQ *cq) { + CQ_ITEM *item; + + pthread_mutex_lock(&cq->lock); + while (NULL == cq->head) + pthread_cond_wait(&cq->cond, &cq->lock); + item = cq->head; + cq->head = item->next; + if (NULL == cq->head) + cq->tail = NULL; + pthread_mutex_unlock(&cq->lock); + + return item; +} + +/* + * Looks for an item on a connection queue, but doesn't block if there isn't + * one. + * Returns the item, or NULL if no item is available + */ +static CQ_ITEM *cq_peek(CQ *cq) { + CQ_ITEM *item; + + pthread_mutex_lock(&cq->lock); + item = cq->head; + if (NULL != item) { + cq->head = item->next; + if (NULL == cq->head) + cq->tail = NULL; + } + pthread_mutex_unlock(&cq->lock); + + return item; +} + +/* + * Adds an item to a connection queue. + */ +static void cq_push(CQ *cq, CQ_ITEM *item) { + item->next = NULL; + + pthread_mutex_lock(&cq->lock); + if (NULL == cq->tail) + cq->head = item; + else + cq->tail->next = item; + cq->tail = item; + pthread_cond_signal(&cq->cond); + pthread_mutex_unlock(&cq->lock); +} + +/* + * Returns a fresh connection queue item. + */ +static CQ_ITEM *cqi_new() { + CQ_ITEM *item = NULL; + pthread_mutex_lock(&cqi_freelist_lock); + if (cqi_freelist) { + item = cqi_freelist; + cqi_freelist = item->next; + } + pthread_mutex_unlock(&cqi_freelist_lock); + + if (NULL == item) { + int i; + + /* Allocate a bunch of items at once to reduce fragmentation */ + item = malloc(sizeof(CQ_ITEM) * ITEMS_PER_ALLOC); + if (NULL == item) + return NULL; + + /* + * Link together all the new items except the first one + * (which we'll return to the caller) for placement on + * the freelist. + */ + for (i = 2; i < ITEMS_PER_ALLOC; i++) + item[i - 1].next = &item[i]; + + pthread_mutex_lock(&cqi_freelist_lock); + item[ITEMS_PER_ALLOC - 1].next = cqi_freelist; + cqi_freelist = &item[1]; + pthread_mutex_unlock(&cqi_freelist_lock); + } + + return item; +} + + +/* + * Frees a connection queue item (adds it to the freelist.) + */ +static void cqi_free(CQ_ITEM *item) { + pthread_mutex_lock(&cqi_freelist_lock); + item->next = cqi_freelist; + cqi_freelist = item; + pthread_mutex_unlock(&cqi_freelist_lock); +} + + +/* + * Creates a worker thread. + */ +static void create_worker(void *(*func)(void *), void *arg) { + pthread_t thread; + pthread_attr_t attr; + int ret; + + pthread_attr_init(&attr); + + if ((ret = pthread_create(&thread, &attr, func, arg)) != 0) { + fprintf(stderr, "Can't create thread: %s\n", + strerror(ret)); + exit(1); + } +} + + +/* + * Pulls a conn structure from the freelist, if one is available. + */ +conn *mt_conn_from_freelist() { + conn *c; + + pthread_mutex_lock(&conn_lock); + c = do_conn_from_freelist(); + pthread_mutex_unlock(&conn_lock); + + return c; +} + + +/* + * Adds a conn structure to the freelist. + * + * Returns 0 on success, 1 if the structure couldn't be added. + */ +bool mt_conn_add_to_freelist(conn *c) { + bool result; + + pthread_mutex_lock(&conn_lock); + result = do_conn_add_to_freelist(c); + pthread_mutex_unlock(&conn_lock); + + return result; +} + +/* + * Pulls a suffix buffer from the freelist, if one is available. + */ +char *mt_suffix_from_freelist() { + char *s; + + pthread_mutex_lock(&suffix_lock); + s = do_suffix_from_freelist(); + pthread_mutex_unlock(&suffix_lock); + + return s; +} + + +/* + * Adds a suffix buffer to the freelist. + * + * Returns 0 on success, 1 if the buffer couldn't be added. + */ +bool mt_suffix_add_to_freelist(char *s) { + bool result; + + pthread_mutex_lock(&suffix_lock); + result = do_suffix_add_to_freelist(s); + pthread_mutex_unlock(&suffix_lock); + + return result; +} + + +/****************************** LIBEVENT THREADS *****************************/ + +/* + * Set up a thread's information. + */ +static void setup_thread(LIBEVENT_THREAD *me) { + if (! me->base) { + me->base = event_init(); + if (! me->base) { + fprintf(stderr, "Can't allocate event base\n"); + exit(1); + } + } + + /* Listen for notifications from other threads */ + event_set(&me->notify_event, me->notify_receive_fd, + EV_READ | EV_PERSIST, thread_libevent_process, me); + event_base_set(me->base, &me->notify_event); + + if (event_add(&me->notify_event, 0) == -1) { + fprintf(stderr, "Can't monitor libevent notify pipe\n"); + exit(1); + } + + cq_init(&me->new_conn_queue); +} + + +/* + * Worker thread: main event loop + */ +static void *worker_libevent(void *arg) { + LIBEVENT_THREAD *me = arg; + + /* Any per-thread setup can happen here; thread_init() will block until + * all threads have finished initializing. + */ + + pthread_mutex_lock(&init_lock); + init_count++; + pthread_cond_signal(&init_cond); + pthread_mutex_unlock(&init_lock); + + return (void*) event_base_loop(me->base, 0); +} + + +/* + * Processes an incoming "handle a new connection" item. This is called when + * input arrives on the libevent wakeup pipe. + */ +static void thread_libevent_process(int fd, short which, void *arg) { + LIBEVENT_THREAD *me = arg; + CQ_ITEM *item; + char buf[1]; + + if (read(fd, buf, 1) != 1) + if (settings.verbose > 0) + fprintf(stderr, "Can't read from libevent pipe\n"); + + item = cq_peek(&me->new_conn_queue); + + if (NULL != item) { + conn *c = conn_new(item->sfd, item->init_state, item->event_flags, + item->read_buffer_size, item->is_udp, me->base); + if (c == NULL) { + if (item->is_udp) { + fprintf(stderr, "Can't listen for events on UDP socket\n"); + exit(1); + } else { + if (settings.verbose > 0) { + fprintf(stderr, "Can't listen for events on fd %d\n", + item->sfd); + } + close(item->sfd); + } + } + cqi_free(item); + } +} + +/* Which thread we assigned a connection to most recently. */ +static int last_thread = -1; + +/* + * Dispatches a new connection to another thread. This is only ever called + * from the main thread, either during initialization (for UDP) or because + * of an incoming connection. + */ +void dispatch_conn_new(int sfd, int init_state, int event_flags, + int read_buffer_size, int is_udp) { + CQ_ITEM *item = cqi_new(); + int thread = (last_thread + 1) % settings.num_threads; + + last_thread = thread; + + item->sfd = sfd; + item->init_state = init_state; + item->event_flags = event_flags; + item->read_buffer_size = read_buffer_size; + item->is_udp = is_udp; + + cq_push(&threads[thread].new_conn_queue, item); + if (write(threads[thread].notify_send_fd, "", 1) != 1) { + perror("Writing to thread notify pipe"); + } +} + +/* + * Returns true if this is the thread that listens for new TCP connections. + */ +int mt_is_listen_thread() { + return pthread_self() == threads[0].thread_id; +} + +/********************************* ITEM ACCESS *******************************/ + +/* + * Walks through the list of deletes that have been deferred because the items + * were locked down at the tmie. + */ +void mt_run_deferred_deletes() { + pthread_mutex_lock(&cache_lock); + do_run_deferred_deletes(); + pthread_mutex_unlock(&cache_lock); +} + +/* + * Allocates a new item. + */ +item *mt_item_alloc(char *key, size_t nkey, int flags, rel_time_t exptime, int nbytes) { + item *it; + pthread_mutex_lock(&cache_lock); + it = do_item_alloc(key, nkey, flags, exptime, nbytes); + pthread_mutex_unlock(&cache_lock); + return it; +} + +/* + * Returns an item if it hasn't been marked as expired or deleted, + * lazy-expiring as needed. + */ +item *mt_item_get_notedeleted(const char *key, const size_t nkey, bool *delete_locked) { + item *it; + pthread_mutex_lock(&cache_lock); + it = do_item_get_notedeleted(key, nkey, delete_locked); + pthread_mutex_unlock(&cache_lock); + return it; +} + +/* + * Links an item into the LRU and hashtable. + */ +int mt_item_link(item *item) { + int ret; + + pthread_mutex_lock(&cache_lock); + ret = do_item_link(item); + pthread_mutex_unlock(&cache_lock); + return ret; +} + +/* + * Decrements the reference count on an item and adds it to the freelist if + * needed. + */ +void mt_item_remove(item *item) { + pthread_mutex_lock(&cache_lock); + do_item_remove(item); + pthread_mutex_unlock(&cache_lock); +} + +/* + * Replaces one item with another in the hashtable. + */ +int mt_item_replace(item *old, item *new) { + int ret; + + pthread_mutex_lock(&cache_lock); + ret = do_item_replace(old, new); + pthread_mutex_unlock(&cache_lock); + return ret; +} + +/* + * Unlinks an item from the LRU and hashtable. + */ +void mt_item_unlink(item *item) { + pthread_mutex_lock(&cache_lock); + do_item_unlink(item); + pthread_mutex_unlock(&cache_lock); +} + +/* + * Moves an item to the back of the LRU queue. + */ +void mt_item_update(item *item) { + pthread_mutex_lock(&cache_lock); + do_item_update(item); + pthread_mutex_unlock(&cache_lock); +} + +/* + * Adds an item to the deferred-delete list so it can be reaped later. + */ +char *mt_defer_delete(item *item, time_t exptime) { + char *ret; + + pthread_mutex_lock(&cache_lock); + ret = do_defer_delete(item, exptime); + pthread_mutex_unlock(&cache_lock); + return ret; +} + +/* + * Does arithmetic on a numeric item value. + */ +char *mt_add_delta(item *item, int incr, const int64_t delta, char *buf) { + char *ret; + + pthread_mutex_lock(&cache_lock); + ret = do_add_delta(item, incr, delta, buf); + pthread_mutex_unlock(&cache_lock); + return ret; +} + +/* + * Stores an item in the cache (high level, obeys set/add/replace semantics) + */ +int mt_store_item(item *item, int comm) { + int ret; + + pthread_mutex_lock(&cache_lock); + ret = do_store_item(item, comm); + pthread_mutex_unlock(&cache_lock); + return ret; +} + +/* + * Flushes expired items after a flush_all call + */ +void mt_item_flush_expired() { + pthread_mutex_lock(&cache_lock); + do_item_flush_expired(); + pthread_mutex_unlock(&cache_lock); +} + +/* + * Dumps part of the cache + */ +char *mt_item_cachedump(unsigned int slabs_clsid, unsigned int limit, unsigned int *bytes) { + char *ret; + + pthread_mutex_lock(&cache_lock); + ret = do_item_cachedump(slabs_clsid, limit, bytes); + pthread_mutex_unlock(&cache_lock); + return ret; +} + +/* + * Dumps statistics about slab classes + */ +char *mt_item_stats(int *bytes) { + char *ret; + + pthread_mutex_lock(&cache_lock); + ret = do_item_stats(bytes); + pthread_mutex_unlock(&cache_lock); + return ret; +} + +/* + * Dumps a list of objects of each size in 32-byte increments + */ +char *mt_item_stats_sizes(int *bytes) { + char *ret; + + pthread_mutex_lock(&cache_lock); + ret = do_item_stats_sizes(bytes); + pthread_mutex_unlock(&cache_lock); + return ret; +} + +/****************************** HASHTABLE MODULE *****************************/ + +void mt_assoc_move_next_bucket() { + pthread_mutex_lock(&cache_lock); + do_assoc_move_next_bucket(); + pthread_mutex_unlock(&cache_lock); +} + +/******************************* SLAB ALLOCATOR ******************************/ + +void *mt_slabs_alloc(size_t size, unsigned int id) { + void *ret; + + pthread_mutex_lock(&slabs_lock); + ret = do_slabs_alloc(size, id); + pthread_mutex_unlock(&slabs_lock); + return ret; +} + +void mt_slabs_free(void *ptr, size_t size, unsigned int id) { + pthread_mutex_lock(&slabs_lock); + do_slabs_free(ptr, size, id); + pthread_mutex_unlock(&slabs_lock); +} + +char *mt_slabs_stats(int *buflen) { + char *ret; + + pthread_mutex_lock(&slabs_lock); + ret = do_slabs_stats(buflen); + pthread_mutex_unlock(&slabs_lock); + return ret; +} + +#ifdef ALLOW_SLABS_REASSIGN +int mt_slabs_reassign(unsigned char srcid, unsigned char dstid) { + int ret; + + pthread_mutex_lock(&slabs_lock); + ret = do_slabs_reassign(srcid, dstid); + pthread_mutex_unlock(&slabs_lock); + return ret; +} +#endif + +/******************************* GLOBAL STATS ******************************/ + +void mt_stats_lock() { + pthread_mutex_lock(&stats_lock); +} + +void mt_stats_unlock() { + pthread_mutex_unlock(&stats_lock); +} + +/* + * Initializes the thread subsystem, creating various worker threads. + * + * nthreads Number of event handler threads to spawn + * main_base Event base for main thread + */ +void thread_init(int nthreads, struct event_base *main_base) { + int i; + + pthread_mutex_init(&cache_lock, NULL); + pthread_mutex_init(&conn_lock, NULL); + pthread_mutex_init(&slabs_lock, NULL); + pthread_mutex_init(&stats_lock, NULL); + + pthread_mutex_init(&init_lock, NULL); + pthread_cond_init(&init_cond, NULL); + + pthread_mutex_init(&cqi_freelist_lock, NULL); + cqi_freelist = NULL; + + threads = malloc(sizeof(LIBEVENT_THREAD) * nthreads); + if (! threads) { + perror("Can't allocate thread descriptors"); + exit(1); + } + + threads[0].base = main_base; + threads[0].thread_id = pthread_self(); + + for (i = 0; i < nthreads; i++) { + int fds[2]; + if (pipe(fds)) { + perror("Can't create notify pipe"); + exit(1); + } + + threads[i].notify_receive_fd = fds[0]; + threads[i].notify_send_fd = fds[1]; + + setup_thread(&threads[i]); + } + + /* Create threads after we've done all the libevent setup. */ + for (i = 1; i < nthreads; i++) { + create_worker(worker_libevent, &threads[i]); + } + + /* Wait for all the threads to set themselves up before returning. */ + pthread_mutex_lock(&init_lock); + init_count++; /* main thread */ + while (init_count < nthreads) { + pthread_cond_wait(&init_cond, &init_lock); + } + pthread_mutex_unlock(&init_lock); +} + +#endif Index: /tags/1.2.5/assoc.c =================================================================== --- /tags/1.2.5/assoc.c (revision 595) +++ /tags/1.2.5/assoc.c (revision 595) @@ -0,0 +1,616 @@ +/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */ +/* + * Hash table + * + * The hash function used here is by Bob Jenkins, 1996: + * + * "By Bob Jenkins, 1996. bob_jenkins@burtleburtle.net. + * You may use this code any way you wish, private, educational, + * or commercial. It's free." + * + * The rest of the file is licensed under the BSD license. See LICENSE. + * + * $Id$ + */ + +#include "memcached.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/* + * Since the hash function does bit manipulation, it needs to know + * whether it's big or little-endian. ENDIAN_LITTLE and ENDIAN_BIG + * are set in the configure script. + */ +#if ENDIAN_BIG == 1 +# define HASH_LITTLE_ENDIAN 0 +# define HASH_BIG_ENDIAN 1 +#else +# if ENDIAN_LITTLE == 1 +# define HASH_LITTLE_ENDIAN 1 +# define HASH_BIG_ENDIAN 0 +# else +# define HASH_LITTLE_ENDIAN 0 +# define HASH_BIG_ENDIAN 0 +# endif +#endif + +#define rot(x,k) (((x)<<(k)) ^ ((x)>>(32-(k)))) + +/* +------------------------------------------------------------------------------- +mix -- mix 3 32-bit values reversibly. + +This is reversible, so any information in (a,b,c) before mix() is +still in (a,b,c) after mix(). + +If four pairs of (a,b,c) inputs are run through mix(), or through +mix() in reverse, there are at least 32 bits of the output that +are sometimes the same for one pair and different for another pair. +This was tested for: +* pairs that differed by one bit, by two bits, in any combination + of top bits of (a,b,c), or in any combination of bottom bits of + (a,b,c). +* "differ" is defined as +, -, ^, or ~^. For + and -, I transformed + the output delta to a Gray code (a^(a>>1)) so a string of 1's (as + is commonly produced by subtraction) look like a single 1-bit + difference. +* the base values were pseudorandom, all zero but one bit set, or + all zero plus a counter that starts at zero. + +Some k values for my "a-=c; a^=rot(c,k); c+=b;" arrangement that +satisfy this are + 4 6 8 16 19 4 + 9 15 3 18 27 15 + 14 9 3 7 17 3 +Well, "9 15 3 18 27 15" didn't quite get 32 bits diffing +for "differ" defined as + with a one-bit base and a two-bit delta. I +used http://burtleburtle.net/bob/hash/avalanche.html to choose +the operations, constants, and arrangements of the variables. + +This does not achieve avalanche. There are input bits of (a,b,c) +that fail to affect some output bits of (a,b,c), especially of a. The +most thoroughly mixed value is c, but it doesn't really even achieve +avalanche in c. + +This allows some parallelism. Read-after-writes are good at doubling +the number of bits affected, so the goal of mixing pulls in the opposite +direction as the goal of parallelism. I did what I could. Rotates +seem to cost as much as shifts on every machine I could lay my hands +on, and rotates are much kinder to the top and bottom bits, so I used +rotates. +------------------------------------------------------------------------------- +*/ +#define mix(a,b,c) \ +{ \ + a -= c; a ^= rot(c, 4); c += b; \ + b -= a; b ^= rot(a, 6); a += c; \ + c -= b; c ^= rot(b, 8); b += a; \ + a -= c; a ^= rot(c,16); c += b; \ + b -= a; b ^= rot(a,19); a += c; \ + c -= b; c ^= rot(b, 4); b += a; \ +} + +/* +------------------------------------------------------------------------------- +final -- final mixing of 3 32-bit values (a,b,c) into c + +Pairs of (a,b,c) values differing in only a few bits will usually +produce values of c that look totally different. This was tested for +* pairs that differed by one bit, by two bits, in any combination + of top bits of (a,b,c), or in any combination of bottom bits of + (a,b,c). +* "differ" is defined as +, -, ^, or ~^. For + and -, I transformed + the output delta to a Gray code (a^(a>>1)) so a string of 1's (as + is commonly produced by subtraction) look like a single 1-bit + difference. +* the base values were pseudorandom, all zero but one bit set, or + all zero plus a counter that starts at zero. + +These constants passed: + 14 11 25 16 4 14 24 + 12 14 25 16 4 14 24 +and these came close: + 4 8 15 26 3 22 24 + 10 8 15 26 3 22 24 + 11 8 15 26 3 22 24 +------------------------------------------------------------------------------- +*/ +#define final(a,b,c) \ +{ \ + c ^= b; c -= rot(b,14); \ + a ^= c; a -= rot(c,11); \ + b ^= a; b -= rot(a,25); \ + c ^= b; c -= rot(b,16); \ + a ^= c; a -= rot(c,4); \ + b ^= a; b -= rot(a,14); \ + c ^= b; c -= rot(b,24); \ +} + +#if HASH_LITTLE_ENDIAN == 1 +uint32_t hash( + const void *key, /* the key to hash */ + size_t length, /* length of the key */ + const uint32_t initval) /* initval */ +{ + uint32_t a,b,c; /* internal state */ + union { const void *ptr; size_t i; } u; /* needed for Mac Powerbook G4 */ + + /* Set up the internal state */ + a = b = c = 0xdeadbeef + ((uint32_t)length) + initval; + + u.ptr = key; + if (HASH_LITTLE_ENDIAN && ((u.i & 0x3) == 0)) { + const uint32_t *k = key; /* read 32-bit chunks */ +#ifdef VALGRIND + const uint8_t *k8; +#endif /* ifdef VALGRIND */ + + /*------ all but last block: aligned reads and affect 32 bits of (a,b,c) */ + while (length > 12) + { + a += k[0]; + b += k[1]; + c += k[2]; + mix(a,b,c); + length -= 12; + k += 3; + } + + /*----------------------------- handle the last (probably partial) block */ + /* + * "k[2]&0xffffff" actually reads beyond the end of the string, but + * then masks off the part it's not allowed to read. Because the + * string is aligned, the masked-off tail is in the same word as the + * rest of the string. Every machine with memory protection I've seen + * does it on word boundaries, so is OK with this. But VALGRIND will + * still catch it and complain. The masking trick does make the hash + * noticably faster for short strings (like English words). + */ +#ifndef VALGRIND + + switch(length) + { + case 12: c+=k[2]; b+=k[1]; a+=k[0]; break; + case 11: c+=k[2]&0xffffff; b+=k[1]; a+=k[0]; break; + case 10: c+=k[2]&0xffff; b+=k[1]; a+=k[0]; break; + case 9 : c+=k[2]&0xff; b+=k[1]; a+=k[0]; break; + case 8 : b+=k[1]; a+=k[0]; break; + case 7 : b+=k[1]&0xffffff; a+=k[0]; break; + case 6 : b+=k[1]&0xffff; a+=k[0]; break; + case 5 : b+=k[1]&0xff; a+=k[0]; break; + case 4 : a+=k[0]; break; + case 3 : a+=k[0]&0xffffff; break; + case 2 : a+=k[0]&0xffff; break; + case 1 : a+=k[0]&0xff; break; + case 0 : return c; /* zero length strings require no mixing */ + } + +#else /* make valgrind happy */ + + k8 = (const uint8_t *)k; + switch(length) + { + case 12: c+=k[2]; b+=k[1]; a+=k[0]; break; + case 11: c+=((uint32_t)k8[10])<<16; /* fall through */ + case 10: c+=((uint32_t)k8[9])<<8; /* fall through */ + case 9 : c+=k8[8]; /* fall through */ + case 8 : b+=k[1]; a+=k[0]; break; + case 7 : b+=((uint32_t)k8[6])<<16; /* fall through */ + case 6 : b+=((uint32_t)k8[5])<<8; /* fall through */ + case 5 : b+=k8[4]; /* fall through */ + case 4 : a+=k[0]; break; + case 3 : a+=((uint32_t)k8[2])<<16; /* fall through */ + case 2 : a+=((uint32_t)k8[1])<<8; /* fall through */ + case 1 : a+=k8[0]; break; + case 0 : return c; /* zero length strings require no mixing */ + } + +#endif /* !valgrind */ + + } else if (HASH_LITTLE_ENDIAN && ((u.i & 0x1) == 0)) { + const uint16_t *k = key; /* read 16-bit chunks */ + const uint8_t *k8; + + /*--------------- all but last block: aligned reads and different mixing */ + while (length > 12) + { + a += k[0] + (((uint32_t)k[1])<<16); + b += k[2] + (((uint32_t)k[3])<<16); + c += k[4] + (((uint32_t)k[5])<<16); + mix(a,b,c); + length -= 12; + k += 6; + } + + /*----------------------------- handle the last (probably partial) block */ + k8 = (const uint8_t *)k; + switch(length) + { + case 12: c+=k[4]+(((uint32_t)k[5])<<16); + b+=k[2]+(((uint32_t)k[3])<<16); + a+=k[0]+(((uint32_t)k[1])<<16); + break; + case 11: c+=((uint32_t)k8[10])<<16; /* @fallthrough */ + case 10: c+=k[4]; /* @fallthrough@ */ + b+=k[2]+(((uint32_t)k[3])<<16); + a+=k[0]+(((uint32_t)k[1])<<16); + break; + case 9 : c+=k8[8]; /* @fallthrough */ + case 8 : b+=k[2]+(((uint32_t)k[3])<<16); + a+=k[0]+(((uint32_t)k[1])<<16); + break; + case 7 : b+=((uint32_t)k8[6])<<16; /* @fallthrough */ + case 6 : b+=k[2]; + a+=k[0]+(((uint32_t)k[1])<<16); + break; + case 5 : b+=k8[4]; /* @fallthrough */ + case 4 : a+=k[0]+(((uint32_t)k[1])<<16); + break; + case 3 : a+=((uint32_t)k8[2])<<16; /* @fallthrough */ + case 2 : a+=k[0]; + break; + case 1 : a+=k8[0]; + break; + case 0 : return c; /* zero length strings require no mixing */ + } + + } else { /* need to read the key one byte at a time */ + const uint8_t *k = key; + + /*--------------- all but the last block: affect some 32 bits of (a,b,c) */ + while (length > 12) + { + a += k[0]; + a += ((uint32_t)k[1])<<8; + a += ((uint32_t)k[2])<<16; + a += ((uint32_t)k[3])<<24; + b += k[4]; + b += ((uint32_t)k[5])<<8; + b += ((uint32_t)k[6])<<16; + b += ((uint32_t)k[7])<<24; + c += k[8]; + c += ((uint32_t)k[9])<<8; + c += ((uint32_t)k[10])<<16; + c += ((uint32_t)k[11])<<24; + mix(a,b,c); + length -= 12; + k += 12; + } + + /*-------------------------------- last block: affect all 32 bits of (c) */ + switch(length) /* all the case statements fall through */ + { + case 12: c+=((uint32_t)k[11])<<24; + case 11: c+=((uint32_t)k[10])<<16; + case 10: c+=((uint32_t)k[9])<<8; + case 9 : c+=k[8]; + case 8 : b+=((uint32_t)k[7])<<24; + case 7 : b+=((uint32_t)k[6])<<16; + case 6 : b+=((uint32_t)k[5])<<8; + case 5 : b+=k[4]; + case 4 : a+=((uint32_t)k[3])<<24; + case 3 : a+=((uint32_t)k[2])<<16; + case 2 : a+=((uint32_t)k[1])<<8; + case 1 : a+=k[0]; + break; + case 0 : return c; /* zero length strings require no mixing */ + } + } + + final(a,b,c); + return c; /* zero length strings require no mixing */ +} + +#elif HASH_BIG_ENDIAN == 1 +/* + * hashbig(): + * This is the same as hashword() on big-endian machines. It is different + * from hashlittle() on all machines. hashbig() takes advantage of + * big-endian byte ordering. + */ +uint32_t hash( const void *key, size_t length, const uint32_t initval) +{ + uint32_t a,b,c; + union { const void *ptr; size_t i; } u; /* to cast key to (size_t) happily */ + + /* Set up the internal state */ + a = b = c = 0xdeadbeef + ((uint32_t)length) + initval; + + u.ptr = key; + if (HASH_BIG_ENDIAN && ((u.i & 0x3) == 0)) { + const uint32_t *k = key; /* read 32-bit chunks */ +#ifdef VALGRIND + const uint8_t *k8; +#endif /* ifdef VALGRIND */ + + /*------ all but last block: aligned reads and affect 32 bits of (a,b,c) */ + while (length > 12) + { + a += k[0]; + b += k[1]; + c += k[2]; + mix(a,b,c); + length -= 12; + k += 3; + } + + /*----------------------------- handle the last (probably partial) block */ + /* + * "k[2]<<8" actually reads beyond the end of the string, but + * then shifts out the part it's not allowed to read. Because the + * string is aligned, the illegal read is in the same word as the + * rest of the string. Every machine with memory protection I've seen + * does it on word boundaries, so is OK with this. But VALGRIND will + * still catch it and complain. The masking trick does make the hash + * noticably faster for short strings (like English words). + */ +#ifndef VALGRIND + + switch(length) + { + case 12: c+=k[2]; b+=k[1]; a+=k[0]; break; + case 11: c+=k[2]&0xffffff00; b+=k[1]; a+=k[0]; break; + case 10: c+=k[2]&0xffff0000; b+=k[1]; a+=k[0]; break; + case 9 : c+=k[2]&0xff000000; b+=k[1]; a+=k[0]; break; + case 8 : b+=k[1]; a+=k[0]; break; + case 7 : b+=k[1]&0xffffff00; a+=k[0]; break; + case 6 : b+=k[1]&0xffff0000; a+=k[0]; break; + case 5 : b+=k[1]&0xff000000; a+=k[0]; break; + case 4 : a+=k[0]; break; + case 3 : a+=k[0]&0xffffff00; break; + case 2 : a+=k[0]&0xffff0000; break; + case 1 : a+=k[0]&0xff000000; break; + case 0 : return c; /* zero length strings require no mixing */ + } + +#else /* make valgrind happy */ + + k8 = (const uint8_t *)k; + switch(length) /* all the case statements fall through */ + { + case 12: c+=k[2]; b+=k[1]; a+=k[0]; break; + case 11: c+=((uint32_t)k8[10])<<8; /* fall through */ + case 10: c+=((uint32_t)k8[9])<<16; /* fall through */ + case 9 : c+=((uint32_t)k8[8])<<24; /* fall through */ + case 8 : b+=k[1]; a+=k[0]; break; + case 7 : b+=((uint32_t)k8[6])<<8; /* fall through */ + case 6 : b+=((uint32_t)k8[5])<<16; /* fall through */ + case 5 : b+=((uint32_t)k8[4])<<24; /* fall through */ + case 4 : a+=k[0]; break; + case 3 : a+=((uint32_t)k8[2])<<8; /* fall through */ + case 2 : a+=((uint32_t)k8[1])<<16; /* fall through */ + case 1 : a+=((uint32_t)k8[0])<<24; break; + case 0 : return c; + } + +#endif /* !VALGRIND */ + + } else { /* need to read the key one byte at a time */ + const uint8_t *k = key; + + /*--------------- all but the last block: affect some 32 bits of (a,b,c) */ + while (length > 12) + { + a += ((uint32_t)k[0])<<24; + a += ((uint32_t)k[1])<<16; + a += ((uint32_t)k[2])<<8; + a += ((uint32_t)k[3]); + b += ((uint32_t)k[4])<<24; + b += ((uint32_t)k[5])<<16; + b += ((uint32_t)k[6])<<8; + b += ((uint32_t)k[7]); + c += ((uint32_t)k[8])<<24; + c += ((uint32_t)k[9])<<16; + c += ((uint32_t)k[10])<<8; + c += ((uint32_t)k[11]); + mix(a,b,c); + length -= 12; + k += 12; + } + + /*-------------------------------- last block: affect all 32 bits of (c) */ + switch(length) /* all the case statements fall through */ + { + case 12: c+=k[11]; + case 11: c+=((uint32_t)k[10])<<8; + case 10: c+=((uint32_t)k[9])<<16; + case 9 : c+=((uint32_t)k[8])<<24; + case 8 : b+=k[7]; + case 7 : b+=((uint32_t)k[6])<<8; + case 6 : b+=((uint32_t)k[5])<<16; + case 5 : b+=((uint32_t)k[4])<<24; + case 4 : a+=k[3]; + case 3 : a+=((uint32_t)k[2])<<8; + case 2 : a+=((uint32_t)k[1])<<16; + case 1 : a+=((uint32_t)k[0])<<24; + break; + case 0 : return c; + } + } + + final(a,b,c); + return c; +} +#else /* HASH_XXX_ENDIAN == 1 */ +#error Must define HASH_BIG_ENDIAN or HASH_LITTLE_ENDIAN +#endif /* HASH_XXX_ENDIAN == 1 */ + +typedef unsigned long int ub4; /* unsigned 4-byte quantities */ +typedef unsigned char ub1; /* unsigned 1-byte quantities */ + +/* how many powers of 2's worth of buckets we use */ +static unsigned int hashpower = 16; + +#define hashsize(n) ((ub4)1<<(n)) +#define hashmask(n) (hashsize(n)-1) + +/* Main hash table. This is where we look except during expansion. */ +static item** primary_hashtable = 0; + +/* + * Previous hash table. During expansion, we look here for keys that haven't + * been moved over to the primary yet. + */ +static item** old_hashtable = 0; + +/* Number of items in the hash table. */ +static unsigned int hash_items = 0; + +/* Flag: Are we in the middle of expanding now? */ +static bool expanding = false; + +/* + * During expansion we migrate values with bucket granularity; this is how + * far we've gotten so far. Ranges from 0 .. hashsize(hashpower - 1) - 1. + */ +static unsigned int expand_bucket = 0; + +void assoc_init(void) { + unsigned int hash_size = hashsize(hashpower) * sizeof(void*); + primary_hashtable = malloc(hash_size); + if (! primary_hashtable) { + fprintf(stderr, "Failed to init hashtable.\n"); + exit(EXIT_FAILURE); + } + memset(primary_hashtable, 0, hash_size); +} + +item *assoc_find(const char *key, const size_t nkey) { + uint32_t hv = hash(key, nkey, 0); + item *it; + unsigned int oldbucket; + + if (expanding && + (oldbucket = (hv & hashmask(hashpower - 1))) >= expand_bucket) + { + it = old_hashtable[oldbucket]; + } else { + it = primary_hashtable[hv & hashmask(hashpower)]; + } + + while (it) { + if ((nkey == it->nkey) && + (memcmp(key, ITEM_key(it), nkey) == 0)) { + return it; + } + it = it->h_next; + } + return 0; +} + +/* returns the address of the item pointer before the key. if *item == 0, + the item wasn't found */ + +static item** _hashitem_before (const char *key, const size_t nkey) { + uint32_t hv = hash(key, nkey, 0); + item **pos; + unsigned int oldbucket; + + if (expanding && + (oldbucket = (hv & hashmask(hashpower - 1))) >= expand_bucket) + { + pos = &old_hashtable[oldbucket]; + } else { + pos = &primary_hashtable[hv & hashmask(hashpower)]; + } + + while (*pos && ((nkey != (*pos)->nkey) || memcmp(key, ITEM_key(*pos), nkey))) { + pos = &(*pos)->h_next; + } + return pos; +} + +/* grows the hashtable to the next power of 2. */ +static void assoc_expand(void) { + old_hashtable = primary_hashtable; + + primary_hashtable = calloc(hashsize(hashpower + 1), sizeof(void *)); + if (primary_hashtable) { + if (settings.verbose > 1) + fprintf(stderr, "Hash table expansion starting\n"); + hashpower++; + expanding = true; + expand_bucket = 0; + do_assoc_move_next_bucket(); + } else { + primary_hashtable = old_hashtable; + /* Bad news, but we can keep running. */ + } +} + +/* migrates the next bucket to the primary hashtable if we're expanding. */ +void do_assoc_move_next_bucket(void) { + item *it, *next; + int bucket; + + if (expanding) { + for (it = old_hashtable[expand_bucket]; NULL != it; it = next) { + next = it->h_next; + + bucket = hash(ITEM_key(it), it->nkey, 0) & hashmask(hashpower); + it->h_next = primary_hashtable[bucket]; + primary_hashtable[bucket] = it; + } + + old_hashtable[expand_bucket] = NULL; + + expand_bucket++; + if (expand_bucket == hashsize(hashpower - 1)) { + expanding = false; + free(old_hashtable); + if (settings.verbose > 1) + fprintf(stderr, "Hash table expansion done\n"); + } + } +} + +/* Note: this isn't an assoc_update. The key must not already exist to call this */ +int assoc_insert(item *it) { + uint32_t hv; + unsigned int oldbucket; + + assert(assoc_find(ITEM_key(it), it->nkey) == 0); /* shouldn't have duplicately named things defined */ + + hv = hash(ITEM_key(it), it->nkey, 0); + if (expanding && + (oldbucket = (hv & hashmask(hashpower - 1))) >= expand_bucket) + { + it->h_next = old_hashtable[oldbucket]; + old_hashtable[oldbucket] = it; + } else { + it->h_next = primary_hashtable[hv & hashmask(hashpower)]; + primary_hashtable[hv & hashmask(hashpower)] = it; + } + + hash_items++; + if (! expanding && hash_items > (hashsize(hashpower) * 3) / 2) { + assoc_expand(); + } + + return 1; +} + +void assoc_delete(const char *key, const size_t nkey) { + item **before = _hashitem_before(key, nkey); + + if (*before) { + item *nxt = (*before)->h_next; + (*before)->h_next = 0; /* probably pointless, but whatever. */ + *before = nxt; + hash_items--; + return; + } + /* Note: we never actually get here. the callers don't delete things + they can't find. */ + assert(*before != 0); +} Index: /tags/1.2.5/slabs.h =================================================================== --- /tags/1.2.5/slabs.h (revision 739) +++ /tags/1.2.5/slabs.h (revision 739) @@ -0,0 +1,33 @@ +/* slabs memory allocation */ + +/** Init the subsystem. 1st argument is the limit on no. of bytes to allocate, + 0 if no limit. 2nd argument is the growth factor; each slab will use a chunk + size equal to the previous slab's chunk size times this factor. + 3rd argument specifies if the slab allocator should allocate all memory + up front (if true), or allocate memory in chunks as it is needed (if false) +*/ +void slabs_init(const size_t limit, const double factor, const bool prealloc); + + +/** + * Given object size, return id to use when allocating/freeing memory for object + * 0 means error: can't store such a large object + */ + +unsigned int slabs_clsid(const size_t size); + +/** Allocate object of given length. 0 on error */ /*@null@*/ +void *do_slabs_alloc(const size_t size, unsigned int id); + +/** Free previously allocated object */ +void do_slabs_free(void *ptr, size_t size, unsigned int id); + +/** Fill buffer with stats */ /*@null@*/ +char* do_slabs_stats(int *buflen); + +/* Request some slab be moved between classes + 1 = success + 0 = fail + -1 = tried. busy. send again shortly. */ +int do_slabs_reassign(unsigned char srcid, unsigned char dstid); + Index: /tags/1.2.5/memcached.h =================================================================== --- /tags/1.2.5/memcached.h (revision 739) +++ /tags/1.2.5/memcached.h (revision 739) @@ -0,0 +1,374 @@ +/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */ +/* $Id$ */ + +#ifdef HAVE_CONFIG_H +#include "config.h" +#endif + +#include +#include +#include +#include +#include +#include + +#define DATA_BUFFER_SIZE 2048 +#define UDP_READ_BUFFER_SIZE 65536 +#define UDP_MAX_PAYLOAD_SIZE 1400 +#define UDP_HEADER_SIZE 8 +#define MAX_SENDBUF_SIZE (256 * 1024 * 1024) +/* I'm told the max legnth of a 64-bit num converted to string is 20 bytes. + * Plus a few for spaces, \r\n, \0 */ +#define SUFFIX_SIZE 24 + +/** Initial size of list of items being returned by "get". */ +#define ITEM_LIST_INITIAL 200 + +/** Initial size of list of CAS suffixes appended to "gets" lines. */ +#define SUFFIX_LIST_INITIAL 20 + +/** Initial size of the sendmsg() scatter/gather array. */ +#define IOV_LIST_INITIAL 400 + +/** Initial number of sendmsg() argument structures to allocate. */ +#define MSG_LIST_INITIAL 10 + +/** High water marks for buffer shrinking */ +#define READ_BUFFER_HIGHWAT 8192 +#define ITEM_LIST_HIGHWAT 400 +#define IOV_LIST_HIGHWAT 600 +#define MSG_LIST_HIGHWAT 100 + +/* Get a consistent bool type */ +#if HAVE_STDBOOL_H +# include +#else + typedef enum {false = 0, true = 1} bool; +#endif + +#if HAVE_STDINT_H +# include +#else + typedef unsigned char uint8_t; +#endif + +/* unistd.h is here */ +#if HAVE_UNISTD_H +# include +#endif + +/** Time relative to server start. Smaller than time_t on 64-bit systems. */ +typedef unsigned int rel_time_t; + +struct stats { + unsigned int curr_items; + unsigned int total_items; + uint64_t curr_bytes; + unsigned int curr_conns; + unsigned int total_conns; + unsigned int conn_structs; + uint64_t get_cmds; + uint64_t set_cmds; + uint64_t get_hits; + uint64_t get_misses; + uint64_t evictions; + time_t started; /* when the process was started */ + uint64_t bytes_read; + uint64_t bytes_written; +}; + +#define MAX_VERBOSITY_LEVEL 2 + +struct settings { + size_t maxbytes; + int maxconns; + int port; + int udpport; + char *inter; + int verbose; + rel_time_t oldest_live; /* ignore existing items older than this */ + bool managed; /* if 1, a tracker manages virtual buckets */ + int evict_to_free; + char *socketpath; /* path to unix socket if using local socket */ + int access; /* access mask (a la chmod) for unix domain socket */ + double factor; /* chunk size growth factor */ + int chunk_size; + int num_threads; /* number of libevent threads to run */ + char prefix_delimiter; /* character that marks a key prefix (for stats) */ + int detail_enabled; /* nonzero if we're collecting detailed stats */ +}; + +extern struct stats stats; +extern struct settings settings; + +#define ITEM_LINKED 1 +#define ITEM_DELETED 2 + +/* temp */ +#define ITEM_SLABBED 4 + +typedef struct _stritem { + struct _stritem *next; + struct _stritem *prev; + struct _stritem *h_next; /* hash chain next */ + rel_time_t time; /* least recent access */ + rel_time_t exptime; /* expire time */ + int nbytes; /* size of data */ + unsigned short refcount; + uint8_t nsuffix; /* length of flags-and-length string */ + uint8_t it_flags; /* ITEM_* above */ + uint8_t slabs_clsid;/* which slab class we're in */ + uint8_t nkey; /* key length, w/terminating null and padding */ + uint64_t cas_id; /* the CAS identifier */ + void * end[]; + /* then null-terminated key */ + /* then " flags length\r\n" (no terminating null) */ + /* then data with terminating \r\n (no terminating null; it's binary!) */ +} item; + +#define ITEM_key(item) ((char*)&((item)->end[0])) + +/* warning: don't use these macros with a function, as it evals its arg twice */ +#define ITEM_suffix(item) ((char*) &((item)->end[0]) + (item)->nkey + 1) +#define ITEM_data(item) ((char*) &((item)->end[0]) + (item)->nkey + 1 + (item)->nsuffix) +#define ITEM_ntotal(item) (sizeof(struct _stritem) + (item)->nkey + 1 + (item)->nsuffix + (item)->nbytes) + +enum conn_states { + conn_listening, /** the socket which listens for connections */ + conn_read, /** reading in a command line */ + conn_write, /** writing out a simple response */ + conn_nread, /** reading in a fixed number of bytes */ + conn_swallow, /** swallowing unnecessary bytes w/o storing */ + conn_closing, /** closing this connection */ + conn_mwrite, /** writing out many items sequentially */ +}; + +#define NREAD_ADD 1 +#define NREAD_SET 2 +#define NREAD_REPLACE 3 +#define NREAD_APPEND 4 +#define NREAD_PREPEND 5 +#define NREAD_CAS 6 + +typedef struct conn conn; +struct conn { + int sfd; + int state; + struct event event; + short ev_flags; + short which; /** which events were just triggered */ + + char *rbuf; /** buffer to read commands into */ + char *rcurr; /** but if we parsed some already, this is where we stopped */ + int rsize; /** total allocated size of rbuf */ + int rbytes; /** how much data, starting from rcur, do we have unparsed */ + + char *wbuf; + char *wcurr; + int wsize; + int wbytes; + int write_and_go; /** which state to go into after finishing current write */ + void *write_and_free; /** free this memory after finishing writing */ + + char *ritem; /** when we read in an item's value, it goes here */ + int rlbytes; + + /* data for the nread state */ + + /** + * item is used to hold an item structure created after reading the command + * line of set/add/replace commands, but before we finished reading the actual + * data. The data is read into ITEM_data(item) to avoid extra copying. + */ + + void *item; /* for commands set/add/replace */ + int item_comm; /* which one is it: set/add/replace */ + + /* data for the swallow state */ + int sbytes; /* how many bytes to swallow */ + + /* data for the mwrite state */ + struct iovec *iov; + int iovsize; /* number of elements allocated in iov[] */ + int iovused; /* number of elements used in iov[] */ + + struct msghdr *msglist; + int msgsize; /* number of elements allocated in msglist[] */ + int msgused; /* number of elements used in msglist[] */ + int msgcurr; /* element in msglist[] being transmitted now */ + int msgbytes; /* number of bytes in current msg */ + + item **ilist; /* list of items to write out */ + int isize; + item **icurr; + int ileft; + + char **suffixlist; + int suffixsize; + char **suffixcurr; + int suffixleft; + + /* data for UDP clients */ + bool udp; /* is this is a UDP "connection" */ + int request_id; /* Incoming UDP request ID, if this is a UDP "connection" */ + struct sockaddr request_addr; /* Who sent the most recent request */ + socklen_t request_addr_size; + unsigned char *hdrbuf; /* udp packet headers */ + int hdrsize; /* number of headers' worth of space is allocated */ + + int binary; /* are we in binary mode */ + int bucket; /* bucket number for the next command, if running as + a managed instance. -1 (_not_ 0) means invalid. */ + int gen; /* generation requested for the bucket */ + bool noreply; /* True if the reply should not be sent. */ + conn *next; /* Used for generating a list of conn structures */ +}; + +/* number of virtual buckets for a managed instance */ +#define MAX_BUCKETS 32768 + +/* current time of day (updated periodically) */ +extern volatile rel_time_t current_time; + +/* + * Functions + */ + +conn *do_conn_from_freelist(); +bool do_conn_add_to_freelist(conn *c); +char *do_suffix_from_freelist(); +bool do_suffix_add_to_freelist(char *s); +char *do_defer_delete(item *item, time_t exptime); +void do_run_deferred_deletes(void); +char *do_add_delta(item *item, const bool incr, const int64_t delta, char *buf); +int do_store_item(item *item, int comm); +conn *conn_new(const int sfd, const int init_state, const int event_flags, const int read_buffer_size, const bool is_udp, struct event_base *base); + + +#include "stats.h" +#include "slabs.h" +#include "assoc.h" +#include "items.h" + + +/* + * In multithreaded mode, we wrap certain functions with lock management and + * replace the logic of some other functions. All wrapped functions have + * "mt_" and "do_" variants. In multithreaded mode, the plain version of a + * function is #define-d to the "mt_" variant, which often just grabs a + * lock and calls the "do_" function. In singlethreaded mode, the "do_" + * function is called directly. + * + * Functions such as the libevent-related calls that need to do cross-thread + * communication in multithreaded mode (rather than actually doing the work + * in the current thread) are called via "dispatch_" frontends, which are + * also #define-d to directly call the underlying code in singlethreaded mode. + */ +#ifdef USE_THREADS + +void thread_init(int nthreads, struct event_base *main_base); +int dispatch_event_add(int thread, conn *c); +void dispatch_conn_new(int sfd, int init_state, int event_flags, int read_buffer_size, int is_udp); + +/* Lock wrappers for cache functions that are called from main loop. */ +char *mt_add_delta(item *item, const int incr, const int64_t delta, char *buf); +void mt_assoc_move_next_bucket(void); +conn *mt_conn_from_freelist(void); +bool mt_conn_add_to_freelist(conn *c); +char *mt_suffix_from_freelist(void); +bool mt_suffix_add_to_freelist(char *s); +char *mt_defer_delete(item *it, time_t exptime); +int mt_is_listen_thread(void); +item *mt_item_alloc(char *key, size_t nkey, int flags, rel_time_t exptime, int nbytes); +char *mt_item_cachedump(const unsigned int slabs_clsid, const unsigned int limit, unsigned int *bytes); +void mt_item_flush_expired(void); +item *mt_item_get_notedeleted(const char *key, const size_t nkey, bool *delete_locked); +int mt_item_link(item *it); +void mt_item_remove(item *it); +int mt_item_replace(item *it, item *new_it); +char *mt_item_stats(int *bytes); +char *mt_item_stats_sizes(int *bytes); +void mt_item_unlink(item *it); +void mt_item_update(item *it); +void mt_run_deferred_deletes(void); +void *mt_slabs_alloc(size_t size, unsigned int id); +void mt_slabs_free(void *ptr, size_t size, unsigned int id); +int mt_slabs_reassign(unsigned char srcid, unsigned char dstid); +char *mt_slabs_stats(int *buflen); +void mt_stats_lock(void); +void mt_stats_unlock(void); +int mt_store_item(item *item, int comm); + + +# define add_delta(x,y,z,a) mt_add_delta(x,y,z,a) +# define assoc_move_next_bucket() mt_assoc_move_next_bucket() +# define conn_from_freelist() mt_conn_from_freelist() +# define conn_add_to_freelist(x) mt_conn_add_to_freelist(x) +# define suffix_from_freelist() mt_suffix_from_freelist() +# define suffix_add_to_freelist(x) mt_suffix_add_to_freelist(x) +# define defer_delete(x,y) mt_defer_delete(x,y) +# define is_listen_thread() mt_is_listen_thread() +# define item_alloc(x,y,z,a,b) mt_item_alloc(x,y,z,a,b) +# define item_cachedump(x,y,z) mt_item_cachedump(x,y,z) +# define item_flush_expired() mt_item_flush_expired() +# define item_get_notedeleted(x,y,z) mt_item_get_notedeleted(x,y,z) +# define item_link(x) mt_item_link(x) +# define item_remove(x) mt_item_remove(x) +# define item_replace(x,y) mt_item_replace(x,y) +# define item_stats(x) mt_item_stats(x) +# define item_stats_sizes(x) mt_item_stats_sizes(x) +# define item_update(x) mt_item_update(x) +# define item_unlink(x) mt_item_unlink(x) +# define run_deferred_deletes() mt_run_deferred_deletes() +# define slabs_alloc(x,y) mt_slabs_alloc(x,y) +# define slabs_free(x,y,z) mt_slabs_free(x,y,z) +# define slabs_reassign(x,y) mt_slabs_reassign(x,y) +# define slabs_stats(x) mt_slabs_stats(x) +# define store_item(x,y) mt_store_item(x,y) + +# define STATS_LOCK() mt_stats_lock() +# define STATS_UNLOCK() mt_stats_unlock() + +#else /* !USE_THREADS */ + +# define add_delta(x,y,z,a) do_add_delta(x,y,z,a) +# define assoc_move_next_bucket() do_assoc_move_next_bucket() +# define conn_from_freelist() do_conn_from_freelist() +# define conn_add_to_freelist(x) do_conn_add_to_freelist(x) +# define suffix_from_freelist() do_suffix_from_freelist() +# define suffix_add_to_freelist(x) do_suffix_add_to_freelist(x) +# define defer_delete(x,y) do_defer_delete(x,y) +# define dispatch_conn_new(x,y,z,a,b) conn_new(x,y,z,a,b,main_base) +# define dispatch_event_add(t,c) event_add(&(c)->event, 0) +# define is_listen_thread() 1 +# define item_alloc(x,y,z,a,b) do_item_alloc(x,y,z,a,b) +# define item_cachedump(x,y,z) do_item_cachedump(x,y,z) +# define item_flush_expired() do_item_flush_expired() +# define item_get_notedeleted(x,y,z) do_item_get_notedeleted(x,y,z) +# define item_link(x) do_item_link(x) +# define item_remove(x) do_item_remove(x) +# define item_replace(x,y) do_item_replace(x,y) +# define item_stats(x) do_item_stats(x) +# define item_stats_sizes(x) do_item_stats_sizes(x) +# define item_unlink(x) do_item_unlink(x) +# define item_update(x) do_item_update(x) +# define run_deferred_deletes() do_run_deferred_deletes() +# define slabs_alloc(x,y) do_slabs_alloc(x,y) +# define slabs_free(x,y,z) do_slabs_free(x,y,z) +# define slabs_reassign(x,y) do_slabs_reassign(x,y) +# define slabs_stats(x) do_slabs_stats(x) +# define store_item(x,y) do_store_item(x,y) +# define thread_init(x,y) 0 + +# define STATS_LOCK() /**/ +# define STATS_UNLOCK() /**/ + +#endif /* !USE_THREADS */ + +/* If supported, give compiler hints for branch prediction. */ +#if !defined(__GNUC__) || (__GNUC__ == 2 && __GNUC_MINOR__ < 96) +#define __builtin_expect(x, expected_value) (x) +#endif + +#define likely(x) __builtin_expect((x),1) +#define unlikely(x) __builtin_expect((x),0) Index: /tags/1.2.5/README =================================================================== --- /tags/1.2.5/README (revision 513) +++ /tags/1.2.5/README (revision 513) @@ -0,0 +1,25 @@ +Dependencies: + + -- libevent, http://www.monkey.org/~provos/libevent/ (libevent-dev) + +If using Linux, you need a kernel with epoll. Sure, libevent will +work with normal select, but it sucks. + +epoll isn't in Linux 2.4 yet, but there's a backport at: + + http://www.xmailserver.org/linux-patches/nio-improve.html + +You want the epoll-lt patch (level-triggered). + +If you're using MacOS, you'll want libevent 1.1 or higher to deal with +a kqueue bug. + +Also, be warned that the -k (mlockall) option to memcached might be +dangerous when using a large cache. Just make sure the memcached machines +don't swap. memcached does non-blocking network I/O, but not disk. (it +should never go to disk, or you've lost the whole point of it) + +The memcached website is at: + + http://www.danga.com/memcached/ + Index: /tags/1.2.5/assoc.h =================================================================== --- /tags/1.2.5/assoc.h (revision 509) +++ /tags/1.2.5/assoc.h (revision 509) @@ -0,0 +1,7 @@ +/* associative array */ +void assoc_init(void); +item *assoc_find(const char *key, const size_t nkey); +int assoc_insert(item *item); +void assoc_delete(const char *key, const size_t nkey); +void do_assoc_move_next_bucket(void); +uint32_t hash( const void *key, size_t length, const uint32_t initval); Index: /tags/1.2.5/devtools/bench_noreply.pl =================================================================== --- /tags/1.2.5/devtools/bench_noreply.pl (revision 708) +++ /tags/1.2.5/devtools/bench_noreply.pl (revision 708) @@ -0,0 +1,51 @@ +#! /usr/bin/perl +# +use warnings; +use strict; + +use IO::Socket::INET; + +use FindBin; + +@ARGV == 1 or @ARGV == 2 + or die "Usage: $FindBin::Script HOST:PORT [COUNT]\n"; + +# Note that it's better to run the test over the wire, because for +# localhost the task may become CPU bound. +my $addr = $ARGV[0]; +my $count = $ARGV[1] || 10_000; + +my $sock = IO::Socket::INET->new(PeerAddr => $addr, + Timeout => 3); +die "$!\n" unless $sock; + + +# By running 'noreply' test first we also ensure there are no reply +# packets left in the network. +foreach my $noreply (1, 0) { + use Time::HiRes qw(gettimeofday tv_interval); + + print "'noreply' is ", $noreply ? "enabled" : "disabled", ":\n"; + my $param = $noreply ? 'noreply' : ''; + my $start = [gettimeofday]; + foreach (1 .. $count) { + print $sock "add foo 0 0 1 $param\r\n1\r\n"; + scalar<$sock> unless $noreply; + print $sock "set foo 0 0 1 $param\r\n1\r\n"; + scalar<$sock> unless $noreply; + print $sock "replace foo 0 0 1 $param\r\n1\r\n"; + scalar<$sock> unless $noreply; + print $sock "append foo 0 0 1 $param\r\n1\r\n"; + scalar<$sock> unless $noreply; + print $sock "prepend foo 0 0 1 $param\r\n1\r\n"; + scalar<$sock> unless $noreply; + print $sock "incr foo 1 $param\r\n"; + scalar<$sock> unless $noreply; + print $sock "decr foo 1 $param\r\n"; + scalar<$sock> unless $noreply; + print $sock "delete foo $param\r\n"; + scalar<$sock> unless $noreply; + } + my $end = [gettimeofday]; + printf("update commands: %.2f secs\n\n", tv_interval($start, $end)); +} Index: /tags/1.2.5/devtools/svn-tarballs.pl =================================================================== --- /tags/1.2.5/devtools/svn-tarballs.pl (revision 323) +++ /tags/1.2.5/devtools/svn-tarballs.pl (revision 323) @@ -0,0 +1,49 @@ +#!/usr/bin/perl + +use strict; +use FindBin qw($Bin); + +my %branch = ( + '1.2.x' => "http://code.sixapart.com/svn/memcached/trunk/server", + '1.1.x' => "http://code.sixapart.com/svn/memcached/branches/memcached-1.1.x", + ); + +foreach my $b (keys %branch) { + chdir $Bin or die; + my $url = $branch{$b}; + my $out = `svn info $b`; + unless ($out =~ /^URL: (.+)/m && $1 eq $url) { + system("rm -rf $b"); + system("svn", "co", $url, $b) + and die "Failed to checkout $url\n"; + } else { + chdir "$Bin/$b" or die; + system("svn up") and die "Failed to svn up"; + } + + chdir "$Bin/$b" or die; + $out = `svn info .`; + + my ($maxrev) = $out =~ /^Last Changed Rev: (\d+)/m + or die "No max rev?"; + + print "$b = $maxrev\n"; + my $distfile = "memcached-$b-svn$maxrev.tar.gz"; + next if -f $distfile && -s _; + + open(my $fh, "configure.ac") or die "no configure.ac in $b?"; + my $ac = do { local $/; <$fh>; }; + close($fh); + $ac =~ s!AC_INIT\(memcached,.+?\)!AC_INIT(memcached, $b-svn$maxrev, brad\@danga.com)! + or die "Failed to replace"; + open (my $fh, ">configure.ac") or die "failed to write configure.ac writeable: $!"; + print $fh $ac; + close ($fh); + + system("./autogen.sh") and die "Autogen failed. Missing autotools?"; + system("./configure") and die "configure failed"; + system("make dist") and die "make dist failed"; + die "Failed to make dist $distfile." unless -s $distfile; +} + + Index: /tags/1.2.5/devtools/clean-whitespace.pl =================================================================== --- /tags/1.2.5/devtools/clean-whitespace.pl (revision 491) +++ /tags/1.2.5/devtools/clean-whitespace.pl (revision 491) @@ -0,0 +1,18 @@ +#!/usr/bin/perl +use strict; +use FindBin qw($Bin); +chdir "$Bin/.." or die; +my @files = (glob("*.h"), glob("*.c"), glob("*.ac")); +foreach my $f (@files) { + open(my $fh, $f) or die; + my $before = do { local $/; <$fh>; }; + close ($fh); + my $after = $before; + $after =~ s/\t/ /g; + $after =~ s/ +$//mg; + $after .= "\n" unless $after =~ /\n$/; + next if $after eq $before; + open(my $fh, ">$f") or die; + print $fh $after; + close($fh); +} Index: /tags/1.2.5/BUILD =================================================================== --- /tags/1.2.5/BUILD (revision 257) +++ /tags/1.2.5/BUILD (revision 257) @@ -0,0 +1,37 @@ +Ideally, you want to make a static binary, otherwise the dynamic +linker pollutes your address space with shared libs right in the +middle. (NOTE: actually, this shouldn't matter so much anymore, now +that we only allocate huge, fixed-size slabs) + +Make sure your libevent has epoll (Linux) or kqueue (BSD) support. +Using poll or select only is slow, and works for testing, but +shouldn't be used for high-traffic memcache installations. + +To build libevent with epoll on Linux, you need two things. First, +you need /usr/include/sys/epoll.h . To get it, you can install the +userspace epoll library, epoll-lib. The link to the latest version +is buried inside +http://www.xmailserver.org/linux-patches/nio-improve.html ; currently +it's http://www.xmailserver.org/linux-patches/epoll-lib-0.9.tar.gz . +If you're having any trouble building/installing it, you can just copy +epoll.h from that tarball to /usr/include/sys as that's the only thing +from there that libevent really needs. + +Secondly, you need to declare syscall numbers of epoll syscalls, so +libevent can use them. Put these declarations somewhere +inside : + +#define __NR_epoll_create 254 +#define __NR_epoll_ctl 255 +#define __NR_epoll_wait 256 + +After this you should be able to build libevent with epoll support. +Once you build/install libevent, you don't need to +compile memcache or link it against libevent. Don't forget that for epoll +support to actually work at runtime you need to use a kernel with epoll +support patch applied, as explained in the README file. + +BSD users are luckier, and will get kqueue support by default. + + + Index: /tags/1.2.5/memcached.spec =================================================================== --- /tags/1.2.5/memcached.spec (revision 741) +++ /tags/1.2.5/memcached.spec (revision 741) @@ -0,0 +1,112 @@ +Name: memcached +Version: 1.2.5 +Release: 1%{?dist} +Summary: High Performance, Distributed Memory Object Cache + +Group: System Environment/Daemons +License: BSD +URL: http://www.danga.com/memcached/ +Source0: http://www.danga.com/memcached/dist/%{name}-%{version}.tar.gz +BuildRoot: %{_tmppath}/%{name}-%{version}-%{release}-root-%(%{__id_u} -n) + +BuildRequires: libevent-devel +BuildRequires: perl(Test::More) +Requires: initscripts +Requires(post): /sbin/chkconfig +Requires(preun): /sbin/chkconfig, /sbin/service +Requires(postun): /sbin/service + +%description +memcached is a high-performance, distributed memory object caching +system, generic in nature, but intended for use in speeding up dynamic +web applications by alleviating database load. + +%prep +%setup -q + + +%build +%configure --enable-threads + +make %{?_smp_mflags} + +%check +make test + +%install +rm -rf %{buildroot} +make install DESTDIR=%{buildroot} + +# remove memcached-debug +rm -f %{buildroot}/%{_bindir}/memcached-debug + +# Perl script for monitoring memcached +install -Dp -m0755 scripts/memcached-tool %{buildroot}%{_bindir}/memcached-tool + +# Init script +install -Dp -m0755 scripts/memcached.sysv %{buildroot}%{_initrddir}/memcached + +# Default configs +mkdir -p %{buildroot}/%{_sysconfdir}/sysconfig +cat <%{buildroot}/%{_sysconfdir}/sysconfig/%{name} +PORT="11211" +USER="nobody" +MAXCONN="1024" +CACHESIZE="64" +OPTIONS="" +EOF + +# pid directory +mkdir -p %{buildroot}/%{_localstatedir}/run/memcached + +%clean +rm -rf %{buildroot} + + +%post +/sbin/chkconfig --add %{name} + +%preun +if [ "$1" = 0 ] ; then + /sbin/service %{name} stop > /dev/null 2>&1 + /sbin/chkconfig --del %{name} +fi +exit 0 + +%postun +if [ "$1" -ge 1 ]; then + /sbin/service %{name} condrestart > /dev/null 2>&1 +fi +exit 0 + + +%files +%defattr(-,root,root,-) +%doc AUTHORS ChangeLog COPYING NEWS README TODO doc/CONTRIBUTORS doc/*.txt +%config(noreplace) %{_sysconfdir}/sysconfig/%{name} + +%dir %attr(750,nobody,nobody) %{_localstatedir}/run/memcached +%{_bindir}/memcached-tool +%{_bindir}/memcached +%{_mandir}/man1/memcached.1* +%{_initrddir}/memcached + + +%changelog +* Wed Jul 4 2007 Paul Lindner - 1.2.2-5 +- Use /var/run/memcached/ directory to hold PID file + +* Sat May 12 2007 Paul Lindner - 1.2.2-4 +- Remove tabs from spec file, rpmlint reports no more errors + +* Thu May 10 2007 Paul Lindner - 1.2.2-3 +- Enable build-time regression tests +- add dependency on initscripts +- remove memcached-debug (not needed in dist) +- above suggestions from Bernard Johnson + +* Mon May 7 2007 Paul Lindner - 1.2.2-2 +- Tidyness improvements suggested by Ruben Kerkhof in bugzilla #238994 + +* Fri May 4 2007 Paul Lindner - 1.2.2-1 +- Initial spec file created via rpmdev-newspec Index: /tags/1.2.5/items.c =================================================================== --- /tags/1.2.5/items.c (revision 740) +++ /tags/1.2.5/items.c (revision 740) @@ -0,0 +1,482 @@ +/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */ +/* $Id$ */ +#include "memcached.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/* Forward Declarations */ +static void item_link_q(item *it); +static void item_unlink_q(item *it); +static uint64_t get_cas_id(); + +/* + * We only reposition items in the LRU queue if they haven't been repositioned + * in this many seconds. That saves us from churning on frequently-accessed + * items. + */ +#define ITEM_UPDATE_INTERVAL 60 + +#define LARGEST_ID 255 +typedef struct { + unsigned int evicted; + unsigned int outofmemory; +} itemstats_t; + +static item *heads[LARGEST_ID]; +static item *tails[LARGEST_ID]; +static itemstats_t itemstats[LARGEST_ID]; +static unsigned int sizes[LARGEST_ID]; + +void item_init(void) { + int i; + memset(itemstats, 0, sizeof(itemstats_t) * LARGEST_ID); + for(i = 0; i < LARGEST_ID; i++) { + heads[i] = NULL; + tails[i] = NULL; + sizes[i] = 0; + } +} + +/* Get the next CAS id for a new item. */ +uint64_t get_cas_id() { + static uint64_t cas_id = 0; + return ++cas_id; +} + +/* Enable this for reference-count debugging. */ +#if 0 +# define DEBUG_REFCNT(it,op) \ + fprintf(stderr, "item %x refcnt(%c) %d %c%c%c\n", \ + it, op, it->refcount, \ + (it->it_flags & ITEM_LINKED) ? 'L' : ' ', \ + (it->it_flags & ITEM_SLABBED) ? 'S' : ' ', \ + (it->it_flags & ITEM_DELETED) ? 'D' : ' ') +#else +# define DEBUG_REFCNT(it,op) while(0) +#endif + +/** + * Generates the variable-sized part of the header for an object. + * + * key - The key + * nkey - The length of the key + * flags - key flags + * nbytes - Number of bytes to hold value and addition CRLF terminator + * suffix - Buffer for the "VALUE" line suffix (flags, size). + * nsuffix - The length of the suffix is stored here. + * + * Returns the total size of the header. + */ +static size_t item_make_header(const uint8_t nkey, const int flags, const int nbytes, + char *suffix, uint8_t *nsuffix) { + /* suffix is defined at 40 chars elsewhere.. */ + *nsuffix = (uint8_t) snprintf(suffix, 40, " %d %d\r\n", flags, nbytes - 2); + return sizeof(item) + nkey + *nsuffix + nbytes; +} + +/*@null@*/ +item *do_item_alloc(char *key, const size_t nkey, const int flags, const rel_time_t exptime, const int nbytes) { + uint8_t nsuffix; + item *it; + char suffix[40]; + size_t ntotal = item_make_header(nkey + 1, flags, nbytes, suffix, &nsuffix); + + unsigned int id = slabs_clsid(ntotal); + if (id == 0) + return 0; + + it = slabs_alloc(ntotal, id); + if (it == 0) { + int tries = 50; + item *search; + + /* If requested to not push old items out of cache when memory runs out, + * we're out of luck at this point... + */ + + if (settings.evict_to_free == 0) { + itemstats[id].outofmemory++; + return NULL; + } + + /* + * try to get one off the right LRU + * don't necessariuly unlink the tail because it may be locked: refcount>0 + * search up from tail an item with refcount==0 and unlink it; give up after 50 + * tries + */ + + if (tails[id] == 0) { + itemstats[id].outofmemory++; + return NULL; + } + + for (search = tails[id]; tries > 0 && search != NULL; tries--, search=search->prev) { + if (search->refcount == 0) { + if (search->exptime == 0 || search->exptime > current_time) { + itemstats[id].evicted++; + STATS_LOCK(); + stats.evictions++; + STATS_UNLOCK(); + } + do_item_unlink(search); + break; + } + } + it = slabs_alloc(ntotal, id); + if (it == 0) { + itemstats[id].outofmemory++; + return NULL; + } + } + + assert(it->slabs_clsid == 0); + + it->slabs_clsid = id; + + assert(it != heads[it->slabs_clsid]); + + it->next = it->prev = it->h_next = 0; + it->refcount = 1; /* the caller will have a reference */ + DEBUG_REFCNT(it, '*'); + it->it_flags = 0; + it->nkey = nkey; + it->nbytes = nbytes; + strcpy(ITEM_key(it), key); + it->exptime = exptime; + memcpy(ITEM_suffix(it), suffix, (size_t)nsuffix); + it->nsuffix = nsuffix; + return it; +} + +void item_free(item *it) { + size_t ntotal = ITEM_ntotal(it); + unsigned int clsid; + assert((it->it_flags & ITEM_LINKED) == 0); + assert(it != heads[it->slabs_clsid]); + assert(it != tails[it->slabs_clsid]); + assert(it->refcount == 0); + + /* so slab size changer can tell later if item is already free or not */ + clsid = it->slabs_clsid; + it->slabs_clsid = 0; + it->it_flags |= ITEM_SLABBED; + DEBUG_REFCNT(it, 'F'); + slabs_free(it, ntotal, clsid); +} + +/** + * Returns true if an item will fit in the cache (its size does not exceed + * the maximum for a cache entry.) + */ +bool item_size_ok(const size_t nkey, const int flags, const int nbytes) { + char prefix[40]; + uint8_t nsuffix; + + return slabs_clsid(item_make_header(nkey + 1, flags, nbytes, + prefix, &nsuffix)) != 0; +} + +static void item_link_q(item *it) { /* item is the new head */ + item **head, **tail; + /* always true, warns: assert(it->slabs_clsid <= LARGEST_ID); */ + assert((it->it_flags & ITEM_SLABBED) == 0); + + head = &heads[it->slabs_clsid]; + tail = &tails[it->slabs_clsid]; + assert(it != *head); + assert((*head && *tail) || (*head == 0 && *tail == 0)); + it->prev = 0; + it->next = *head; + if (it->next) it->next->prev = it; + *head = it; + if (*tail == 0) *tail = it; + sizes[it->slabs_clsid]++; + return; +} + +static void item_unlink_q(item *it) { + item **head, **tail; + /* always true, warns: assert(it->slabs_clsid <= LARGEST_ID); */ + head = &heads[it->slabs_clsid]; + tail = &tails[it->slabs_clsid]; + + if (*head == it) { + assert(it->prev == 0); + *head = it->next; + } + if (*tail == it) { + assert(it->next == 0); + *tail = it->prev; + } + assert(it->next != it); + assert(it->prev != it); + + if (it->next) it->next->prev = it->prev; + if (it->prev) it->prev->next = it->next; + sizes[it->slabs_clsid]--; + return; +} + +int do_item_link(item *it) { + assert((it->it_flags & (ITEM_LINKED|ITEM_SLABBED)) == 0); + assert(it->nbytes < (1024 * 1024)); /* 1MB max size */ + it->it_flags |= ITEM_LINKED; + it->time = current_time; + assoc_insert(it); + + STATS_LOCK(); + stats.curr_bytes += ITEM_ntotal(it); + stats.curr_items += 1; + stats.total_items += 1; + STATS_UNLOCK(); + + /* Allocate a new CAS ID on link. */ + it->cas_id = get_cas_id(); + + item_link_q(it); + + return 1; +} + +void do_item_unlink(item *it) { + if ((it->it_flags & ITEM_LINKED) != 0) { + it->it_flags &= ~ITEM_LINKED; + STATS_LOCK(); + stats.curr_bytes -= ITEM_ntotal(it); + stats.curr_items -= 1; + STATS_UNLOCK(); + assoc_delete(ITEM_key(it), it->nkey); + item_unlink_q(it); + if (it->refcount == 0) item_free(it); + } +} + +void do_item_remove(item *it) { + assert((it->it_flags & ITEM_SLABBED) == 0); + if (it->refcount != 0) { + it->refcount--; + DEBUG_REFCNT(it, '-'); + } + assert((it->it_flags & ITEM_DELETED) == 0 || it->refcount != 0); + if (it->refcount == 0 && (it->it_flags & ITEM_LINKED) == 0) { + item_free(it); + } +} + +void do_item_update(item *it) { + if (it->time < current_time - ITEM_UPDATE_INTERVAL) { + assert((it->it_flags & ITEM_SLABBED) == 0); + + if ((it->it_flags & ITEM_LINKED) != 0) { + item_unlink_q(it); + it->time = current_time; + item_link_q(it); + } + } +} + +int do_item_replace(item *it, item *new_it) { + assert((it->it_flags & ITEM_SLABBED) == 0); + + do_item_unlink(it); + return do_item_link(new_it); +} + +/*@null@*/ +char *do_item_cachedump(const unsigned int slabs_clsid, const unsigned int limit, unsigned int *bytes) { + unsigned int memlimit = 2 * 1024 * 1024; /* 2MB max response size */ + char *buffer; + unsigned int bufcurr; + item *it; + unsigned int len; + unsigned int shown = 0; + char temp[512]; + + if (slabs_clsid > LARGEST_ID) return NULL; + it = heads[slabs_clsid]; + + buffer = malloc((size_t)memlimit); + if (buffer == 0) return NULL; + bufcurr = 0; + + while (it != NULL && (limit == 0 || shown < limit)) { + len = snprintf(temp, sizeof(temp), "ITEM %s [%d b; %lu s]\r\n", ITEM_key(it), it->nbytes - 2, it->exptime + stats.started); + if (bufcurr + len + 6 > memlimit) /* 6 is END\r\n\0 */ + break; + strcpy(buffer + bufcurr, temp); + bufcurr += len; + shown++; + it = it->next; + } + + memcpy(buffer + bufcurr, "END\r\n", 6); + bufcurr += 5; + + *bytes = bufcurr; + return buffer; +} + +char *do_item_stats(int *bytes) { + size_t bufleft = (size_t) LARGEST_ID * 160; + char *buffer = malloc(bufleft); + char *bufcurr = buffer; + rel_time_t now = current_time; + int i; + int linelen; + + if (buffer == NULL) { + return NULL; + } + + for (i = 0; i < LARGEST_ID; i++) { + if (tails[i] != NULL) { + linelen = snprintf(bufcurr, bufleft, + "STAT items:%d:number %u\r\n" + "STAT items:%d:age %u\r\n" + "STAT items:%d:evicted %u\r\n" + "STAT items:%d:outofmemory %u\r\n", + i, sizes[i], i, now - tails[i]->time, i, + itemstats[i].evicted, i, itemstats[i].outofmemory); + if (linelen + sizeof("END\r\n") < bufleft) { + bufcurr += linelen; + bufleft -= linelen; + } + else { + /* The caller didn't allocate enough buffer space. */ + break; + } + } + } + memcpy(bufcurr, "END\r\n", 6); + bufcurr += 5; + + *bytes = bufcurr - buffer; + return buffer; +} + +/** dumps out a list of objects of each size, with granularity of 32 bytes */ +/*@null@*/ +char* do_item_stats_sizes(int *bytes) { + const int num_buckets = 32768; /* max 1MB object, divided into 32 bytes size buckets */ + unsigned int *histogram = (unsigned int *)malloc((size_t)num_buckets * sizeof(int)); + char *buf = (char *)malloc(2 * 1024 * 1024); /* 2MB max response size */ + int i; + + if (histogram == 0 || buf == 0) { + if (histogram) free(histogram); + if (buf) free(buf); + return NULL; + } + + /* build the histogram */ + memset(histogram, 0, (size_t)num_buckets * sizeof(int)); + for (i = 0; i < LARGEST_ID; i++) { + item *iter = heads[i]; + while (iter) { + int ntotal = ITEM_ntotal(iter); + int bucket = ntotal / 32; + if ((ntotal % 32) != 0) bucket++; + if (bucket < num_buckets) histogram[bucket]++; + iter = iter->next; + } + } + + /* write the buffer */ + *bytes = 0; + for (i = 0; i < num_buckets; i++) { + if (histogram[i] != 0) { + *bytes += sprintf(&buf[*bytes], "%d %u\r\n", i * 32, histogram[i]); + } + } + *bytes += sprintf(&buf[*bytes], "END\r\n"); + free(histogram); + return buf; +} + +/** returns true if a deleted item's delete-locked-time is over, and it + should be removed from the namespace */ +bool item_delete_lock_over (item *it) { + assert(it->it_flags & ITEM_DELETED); + return (current_time >= it->exptime); +} + +/** wrapper around assoc_find which does the lazy expiration/deletion logic */ +item *do_item_get_notedeleted(const char *key, const size_t nkey, bool *delete_locked) { + item *it = assoc_find(key, nkey); + if (delete_locked) *delete_locked = false; + if (it != NULL && (it->it_flags & ITEM_DELETED)) { + /* it's flagged as delete-locked. let's see if that condition + is past due, and the 5-second delete_timer just hasn't + gotten to it yet... */ + if (!item_delete_lock_over(it)) { + if (delete_locked) *delete_locked = true; + it = NULL; + } + } + if (it != NULL && settings.oldest_live != 0 && settings.oldest_live <= current_time && + it->time <= settings.oldest_live) { + do_item_unlink(it); /* MTSAFE - cache_lock held */ + it = NULL; + } + if (it != NULL && it->exptime != 0 && it->exptime <= current_time) { + do_item_unlink(it); /* MTSAFE - cache_lock held */ + it = NULL; + } + + if (it != NULL) { + it->refcount++; + DEBUG_REFCNT(it, '+'); + } + return it; +} + +item *item_get(const char *key, const size_t nkey) { + return item_get_notedeleted(key, nkey, 0); +} + +/** returns an item whether or not it's delete-locked or expired. */ +item *do_item_get_nocheck(const char *key, const size_t nkey) { + item *it = assoc_find(key, nkey); + if (it) { + it->refcount++; + DEBUG_REFCNT(it, '+'); + } + return it; +} + +/* expires items that are more recent than the oldest_live setting. */ +void do_item_flush_expired(void) { + int i; + item *iter, *next; + if (settings.oldest_live == 0) + return; + for (i = 0; i < LARGEST_ID; i++) { + /* The LRU is sorted in decreasing time order, and an item's timestamp + * is never newer than its last access time, so we only need to walk + * back until we hit an item older than the oldest_live time. + * The oldest_live checking will auto-expire the remaining items. + */ + for (iter = heads[i]; iter != NULL; iter = next) { + if (iter->time >= settings.oldest_live) { + next = iter->next; + if ((iter->it_flags & ITEM_SLABBED) == 0) { + do_item_unlink(iter); + } + } else { + /* We've hit the first old item. Continue to the next queue. */ + break; + } + } + } +} Index: /tags/1.2.5/configure.ac =================================================================== --- /tags/1.2.5/configure.ac (revision 741) +++ /tags/1.2.5/configure.ac (revision 741) @@ -0,0 +1,206 @@ +AC_PREREQ(2.52) +AC_INIT(memcached, 1.2.5, brad@danga.com) +AC_CANONICAL_SYSTEM +AC_CONFIG_SRCDIR(memcached.c) +AM_INIT_AUTOMAKE(AC_PACKAGE_NAME, AC_PACKAGE_VERSION) +AM_CONFIG_HEADER(config.h) + +AC_PROG_CC +AM_PROG_CC_C_O +AC_PROG_INSTALL + +AC_ARG_ENABLE(64bit, + [AS_HELP_STRING([--enable-64bit],[build 64bit verison])]) +if test "x$enable_64bit" == "xyes" +then + org_cflags=$CFLAGS + CFLAGS=-m64 + AC_RUN_IFELSE( + [AC_LANG_PROGRAM([], [dnl +return sizeof(void*) == 8 ? 0 : 1; + ]) + ],[ + CFLAGS="-m64 $org_cflags" + ],[ + AC_MSG_ERROR([Don't know how to build a 64-bit object.]) + ]) +fi + +trylibeventdir="" +AC_ARG_WITH(libevent, + [ --with-libevent=PATH Specify path to libevent installation ], + [ + if test "x$withval" != "xno" ; then + trylibeventdir=$withval + fi + ] +) + +dnl ------------------------------------------------------ +dnl libevent detection. swiped from Tor. modified a bit. + +LIBEVENT_URL=http://www.monkey.org/~provos/libevent/ + +AC_CACHE_CHECK([for libevent directory], ac_cv_libevent_dir, [ + saved_LIBS="$LIBS" + saved_LDFLAGS="$LDFLAGS" + saved_CPPFLAGS="$CPPFLAGS" + le_found=no + for ledir in $trylibeventdir "" $prefix /usr/local ; do + LDFLAGS="$saved_LDFLAGS" + LIBS="$saved_LIBS -levent" + + # Skip the directory if it isn't there. + if test ! -z "$ledir" -a ! -d "$ledir" ; then + continue; + fi + if test ! -z "$ledir" ; then + if test -d "$ledir/lib" ; then + LDFLAGS="-L$ledir/lib $LDFLAGS" + else + LDFLAGS="-L$ledir $LDFLAGS" + fi + if test -d "$ledir/include" ; then + CPPFLAGS="-I$ledir/include $CPPFLAGS" + else + CPPFLAGS="-I$ledir $CPPFLAGS" + fi + fi + # Can I compile and link it? + AC_TRY_LINK([#include +#include +#include ], [ event_init(); ], + [ libevent_linked=yes ], [ libevent_linked=no ]) + if test $libevent_linked = yes; then + if test ! -z "$ledir" ; then + ac_cv_libevent_dir=$ledir + else + ac_cv_libevent_dir="(system)" + fi + le_found=yes + break + fi + done + LIBS="$saved_LIBS" + LDFLAGS="$saved_LDFLAGS" + CPPFLAGS="$saved_CPPFLAGS" + if test $le_found = no ; then + AC_MSG_ERROR([libevent is required. You can get it from $LIBEVENT_URL + + If it's already installed, specify its path using --with-libevent=/dir/ +]) + fi +]) +LIBS="$LIBS -levent" +if test $ac_cv_libevent_dir != "(system)"; then + if test -d "$ac_cv_libevent_dir/lib" ; then + LDFLAGS="-L$ac_cv_libevent_dir/lib $LDFLAGS" + le_libdir="$ac_cv_libevent_dir/lib" + else + LDFLAGS="-L$ac_cv_libevent_dir $LDFLAGS" + le_libdir="$ac_cv_libevent_dir" + fi + if test -d "$ac_cv_libevent_dir/include" ; then + CPPFLAGS="-I$ac_cv_libevent_dir/include $CPPFLAGS" + else + CPPFLAGS="-I$ac_cv_libevent_dir $CPPFLAGS" + fi +fi + +dnl ---------------------------------------------------------------------------- + +AC_SEARCH_LIBS(socket, socket) +AC_SEARCH_LIBS(gethostbyname, nsl) +AC_SEARCH_LIBS(mallinfo, malloc) + +AC_CHECK_FUNC(daemon,AC_DEFINE([HAVE_DAEMON],,[Define this if you have daemon()]),[AC_LIBOBJ(daemon)]) + +AC_HEADER_STDBOOL +AC_C_CONST +AC_CHECK_HEADER(malloc.h, AC_DEFINE(HAVE_MALLOC_H,,[do we have malloc.h?])) +AC_CHECK_MEMBER([struct mallinfo.arena], [ + AC_DEFINE(HAVE_STRUCT_MALLINFO,,[do we have stuct mallinfo?]) + ], ,[ +# include + ] +) + +dnl From licq: Copyright (c) 2000 Dirk Mueller +dnl Check if the type socklen_t is defined anywhere +AC_DEFUN([AC_C_SOCKLEN_T], +[AC_CACHE_CHECK(for socklen_t, ac_cv_c_socklen_t, +[ + AC_TRY_COMPILE([ + #include + #include + ],[ + socklen_t foo; + ],[ + ac_cv_c_socklen_t=yes + ],[ + ac_cv_c_socklen_t=no + ]) +]) +if test $ac_cv_c_socklen_t = no; then + AC_DEFINE(socklen_t, int, [define to int if socklen_t not available]) +fi +]) + +AC_C_SOCKLEN_T + +dnl Check if we're a little-endian or a big-endian system, needed by hash code +AC_DEFUN([AC_C_ENDIAN], +[AC_CACHE_CHECK(for endianness, ac_cv_c_endian, +[ + AC_RUN_IFELSE( + [AC_LANG_PROGRAM([], [dnl + long val = 1; + char *c = (char *) &val; + exit(*c == 1); + ]) + ],[ + ac_cv_c_endian=big + ],[ + ac_cv_c_endian=little + ]) +]) +if test $ac_cv_c_endian = big; then + AC_DEFINE(ENDIAN_BIG, 1, [machine is bigendian]) +fi +if test $ac_cv_c_endian = little; then + AC_DEFINE(ENDIAN_LITTLE, 1, [machine is littleendian]) +fi +]) + +AC_C_ENDIAN + +dnl Check whether the user wants threads or not +AC_ARG_ENABLE(threads, + [AS_HELP_STRING([--enable-threads],[support multithreaded execution])]) +if test "x$enable_threads" == "xyes"; then + AC_SEARCH_LIBS(pthread_create, pthread) + if test "x$ac_cv_search_pthread_create" != "xno"; then + AC_DEFINE([USE_THREADS],,[Define this if you want to use pthreads]) + dnl Sun compilers need the -mt flag! + AC_RUN_IFELSE( + [AC_LANG_PROGRAM([], [dnl +#ifdef __SUNPRO_C + return 0; +#else + return 1; +#endif + ]) + ],[ + CFLAGS="-mt $CFLAGS" + ]) + else + AC_MSG_ERROR([Can't enable threads without the POSIX thread library.]) + fi +fi + +AC_CHECK_FUNCS(mlockall) +AC_CHECK_FUNCS(getpagesizes) +AC_CHECK_FUNCS(memcntl) + +AC_CONFIG_FILES(Makefile doc/Makefile) +AC_OUTPUT Index: /tags/1.2.5/t/bogus-commands.t =================================================================== --- /tags/1.2.5/t/bogus-commands.t (revision 348) +++ /tags/1.2.5/t/bogus-commands.t (revision 348) @@ -0,0 +1,13 @@ +#!/usr/bin/perl + +use strict; +use Test::More tests => 1; +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); +my $sock = $server->sock; + +print $sock "boguscommand slkdsldkfjsd\r\n"; +is(scalar <$sock>, "ERROR\r\n", "got error back"); Index: /tags/1.2.5/t/00-startup.t =================================================================== --- /tags/1.2.5/t/00-startup.t (revision 534) +++ /tags/1.2.5/t/00-startup.t (revision 534) @@ -0,0 +1,21 @@ +#!/usr/bin/perl + +use strict; +use Test::More tests => 3; +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); + +ok($server, "started the server"); + +eval { + my $server = new_memcached("-l fooble"); +}; +ok($@, "Died with illegal -l args"); + +eval { + my $server = new_memcached("-l 127.0.0.1"); +}; +is($@,'', "-l 127.0.0.1 works"); Index: /tags/1.2.5/t/maxconns.t =================================================================== --- /tags/1.2.5/t/maxconns.t (revision 557) +++ /tags/1.2.5/t/maxconns.t (revision 557) @@ -0,0 +1,26 @@ +#!/usr/bin/perl + +use strict; +use warnings; + +use Test::More tests => 11; + +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + + +# start up a server with 10 maximum connections +my $server = new_memcached('-c 10'); +my $sock = $server->sock; +my @sockets; + +ok(defined($sock), 'Connection 0'); +push (@sockets, $sock); + + +foreach my $conn (1..10) { + $sock = $server->new_sock; + ok(defined($sock), "Made connection $conn"); + push(@sockets, $sock); +} Index: /tags/1.2.5/t/flush-all.t =================================================================== --- /tags/1.2.5/t/flush-all.t (revision 666) +++ /tags/1.2.5/t/flush-all.t (revision 666) @@ -0,0 +1,44 @@ +#!/usr/bin/perl + +use strict; +use Test::More tests => 14; +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); +my $sock = $server->sock; +my $expire; + +print $sock "set foo 0 0 6\r\nfooval\r\n"; +is(scalar <$sock>, "STORED\r\n", "stored foo"); + +mem_get_is($sock, "foo", "fooval"); +print $sock "flush_all\r\n"; +is(scalar <$sock>, "OK\r\n", "did flush_all"); +mem_get_is($sock, "foo", undef); + +# Test flush_all with zero delay. +print $sock "set foo 0 0 6\r\nfooval\r\n"; +is(scalar <$sock>, "STORED\r\n", "stored foo"); + +mem_get_is($sock, "foo", "fooval"); +print $sock "flush_all 0\r\n"; +is(scalar <$sock>, "OK\r\n", "did flush_all"); +mem_get_is($sock, "foo", undef); + +# check that flush_all doesn't blow away items that immediately get set +print $sock "set foo 0 0 3\r\nnew\r\n"; +is(scalar <$sock>, "STORED\r\n", "stored foo = 'new'"); +mem_get_is($sock, "foo", 'new'); + +# and the other form, specifying a flush_all time... +my $expire = time() + 2; +print $sock "flush_all $expire\r\n"; +is(scalar <$sock>, "OK\r\n", "did flush_all in future"); + +print $sock "set foo 0 0 4\r\n1234\r\n"; +is(scalar <$sock>, "STORED\r\n", "stored foo = '1234'"); +mem_get_is($sock, "foo", '1234'); +sleep(2.2); +mem_get_is($sock, "foo", undef); Index: /tags/1.2.5/t/udp.t =================================================================== --- /tags/1.2.5/t/udp.t (revision 378) +++ /tags/1.2.5/t/udp.t (revision 378) @@ -0,0 +1,131 @@ +#!/usr/bin/perl + +use strict; +use Test::More tests => 33; +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); +my $sock = $server->sock; + +# set foo (and should get it) +print $sock "set foo 0 0 6\r\nfooval\r\n"; +is(scalar <$sock>, "STORED\r\n", "stored foo"); +mem_get_is($sock, "foo", "fooval"); + +my $usock = $server->new_udp_sock + or die "Can't bind : $@\n"; + +# test all the steps, one by one: +test_single($usock); + +# testing sequence numbers +for my $offt (1, 1, 2) { + my $seq = 160 + $offt; + my $res = send_udp_request($usock, $seq, "get foo\r\n"); + ok($res, "got result"); + is(keys %$res, 1, "one key (one packet)"); + ok($res->{0}, "only got seq number 0"); + is(substr($res->{0}, 8), "VALUE foo 0 6\r\nfooval\r\nEND\r\n"); + is(hexify(substr($res->{0}, 0, 2)), hexify(pack("n", $seq)), "sequence number in response ($seq) is correct"); +} + +# testing non-existent stuff +my $res = send_udp_request($usock, 404, "get notexist\r\n"); +ok($res, "got result"); +is(keys %$res, 1, "one key (one packet)"); +ok($res->{0}, "only got seq number 0"); +is(hexify(substr($res->{0}, 0, 2)), hexify(pack("n", 404)), "sequence number 404 correct"); +is(substr($res->{0}, 8), "END\r\n"); + +# test multi-packet response +{ + my $big = "abcd" x 1024; + my $len = length $big; + print $sock "set big 0 0 $len\r\n$big\r\n"; + is(scalar <$sock>, "STORED\r\n", "stored big"); + mem_get_is($sock, "big", $big, "big value matches"); + my $res = send_udp_request($usock, 999, "get big\r\n"); + is(scalar keys %$res, 3, "three packet response"); + like($res->{0}, qr/VALUE big 0 4096/, "first packet has value line"); + like($res->{2}, qr/\r\nEND\r\n/, "last packet has end"); + is(hexify(substr($res->{1}, 0, 2)), hexify(pack("n", 999)), "sequence number of middle packet is correct"); +} + +sub test_single { + my $usock = shift; + my $req = pack("nnnn", 45, 0, 1, 0); # request id (opaque), seq num, #packets, reserved (must be 0) + $req .= "get foo\r\n"; + ok(defined send($usock, $req, 0), "sent request"); + + my $rin = ''; + vec($rin, fileno($usock), 1) = 1; + my $rout; + ok(select($rout = $rin, undef, undef, 2.0), "got readability"); + + my $sender; + my $res; + $sender = $usock->recv($res, 1500, 0); + + my $id = pack("n", 45); + is(hexify(substr($res, 0, 8)), hexify($id) . '0000' . '0001' . '0000', "header is correct"); + is(length $res, 36, ''); + is(substr($res, 8), "VALUE foo 0 6\r\nfooval\r\nEND\r\n", "payload is as expected"); +} + +sub hexify { + my $val = shift; + $val =~ s/(.)/sprintf("%02x", ord($1))/egs; + return $val; +} + +# returns undef on select timeout, or hashref of "seqnum" -> payload (including headers) +sub send_udp_request { + my ($sock, $reqid, $req) = @_; + + my $pkt = pack("nnnn", $reqid, 0, 1, 0); # request id (opaque), seq num, #packets, reserved (must be 0) + $pkt .= $req; + my $fail = sub { + my $msg = shift; + warn " FAILING send_udp because: $msg\n"; + return undef; + }; + return $fail->("send") unless send($sock, $pkt, 0); + + my $ret = {}; + + my $got = 0; # packets got + my $numpkts = undef; + + while (!defined($numpkts) || $got < $numpkts) { + my $rin = ''; + vec($rin, fileno($sock), 1) = 1; + my $rout; + return $fail->("timeout after $got packets") unless + select($rout = $rin, undef, undef, 1.5); + + my $res; + my $sender = $sock->recv($res, 1500, 0); + my ($resid, $seq, $this_numpkts, $resv) = unpack("nnnn", substr($res, 0, 8)); + die "Response ID of $resid doesn't match request if of $reqid" unless $resid == $reqid; + die "Reserved area not zero" unless $resv == 0; + die "num packets changed midstream!" if defined $numpkts && $this_numpkts != $numpkts; + $numpkts = $this_numpkts; + $ret->{$seq} = $res; + $got++; + } + return $ret; +} + +__END__ +$sender = recv($usock, $ans, 1050, 0); + +__END__ +$usock->send + + + ($hispaddr = recv(SOCKET, $rtime, 4, 0)) || die "recv: $!"; +($port, $hisiaddr) = sockaddr_in($hispaddr); +$host = gethostbyaddr($hisiaddr, AF_INET); +$histime = unpack("N", $rtime) - $SECS_of_70_YEARS ; Index: /tags/1.2.5/t/noreply.t =================================================================== --- /tags/1.2.5/t/noreply.t (revision 708) +++ /tags/1.2.5/t/noreply.t (revision 708) @@ -0,0 +1,53 @@ +#!/usr/bin/perl + +use strict; +use Test::More tests => 10; +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + + +my $server = new_memcached(); +my $sock = $server->sock; + + +# Test that commands can take 'noreply' parameter. +print $sock "flush_all noreply\r\n"; +print $sock "flush_all 0 noreply\r\n"; + +print $sock "verbosity 0 noreply\r\n"; + +print $sock "add noreply:foo 0 0 1 noreply\r\n1\r\n"; +mem_get_is($sock, "noreply:foo", "1"); + +print $sock "set noreply:foo 0 0 1 noreply\r\n2\r\n"; +mem_get_is($sock, "noreply:foo", "2"); + +print $sock "replace noreply:foo 0 0 1 noreply\r\n3\r\n"; +mem_get_is($sock, "noreply:foo", "3"); + +print $sock "append noreply:foo 0 0 1 noreply\r\n4\r\n"; +mem_get_is($sock, "noreply:foo", "34"); + +print $sock "prepend noreply:foo 0 0 1 noreply\r\n5\r\n"; +my @result = mem_gets($sock, "noreply:foo"); +ok($result[1] eq "534"); + +print $sock "cas noreply:foo 0 0 1 $result[0] noreply\r\n6\r\n"; +mem_get_is($sock, "noreply:foo", "6"); + +print $sock "incr noreply:foo 3 noreply\r\n"; +mem_get_is($sock, "noreply:foo", "9"); + +print $sock "decr noreply:foo 2 noreply\r\n"; +mem_get_is($sock, "noreply:foo", "7"); + +print $sock "delete noreply:foo noreply\r\n"; +mem_get_is($sock, "noreply:foo"); + +# Test that delete accepts both