Index: /tags/server/1.2.1/LICENSE =================================================================== --- /tags/server/1.2.1/LICENSE (revision 257) +++ /tags/server/1.2.1/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/server/1.2.1/daemon.c =================================================================== --- /tags/server/1.2.1/daemon.c (revision 335) +++ /tags/server/1.2.1/daemon.c (revision 335) @@ -0,0 +1,70 @@ +/* $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(nochdir, noclose) + int nochdir, noclose; +{ + int fd; + + switch (fork()) { + case -1: + return (-1); + case 0: + break; + default: + _exit(0); + } + + if (setsid() == -1) + return (-1); + + if (!nochdir) + (void)chdir("/"); + + if (!noclose && (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/server/1.2.1/slabs.c =================================================================== --- /tags/server/1.2.1/slabs.c (revision 369) +++ /tags/server/1.2.1/slabs.c (revision 369) @@ -0,0 +1,356 @@ +/* -*- 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 +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "memcached.h" + +#define POWER_SMALLEST 1 +#define POWER_LARGEST 200 +#define POWER_BLOCK 1048576 +#define CHUNK_ALIGN_BYTES (sizeof(void *)) + +/* 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; + +/* + * Figures out which slab class (chunk size) is required to store an item of + * a given size. + */ +unsigned int slabs_clsid(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(size_t limit, double factor) { + 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; + 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 %6d perslab %5d\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 = atol(getenv("T_MEMD_INITIAL_MALLOC")); + } + + } + +#ifndef DONT_PREALLOC_SLABS + { + char *pre_alloc = getenv("T_MEMD_SLABS_ALLOC"); + if (!pre_alloc || atoi(pre_alloc)) { + slabs_preallocate(limit / POWER_BLOCK); + } + } +#endif +} + +void slabs_preallocate (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; + slabs_newslab(i); + } + +} + +static int grow_slab_list (unsigned int id) { + slabclass_t *p = &slabclass[id]; + if (p->slabs == p->list_size) { + size_t new_size = p->list_size ? 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; +} + +int slabs_newslab(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)) return 0; + + ptr = malloc(len); + if (ptr == 0) return 0; + + memset(ptr, 0, len); + p->end_page_ptr = ptr; + p->end_page_free = p->perslab; + + p->slab_list[p->slabs++] = ptr; + mem_malloced += len; + return 1; +} + +void *slabs_alloc(size_t size) { + slabclass_t *p; + + unsigned char id = slabs_clsid(size); + if (id < POWER_SMALLEST || id > power_largest) + return 0; + + 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 || p->sl_curr || slabs_newslab(id))) + return 0; + + /* return off our freelist, if we have one */ + if (p->sl_curr) + 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) { + p->end_page_ptr += p->size; + } else { + p->end_page_ptr = 0; + } + return ptr; + } + + return 0; /* shouldn't ever get here */ +} + +void slabs_free(void *ptr, size_t size) { + unsigned char id = slabs_clsid(size); + 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 ? 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; +} + +char* slabs_stats(int *buflen) { + int i, total; + char *buf = (char*) malloc(power_largest * 200 + 100); + char *bufcurr = buf; + + *buflen = 0; + if (!buf) return 0; + + total = 0; + for(i = POWER_SMALLEST; i <= power_largest; i++) { + slabclass_t *p = &slabclass[i]; + if (p->slabs) { + 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 slabs_reassign(unsigned char srcid, unsigned char dstid) { + void *slab, *slab_end; + slabclass_t *p, *dp; + void *iter; + int was_busy = 0; + + 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 = slab + POWER_BLOCK; + + for (iter=slab; itersize) { + item *it = (item *) iter; + if (it->slabs_clsid) { + if (it->refcount) was_busy = 1; + 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; itersize) { + ((item *)iter)->slabs_clsid = 0; + } + return 1; +} +#endif Index: /tags/server/1.2.1/AUTHORS =================================================================== --- /tags/server/1.2.1/AUTHORS (revision 257) +++ /tags/server/1.2.1/AUTHORS (revision 257) @@ -0,0 +1,2 @@ +Anatoly Vorobey +Brad Fitzpatrick Index: /tags/server/1.2.1/scripts/memcached-tool =================================================================== --- /tags/server/1.2.1/scripts/memcached-tool (revision 213) +++ /tags/server/1.2.1/scripts/memcached-tool (revision 213) @@ -0,0 +1,111 @@ +#!/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; +} 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 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; +} + +# 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 Full?\n"; +foreach my $n (6..17) { + my $it = $items{$n}; + my $size = $it->{chunk_size} < 1024 ? "$it->{chunk_size} B" : + sprintf("%d kB", $it->{chunk_size} / 1024); + my $full = $it->{free_chunks_end} == 0 ? "yes" : " no"; + printf "%3d %6s%7d s %7d $full\n", $n, $size, $it->{age}, $it->{total_pages}; +} + Index: /tags/server/1.2.1/scripts/start-memcached =================================================================== --- /tags/server/1.2.1/scripts/start-memcached (revision 201) +++ /tags/server/1.2.1/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/server/1.2.1/scripts/memcached-init =================================================================== --- /tags/server/1.2.1/scripts/memcached-init (revision 176) +++ /tags/server/1.2.1/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/server/1.2.1/memcached.c =================================================================== --- /tags/server/1.2.1/memcached.c (revision 450) +++ /tags/server/1.2.1/memcached.c (revision 450) @@ -0,0 +1,2501 @@ +/* -*- 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 + * + * $Id$ + */ +#include "config.h" +#include +#include +#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 +#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__) +# define IOV_MAX 1024 +#endif +#endif + +#include "memcached.h" + +struct stats stats; +struct settings settings; + +static item **todelete = 0; +static int delcurr; +static int deltotal; +static conn *listen_conn; + +#define TRANSMIT_COMPLETE 0 +#define TRANSMIT_INCOMPLETE 1 +#define TRANSMIT_SOFT_ERROR 2 +#define TRANSMIT_HARD_ERROR 3 + +int *buckets = 0; /* bucket->generation array for a managed instance */ + +#define REALTIME_MAXDELTA 60*60*24*30 +rel_time_t realtime(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) + return (rel_time_t) (exptime - stats.started); + else { + return (rel_time_t) (exptime + current_time); + } +} + +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 = 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; +} + +void stats_reset(void) { + stats.total_items = stats.total_conns = 0; + stats.get_cmds = stats.set_cmds = stats.get_hits = stats.get_misses = 0; + stats.bytes_read = stats.bytes_written = 0; +} + +void settings_init(void) { + settings.port = 11211; + settings.udpport = 0; + settings.interface.s_addr = htonl(INADDR_ANY); + 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 = 0; + settings.factor = 1.25; + settings.chunk_size = 48; /* space for a modest key and value */ +} + +/* returns true if a deleted item's delete-locked-time is over, and it + should be removed from the namespace */ +int 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 *get_item_notedeleted(char *key, size_t nkey, int *delete_locked) { + item *it = assoc_find(key, nkey); + if (delete_locked) *delete_locked = 0; + if (it && (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 = 1; + it = 0; + } + } + if (it && settings.oldest_live && settings.oldest_live <= current_time && + it->time <= settings.oldest_live) { + item_unlink(it); + it = 0; + } + if (it && it->exptime && it->exptime <= current_time) { + item_unlink(it); + it = 0; + } + return it; +} + +item *get_item(char *key, size_t nkey) { + return get_item_notedeleted(key, nkey, 0); +} + +/* + * Adds a message header to a connection. + * + * Returns 0 on success, -1 on out-of-memory. + */ +int add_msghdr(conn *c) +{ + struct msghdr *msg; + + 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]; + 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; +} + +conn **freeconns; +int freetotal; +int freecurr; + +void conn_init(void) { + freetotal = 200; + freecurr = 0; + freeconns = (conn **)malloc(sizeof (conn *)*freetotal); + return; +} + +conn *conn_new(int sfd, int init_state, int event_flags, int read_buffer_size, + int is_udp) { + conn *c; + + /* do we have a free conn structure from a previous close? */ + if (freecurr > 0) { + c = freeconns[--freecurr]; + } else { /* allocate a new one */ + if (!(c = (conn *)malloc(sizeof(conn)))) { + perror("malloc()"); + return 0; + } + c->rbuf = c->wbuf = 0; + c->ilist = 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->iovsize = IOV_LIST_INITIAL; + c->msgsize = MSG_LIST_INITIAL; + c->hdrsize = 0; + + c->rbuf = (char *) malloc(c->rsize); + c->wbuf = (char *) malloc(c->wsize); + c->ilist = (item **) malloc(sizeof(item *) * c->isize); + 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) { + if (c->rbuf != 0) free(c->rbuf); + if (c->wbuf != 0) free(c->wbuf); + if (c->ilist !=0) free(c->ilist); + if (c->iov != 0) free(c->iov); + if (c->msglist != 0) free(c->msglist); + free(c); + perror("malloc()"); + return 0; + } + + stats.conn_structs++; + } + + 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->ileft = 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; + + event_set(&c->event, sfd, event_flags, event_handler, (void *)c); + c->ev_flags = event_flags; + + if (event_add(&c->event, 0) == -1) { + if (freecurr < freetotal) { + freeconns[freecurr++] = c; + } else { + if (c->hdrbuf) + free (c->hdrbuf); + free (c->msglist); + free (c->rbuf); + free (c->wbuf); + free (c->ilist); + free (c->iov); + free (c); + } + return 0; + } + + stats.curr_conns++; + stats.total_conns++; + + return c; +} + +void conn_cleanup(conn *c) { + if (c->item) { + item_free(c->item); + c->item = 0; + } + + if (c->ileft) { + for (; c->ileft > 0; c->ileft--,c->icurr++) { + item_remove(*(c->icurr)); + } + } + + if (c->write_and_free) { + free(c->write_and_free); + c->write_and_free = 0; + } +} + +/* + * Frees a connection. + */ +static 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->iov) + free(c->iov); + free(c); + } +} + +void conn_close(conn *c) { + /* 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(1); + conn_cleanup(c); + + /* if the connection has big buffers, just free it */ + if (c->rsize > READ_BUFFER_HIGHWAT) { + conn_free(c); + } else if (freecurr < freetotal) { + /* if we have enough space in the free connections array, put the structure there */ + freeconns[freecurr++] = c; + } 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; + } else { + conn_free(c); + } + } + + stats.curr_conns--; + + return; +} + +/* + * Reallocates memory and updates a buffer size if successful. + */ +int do_realloc(void **orig, int newsize, int bytes_per_item, int *size) { + void *newbuf = realloc(*orig, newsize * bytes_per_item); + if (newbuf) { + *orig = newbuf; + *size = newsize; + return 1; + } + return 0; +} + +/* + * 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! + */ +void conn_shrink(conn *c) { + if (c->udp) + return; + + if (c->rsize > READ_BUFFER_HIGHWAT && c->rbytes < DATA_BUFFER_SIZE) { + if (c->rcurr != c->rbuf) + memmove(c->rbuf, c->rcurr, c->rbytes); + do_realloc((void **)&c->rbuf, DATA_BUFFER_SIZE, 1, &c->rsize); + c->rcurr = c->rbuf; + } + + if (c->isize > ITEM_LIST_HIGHWAT) { + do_realloc((void **)&c->ilist, ITEM_LIST_INITIAL, sizeof(c->ilist[0]), &c->isize); + } + + if (c->msgsize > MSG_LIST_HIGHWAT) { + do_realloc((void **)&c->msglist, MSG_LIST_INITIAL, sizeof(c->msglist[0]), &c->msgsize); + } + + if (c->iovsize > IOV_LIST_HIGHWAT) { + do_realloc((void **)&c->iov, IOV_LIST_INITIAL, sizeof(c->iov[0]), &c->iovsize); + } +} + +/* + * Sets a connection's current state in the state machine. Any special + * processing that needs to happen on certain state transitions can + * happen here. + */ +void conn_set_state(conn *c, int state) { + if (state != c->state) { + if (state == conn_read) { + conn_shrink(c); + assoc_move_next_bucket(); + } + c->state = state; + } +} + + +/* + * Ensures that there is room for another struct iovec in a connection's + * iov list. + * + * Returns 0 on success, -1 on out-of-memory. + */ +int ensure_iov_space(conn *c) { + 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. + */ + +int add_iov(conn *c, const void *buf, int len) { + struct msghdr *m; + int leftover; + int limit_to_mtu; + + 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)) + 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. + */ +int build_udp_headers(conn *c) { + int i; + unsigned char *hdr; + + 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; +} + + +void out_string(conn *c, char *str) { + int len; + + 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); + } + + strcpy(c->wbuf, str); + strcpy(c->wbuf + len, "\r\n"); + 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. + */ + +void complete_nread(conn *c) { + item *it = c->item; + int comm = c->item_comm; + item *old_it; + int delete_locked = 0; + char *key = ITEM_key(it); + + stats.set_cmds++; + + if (strncmp(ITEM_data(it) + it->nbytes - 2, "\r\n", 2) != 0) { + out_string(c, "CLIENT_ERROR bad data chunk"); + goto err; + } + + old_it = get_item_notedeleted(key, it->nkey, &delete_locked); + + if (old_it && comm == NREAD_ADD) { + item_update(old_it); /* touches item, promotes to head of LRU */ + out_string(c, "NOT_STORED"); + goto err; + } + + if (!old_it && comm == NREAD_REPLACE) { + out_string(c, "NOT_STORED"); + goto err; + } + + if (delete_locked) { + if (comm == NREAD_REPLACE || comm == NREAD_ADD) { + out_string(c, "NOT_STORED"); + goto err; + } + + /* but "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 + get_item.... because we need to replace it (below) */ + old_it = assoc_find(key, it->nkey); + } + + if (old_it) + item_replace(old_it, it); + else + item_link(it); + + c->item = 0; + out_string(c, "STORED"); + return; + +err: + item_free(it); + c->item = 0; + return; +} + +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 6 + +/* + * 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; + * } + */ +size_t tokenize_command(char* command, token_t* tokens, size_t max_tokens) { + char* cp; + char* value = NULL; + size_t length = 0; + size_t ntokens = 0; + + assert(command != NULL && tokens != NULL && max_tokens > 1); + + cp = command; + while(*cp != '\0' && ntokens < max_tokens - 1) { + if(*cp == ' ') { + // If we've accumulated a token, this is the end of it. + if(length > 0) { + tokens[ntokens].value = value; + tokens[ntokens].length = length; + ntokens++; + length = 0; + value = NULL; + } + *cp = '\0'; + } else { + if(length == 0) { + value = cp; + } + length++; + } + cp++; + } + + if(ntokens < max_tokens - 1 && length > 0) { + tokens[ntokens].value = value; + tokens[ntokens].length = length; + ntokens++; + } + + /* + * If we scanned the whole string, the terminal value pointer is null, + * otherwise it is the first unprocessed character. + */ + tokens[ntokens].value = *cp == '\0' ? NULL : cp; + tokens[ntokens].length = 0; + ntokens++; + + return ntokens; +} + +void process_stat(conn *c, token_t* tokens, size_t ntokens) { + rel_time_t now = current_time; + char* command; + char* subcommand; + + 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; + struct rusage usage; + + getrusage(RUSAGE_SELF, &usage); + + 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*)); + 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); + 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 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", (unsigned long long) settings.maxbytes); + pos += sprintf(pos, "END"); + 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 (strcmp(subcommand, "maps") == 0) { + char *wbuf; + int wsize = 8192; /* should be enough */ + int fd; + int res; + + wbuf = (char *)malloc(wsize); + if (wbuf == 0) { + out_string(c, "SERVER_ERROR out of memory"); + 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; + } + strcpy(wbuf + res, "END\r\n"); + c->write_and_free=wbuf; + c->wcurr=wbuf; + c->wbytes = res + 5; // Don't write the terminal '\0' + conn_set_state(c, conn_write); + c->write_and_go = conn_read; + close(fd); + return; + } + + 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); + if (buf == 0) { + out_string(c, "SERVER_ERROR out of memory"); + return; + } + + c->write_and_free = buf; + c->wcurr = buf; + c->wbytes = bytes; + conn_set_state(c, conn_write); + c->write_and_go = conn_read; + return; + } + + if (strcmp(subcommand, "slabs")==0) { + int bytes = 0; + char *buf = slabs_stats(&bytes); + if (!buf) { + out_string(c, "SERVER_ERROR out of memory"); + return; + } + c->write_and_free = buf; + c->wcurr = buf; + c->wbytes = bytes; + conn_set_state(c, conn_write); + c->write_and_go = conn_read; + return; + } + + if (strcmp(subcommand, "items")==0) { + char buffer[4096]; + item_stats(buffer, 4096); + out_string(c, buffer); + return; + } + + if (strcmp(subcommand, "sizes")==0) { + int bytes = 0; + char *buf = item_stats_sizes(&bytes); + if (! buf) { + out_string(c, "SERVER_ERROR out of memory"); + return; + } + + c->write_and_free = buf; + c->wcurr = buf; + c->wbytes = bytes; + conn_set_state(c, conn_write); + c->write_and_go = conn_read; + return; + } + + out_string(c, "ERROR"); +} + +inline void process_get_command(conn *c, token_t* tokens, size_t ntokens) { + char *key; + size_t nkey; + int i = 0; + item *it; + token_t* key_token = &tokens[KEY_TOKEN]; + + 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) { + out_string(c, "CLIENT_ERROR bad command line format"); + return; + } + + stats.get_cmds++; + it = get_item(key, nkey); + 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 (add_iov(c, "VALUE ", 6) || + add_iov(c, ITEM_key(it), it->nkey) || + add_iov(c, ITEM_suffix(it), it->nsuffix + it->nbytes)) + { + break; + } + if (settings.verbose > 1) + fprintf(stderr, ">%d sending key %s\n", c->sfd, ITEM_key(it)); + + stats.get_hits++; + it->refcount++; + 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 (settings.verbose > 1) + fprintf(stderr, ">%d END\n", c->sfd); + add_iov(c, "END\r\n", 5); + + if (c->udp && build_udp_headers(c)) { + out_string(c, "SERVER_ERROR out of memory"); + } + else { + conn_set_state(c, conn_mwrite); + c->msgcurr = 0; + } + return; +} + +void process_update_command(conn *c, token_t* tokens, size_t ntokens, int comm) { + char *key; + size_t nkey; + int flags; + time_t exptime; + int vlen; + item *it; + + 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); + + if(errno == ERANGE || ((flags == 0 || exptime == 0) && errno == EINVAL)) { + out_string(c, "CLIENT_ERROR bad command line format"); + return; + } + + 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(key, nkey, flags, vlen + 2)) + out_string(c, "SERVER_ERROR object too large for cache"); + else + out_string(c, "SERVER_ERROR out of memory"); + /* swallow the data line */ + c->write_and_go = conn_swallow; + c->sbytes = vlen+2; + return; + } + + c->item_comm = comm; + c->item = it; + c->ritem = ITEM_data(it); + c->rlbytes = it->nbytes; + conn_set_state(c, conn_nread); + return; +} + +void process_arithmetic_command(conn *c, token_t* tokens, size_t ntokens, int incr) { + char temp[32]; + unsigned int value; + item *it; + unsigned int delta; + char *key; + size_t nkey; + int res; + char *ptr; + + 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; + } + } + + it = get_item(key, nkey); + if (!it) { + out_string(c, "NOT_FOUND"); + return; + } + + delta = strtoul(tokens[2].value, NULL, 10); + + if(errno == ERANGE) { + out_string(c, "CLIENT_ERROR bad command line format"); + return; + } + + ptr = ITEM_data(it); + while (*ptr && (*ptr<'0' && *ptr>'9')) ptr++; // BUG: can't be true + + value = strtol(ptr, NULL, 10); + + if(errno == ERANGE) { + out_string(c, "CLIENT_ERROR cannot increment or decrement non-numeric value"); + return; + } + + if (incr) + value+=delta; + else { + if (delta >= value) value = 0; + else value-=delta; + } + sprintf(temp, "%u", value); + res = strlen(temp); + if (res + 2 > it->nbytes) { /* need to realloc */ + item *new_it; + new_it = item_alloc(ITEM_key(it), it->nkey, atoi(ITEM_suffix(it) + 1), it->exptime, res + 2 ); + if (new_it == 0) { + out_string(c, "SERVER_ERROR out of memory"); + return; + } + memcpy(ITEM_data(new_it), temp, res); + memcpy(ITEM_data(new_it) + res, "\r\n", 2); + item_replace(it, new_it); + } else { /* replace in-place */ + memcpy(ITEM_data(it), temp, res); + memset(ITEM_data(it) + res, ' ', it->nbytes-res-2); + } + out_string(c, temp); + return; +} + +void process_delete_command(conn *c, token_t* tokens, size_t ntokens) { + char *key; + size_t nkey; + item *it; + time_t exptime = 0; + + 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 == 4) { + exptime = strtol(tokens[2].value, NULL, 10); + + if(errno == ERANGE) { + out_string(c, "CLIENT_ERROR bad command line format"); + return; + } + } + + it = get_item(key, nkey); + if (!it) { + out_string(c, "NOT_FOUND"); + return; + } + + if (exptime == 0) { + item_unlink(it); + out_string(c, "DELETED"); + return; + } + 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 + */ + out_string(c, "SERVER_ERROR out of memory"); + return; + } + } + + it->refcount++; + /* use its expiration time as its deletion time now */ + it->exptime = realtime(exptime); + it->it_flags |= ITEM_DELETED; + todelete[delcurr++] = it; + out_string(c, "DELETED"); + return; +} + +void process_command(conn *c, char *command) { + + token_t tokens[MAX_TOKENS]; + size_t ntokens; + int comm; + + 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)) { + out_string(c, "SERVER_ERROR out of memory"); + 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); + + } else if (ntokens == 6 && + ((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)))) { + + process_update_command(c, tokens, ntokens, comm); + + } else if (ntokens == 4 && (strcmp(tokens[COMMAND_TOKEN].value, "incr") == 0)) { + + process_arithmetic_command(c, tokens, ntokens, 1); + + } else if (ntokens == 4 && (strcmp(tokens[COMMAND_TOKEN].value, "decr") == 0)) { + + process_arithmetic_command(c, tokens, ntokens, 0); + + } else if (ntokens >= 3 && ntokens <= 4 && (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 <= 3 && (strcmp(tokens[COMMAND_TOKEN].value, "flush_all") == 0)) { + time_t exptime = 0; + set_current_time(); + + if(ntokens == 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; + } + + settings.oldest_live = realtime(exptime) - 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 { + out_string(c, "ERROR"); + } + return; +} + +/* + * if we have a complete line in the buffer, process it. + */ +int try_read_command(conn *c) { + char *el, *cont; + + assert(c->rcurr <= c->rbuf + c->rsize); + + if (!c->rbytes) + 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. + */ +int try_read_udp(conn *c) { + int res; + + 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.bytes_read += res; + + /* 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. + */ +int try_read_network(conn *c) { + int gotdata = 0; + int res; + + 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"); + 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; + } + + res = read(c->sfd, c->rbuf + c->rbytes, c->rsize - c->rbytes); + if (res > 0) { + stats.bytes_read += res; + gotdata = 1; + c->rbytes += res; + continue; + } + if (res == 0) { + /* connection closed */ + conn_set_state(c, conn_closing); + return 1; + } + if (res == -1) { + if (errno == EAGAIN || errno == EWOULDBLOCK) break; + else return 0; + } + } + return gotdata; +} + +int update_event(conn *c, int new_flags) { + if (c->ev_flags == new_flags) + return 1; + if (event_del(&c->event) == -1) return 0; + event_set(&c->event, c->sfd, new_flags, event_handler, (void *)c); + c->ev_flags = new_flags; + if (event_add(&c->event, 0) == -1) return 0; + return 1; +} + +/* + * Sets whether we are listening for new connections or not. + */ +void accept_new_conns(int do_accept) { + if (do_accept) { + update_event(listen_conn, EV_READ | EV_PERSIST); + if (listen(listen_conn->sfd, 1024)) { + perror("listen"); + } + } + else { + update_event(listen_conn, 0); + if (listen(listen_conn->sfd, 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) + */ +int transmit(conn *c) { + int res; + + 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) { + struct msghdr *m = &c->msglist[c->msgcurr]; + res = sendmsg(c->sfd, m, 0); + if (res > 0) { + stats.bytes_written += res; + + /* 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; + } +} + +void drive_machine(conn *c) { + + int stop = 0; + int sfd, flags = 1; + socklen_t addrlen; + struct sockaddr addr; + conn *newc; + int res; + + while (!stop) { + switch(c->state) { + case conn_listening: + addrlen = sizeof(addr); + if ((sfd = accept(c->sfd, &addr, &addrlen)) == -1) { + if (errno == EAGAIN || errno == EWOULDBLOCK) { + stop = 1; + break; + } else if (errno == EMFILE) { + if (settings.verbose > 0) + fprintf(stderr, "Too many open connections\n"); + accept_new_conns(0); + } else { + perror("accept()"); + } + break; + } + if ((flags = fcntl(sfd, F_GETFL, 0)) < 0 || + fcntl(sfd, F_SETFL, flags | O_NONBLOCK) < 0) { + perror("setting O_NONBLOCK"); + close(sfd); + break; + } + newc = conn_new(sfd, conn_read, EV_READ | EV_PERSIST, + DATA_BUFFER_SIZE, 0); + if (!newc) { + if (settings.verbose > 0) + fprintf(stderr, "couldn't create new connection\n"); + close(sfd); + break; + } + + break; + + case conn_read: + if (try_read_command(c)) { + continue; + } + if (c->udp ? try_read_udp(c) : try_read_network(c)) { + 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 = 1; + 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.bytes_read += res; + 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 = 1; + 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.bytes_read += res; + 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 = 1; + 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) || + (c->udp && build_udp_headers(c))) { + 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--; + } + 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 = 1; + break; + } + break; + + case conn_closing: + if (c->udp) + conn_cleanup(c); + else + conn_close(c); + stop = 1; + break; + } + + } + + return; +} + +void event_handler(int fd, short which, void *arg) { + conn *c; + + c = (conn *)arg; + 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; + } + + /* do as much I/O as possible until we block */ + drive_machine(c); + + /* wait for next event */ + return; +} + +int new_socket(int is_udp) { + int sfd; + int flags; + + if ((sfd = socket(AF_INET, is_udp ? SOCK_DGRAM : 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; +} + + +/* + * Sets a socket's send buffer size to the maximum allowed by the system. + */ +void maximize_sndbuf(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)) { + 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, &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); +} + + +int server_socket(int port, int is_udp) { + int sfd; + struct linger ling = {0, 0}; + struct sockaddr_in addr; + int flags =1; + + if ((sfd = new_socket(is_udp)) == -1) { + return -1; + } + + setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, &flags, sizeof(flags)); + if (is_udp) { + maximize_sndbuf(sfd); + } else { + setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, &flags, sizeof(flags)); + setsockopt(sfd, SOL_SOCKET, SO_LINGER, &ling, sizeof(ling)); + setsockopt(sfd, IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(flags)); + } + + /* + * the memset call clears nonstandard fields in some impementations + * that otherwise mess things up. + */ + memset(&addr, 0, sizeof(addr)); + + addr.sin_family = AF_INET; + addr.sin_port = htons(port); + addr.sin_addr = settings.interface; + if (bind(sfd, (struct sockaddr *) &addr, sizeof(addr)) == -1) { + perror("bind()"); + close(sfd); + return -1; + } + if (! is_udp && listen(sfd, 1024) == -1) { + perror("listen()"); + close(sfd); + return -1; + } + return sfd; +} + +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; +} + +int server_socket_unix(char *path) { + int sfd; + struct linger ling = {0, 0}; + struct sockaddr_un addr; + struct stat tstat; + int flags =1; + + 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)) { + if (S_ISSOCK(tstat.st_mode)) + unlink(path); + } + + setsockopt(sfd, SOL_SOCKET, SO_REUSEADDR, &flags, sizeof(flags)); + setsockopt(sfd, SOL_SOCKET, SO_KEEPALIVE, &flags, sizeof(flags)); + setsockopt(sfd, SOL_SOCKET, SO_LINGER, &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); + if (bind(sfd, (struct sockaddr *) &addr, sizeof(addr)) == -1) { + perror("bind()"); + close(sfd); + return -1; + } + if (listen(sfd, 1024) == -1) { + perror("listen()"); + close(sfd); + return -1; + } + return sfd; +} + + +/* invoke right before gdb is called, on assert */ +void pre_gdb () { + int i = 0; + if(l_socket) close(l_socket); + if(u_socket > -1) close(u_socket); + for (i=3; i<=500; i++) close(i); /* so lame */ + kill(getpid(), SIGABRT); +} + +/* + * 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; +struct event clockevent; + +/* time-sensitive callers can call it by hand with this, outside the normal ever-1-second timer */ +void set_current_time () { + current_time = (rel_time_t) (time(0) - stats.started); +} + +void clock_handler(int fd, short which, void *arg) { + struct timeval t; + static int initialized = 0; + + if (initialized) { + /* only delete the event if it's actually there. */ + evtimer_del(&clockevent); + } else { + initialized = 1; + } + + evtimer_set(&clockevent, clock_handler, 0); + t.tv_sec = 1; + t.tv_usec = 0; + evtimer_add(&clockevent, &t); + + set_current_time(); +} + +struct event deleteevent; + +void delete_handler(int fd, short which, void *arg) { + struct timeval t; + static int initialized = 0; + + 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 = 1; + } + + evtimer_set(&deleteevent, delete_handler, 0); + t.tv_sec = 5; t.tv_usec=0; + evtimer_add(&deleteevent, &t); + + { + int i, j=0; + for (i=0; irefcount > 0); + it->it_flags &= ~ITEM_DELETED; + item_unlink(it); + item_remove(it); + } else { + todelete[j++] = it; + } + } + delcurr = j; + } +} + +void usage(void) { + printf(PACKAGE " " VERSION "\n"); + printf("-p TCP port number to listen on (default: 11211)\n"); + printf("-U UDP port number to listen on (default: 0, off)\n"); + printf("-s unix socket path to listen on (disables network support)\n"); + printf("-l interface to listen on, default is INDRR_ANY\n"); + printf("-d run as a daemon\n"); + printf("-r maximize core file limit\n"); + printf("-u assume identity of (only when run as root)\n"); + printf("-m max memory to use for items in megabytes, default is 64 MB\n"); + printf("-M return error on memory exhausted (rather than removing items)\n"); + printf("-c max simultaneous connections, default is 1024\n"); + printf("-k lock down all paged memory\n"); + printf("-v verbose (print errors/warnings while in event loop)\n"); + printf("-vv very verbose (also print client commands/reponses)\n"); + printf("-h print this help and exit\n"); + printf("-i print memcached and libevent license\n"); + printf("-b run a managed instanced (mnemonic: buckets)\n"); + printf("-P save PID in , only used with -d option\n"); + printf("-f chunk size growth factor, default 1.25\n"); + printf("-n minimum space allocated for key+value+flags, default 48\n"); + return; +} + +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; +} + +void save_pid(pid_t pid,char *pid_file) { + FILE *fp; + if (!pid_file) + return; + + if (!(fp = fopen(pid_file,"w"))) { + 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; + } +} + +void remove_pidfile(char *pid_file) { + if (!pid_file) + return; + + if (unlink(pid_file)) { + fprintf(stderr,"Could not remove the pid file %s.\n",pid_file); + } + +} + +int l_socket=0; +int u_socket=-1; + +void sig_handler(int sig) { + printf("SIGINT handled.\n"); + exit(0); +} + +int main (int argc, char **argv) { + int c; + conn *u_conn; + struct in_addr addr; + int lock_memory = 0; + int daemonize = 0; + int maxcore = 0; + char *username = 0; + struct passwd *pw; + struct sigaction sa; + struct rlimit rlim; + char *pid_file = NULL; + + /* 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, "bp:s:U:m:Mc:khirvdl:u:P:f:s:")) != -1) { + switch (c) { + case 'U': + settings.udpport = atoi(optarg); + break; + case 'b': + settings.managed = 1; + 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(0); + case 'i': + usage_license(); + exit(0); + case 'k': + lock_memory = 1; + break; + case 'v': + settings.verbose++; + break; + case 'l': + if (!inet_pton(AF_INET, optarg, &addr)) { + fprintf(stderr, "Illegal address: %s\n", optarg); + return 1; + } else { + settings.interface = addr; + } + break; + case 'd': + daemonize = 1; + 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; + default: + fprintf(stderr, "Illegal argument \"%c\"\n", c); + return 1; + } + } + + if (maxcore) { + 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(1); + } + } + + /* + * 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(1); + } 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(1); + } + } + + /* + * 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). + */ + + /* create the listening socket and bind it */ + if (!settings.socketpath) { + l_socket = server_socket(settings.port, 0); + if (l_socket == -1) { + fprintf(stderr, "failed to listen\n"); + exit(1); + } + } + + if (settings.udpport > 0 && ! settings.socketpath) { + /* create the UDP listening socket and bind it */ + u_socket = server_socket(settings.udpport, 1); + if (u_socket == -1) { + fprintf(stderr, "failed to listen on UDP port %d\n", settings.udpport); + exit(1); + } + } + + /* 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; + } + } + + /* create unix mode sockets after dropping privileges */ + if (settings.socketpath) { + l_socket = server_socket_unix(settings.socketpath); + if (l_socket == -1) { + fprintf(stderr, "failed to listen\n"); + exit(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 other stuff */ + item_init(); + event_init(); + stats_init(); + assoc_init(); + conn_init(); + slabs_init(settings.maxbytes, settings.factor); + + /* 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(1); + } + memset(buckets, 0, sizeof(int)*MAX_BUCKETS); + } + + /* lock paged memory if needed */ + if (lock_memory) { +#ifdef HAVE_MLOCKALL + mlockall(MCL_CURRENT | MCL_FUTURE); +#else + fprintf(stderr, "warning: mlockall() not supported on this platform. proceeding without.\n"); +#endif + } + + /* + * 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(1); + } + /* create the initial listening connection */ + if (!(listen_conn = conn_new(l_socket, conn_listening, EV_READ | EV_PERSIST, 1, 0))) { + fprintf(stderr, "failed to create listening connection"); + exit(1); + } + /* create the initial listening udp connection */ + if (u_socket > -1 && + !(u_conn = conn_new(u_socket, conn_read, EV_READ | EV_PERSIST, UDP_READ_BUFFER_SIZE, 1))) { + fprintf(stderr, "failed to create udp connection"); + exit(1); + } + /* initialise clock event */ + clock_handler(0,0,0); + /* initialise deletion array and timer event */ + deltotal = 200; delcurr = 0; + todelete = malloc(sizeof(item *)*deltotal); + delete_handler(0,0,0); /* sets up the event */ + /* save the PID in if we're a daemon */ + if (daemonize) + save_pid(getpid(),pid_file); + /* enter the loop */ + event_loop(0); + /* remove the PID file if we're a daemon */ + if (daemonize) + remove_pidfile(pid_file); + return 0; +} Index: /tags/server/1.2.1/ChangeLog =================================================================== --- /tags/server/1.2.1/ChangeLog (revision 450) +++ /tags/server/1.2.1/ChangeLog (revision 450) @@ -0,0 +1,267 @@ +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-13 + * Iain Wade : Fix for UDP responses on non-"get" + commands. + +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/server/1.2.1/assoc.c =================================================================== --- /tags/server/1.2.1/assoc.c (revision 450) +++ /tags/server/1.2.1/assoc.c (revision 450) @@ -0,0 +1,619 @@ +/* -*- 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 "config.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "memcached.h" + +/* + * 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 */ + 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; /* fall through */ + case 10: c+=k[4]; + b+=k[2]+(((uint32_t)k[3])<<16); + a+=k[0]+(((uint32_t)k[1])<<16); + break; + case 9 : c+=k8[8]; /* fall through */ + 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; /* fall through */ + case 6 : b+=k[2]; + a+=k[0]+(((uint32_t)k[1])<<16); + break; + case 5 : b+=k8[4]; /* fall through */ + case 4 : a+=k[0]+(((uint32_t)k[1])<<16); + break; + case 3 : a+=((uint32_t)k8[2])<<16; /* fall through */ + 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, 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 */ +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 int hash_items = 0; + +/* Flag: Are we in the middle of expanding now? */ +static int expanding = 0; + +/* + * 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 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(1); + } + memset(primary_hashtable, 0, hash_size); +} + +item *assoc_find(const char *key, size_t nkey) { + uint32_t hv = hash(key, nkey, 0); + item *it; + 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, size_t nkey) { + uint32_t hv = hash(key, nkey, 0); + item **pos; + 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 = 1; + expand_bucket = 0; + 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 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; + } + + expand_bucket++; + if (expand_bucket == hashsize(hashpower - 1)) { + expanding = 0; + 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; + 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, 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/server/1.2.1/memcached.h =================================================================== --- /tags/server/1.2.1/memcached.h (revision 450) +++ /tags/server/1.2.1/memcached.h (revision 450) @@ -0,0 +1,284 @@ +/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */ +/* $Id$ */ +#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) + +/* Initial size of list of items being returned by "get". */ +#define ITEM_LIST_INITIAL 200 + +/* 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 + +/* 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; + unsigned long long curr_bytes; + unsigned int curr_conns; + unsigned int total_conns; + unsigned int conn_structs; + unsigned long long get_cmds; + unsigned long long set_cmds; + unsigned long long get_hits; + unsigned long long get_misses; + time_t started; /* when the process was started */ + unsigned long long bytes_read; + unsigned long long bytes_written; +}; + +struct settings { + size_t maxbytes; + int maxconns; + int port; + int udpport; + struct in_addr interface; + int verbose; + rel_time_t oldest_live; /* ignore existing items older than this */ + int managed; /* if 1, a tracker manages virtual buckets */ + int evict_to_free; + char *socketpath; /* path to unix socket if using local socket */ + double factor; /* chunk size growth factor */ + int chunk_size; +}; + +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; + unsigned char nsuffix; /* length of flags-and-length string */ + unsigned char it_flags; /* ITEM_* above */ + unsigned char slabs_clsid;/* which slab class we're in */ + unsigned char nkey; /* key length, w/terminating null and padding */ + void * end[0]; + /* 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 + +typedef struct { + 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; + + /* data for UDP clients */ + int udp; /* 1 if 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 */ +} conn; + +/* number of virtual buckets for a managed instance */ +#define MAX_BUCKETS 32768 + +/* listening socket */ +extern int l_socket; + +/* udp socket */ +extern int u_socket; + +/* current time of day (updated periodically) */ +extern volatile rel_time_t current_time; + +/* temporary hack */ +/* #define assert(x) if(!(x)) { printf("assert failure: %s\n", #x); pre_gdb(); } + void pre_gdb (); */ + +/* + * Functions + */ + +/* + * 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). + */ + +rel_time_t realtime(time_t exptime); + +/* 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. */ +void slabs_init(size_t limit, double factor); + +/* 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. */ +void slabs_preallocate (unsigned int maxslabs); + +/* 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(size_t size); + +/* Allocate object of given length. 0 on error */ +void *slabs_alloc(size_t size); + +/* Free previously allocated object */ +void slabs_free(void *ptr, size_t size); + +/* Fill buffer with stats */ +char* slabs_stats(int *buflen); + +/* Request some slab be moved between classes + 1 = success + 0 = fail + -1 = tried. busy. send again shortly. */ +int slabs_reassign(unsigned char srcid, unsigned char dstid); +int slabs_newslab(unsigned int id); + +/* event handling, network IO */ +void event_handler(int fd, short which, void *arg); +conn *conn_new(int sfd, int init_state, int event_flags, int read_buffer_size, int is_udp); +void conn_close(conn *c); +void conn_init(void); +void accept_new_conns(int do_accept); +void drive_machine(conn *c); +int new_socket(int isUdp); +int server_socket(int port, int isUdp); +int update_event(conn *c, int new_flags); +int try_read_command(conn *c); +int try_read_network(conn *c); +int try_read_udp(conn *c); +void complete_nread(conn *c); +void process_command(conn *c, char *command); +int transmit(conn *c); +int ensure_iov_space(conn *c); +int add_iov(conn *c, const void *buf, int len); +int add_msghdr(conn *c); +/* stats */ +void stats_reset(void); +void stats_init(void); +/* defaults */ +void settings_init(void); +/* associative array */ +void assoc_init(void); +item *assoc_find(const char *key, size_t nkey); +int assoc_insert(item *item); +void assoc_delete(const char *key, size_t nkey); +void assoc_move_next_bucket(void); + +void item_init(void); +item *item_alloc(char *key, size_t nkey, int flags, rel_time_t exptime, int nbytes); +void item_free(item *it); +int item_size_ok(char *key, size_t nkey, int flags, int nbytes); + +int item_link(item *it); /* may fail if transgresses limits */ +void item_unlink(item *it); +void item_remove(item *it); +void item_update(item *it); /* update LRU time to current and reposition */ +int item_replace(item *it, item *new_it); +char *item_cachedump(unsigned int slabs_clsid, unsigned int limit, unsigned int *bytes); +char *item_stats_sizes(int *bytes); +void item_stats(char *buffer, int buflen); +void item_flush_expired(void); + +/* time handling */ +void set_current_time (); /* update the global variable holding + global 32-bit seconds-since-start time + (to avoid 64 bit time_t) */ Index: /tags/server/1.2.1/CONTRIBUTORS =================================================================== --- /tags/server/1.2.1/CONTRIBUTORS (revision 257) +++ /tags/server/1.2.1/CONTRIBUTORS (revision 257) @@ -0,0 +1,48 @@ +Brad Fitzpatrick + -- design/protocol + -- Perl client + -- prototype Perl server + -- memory allocator design + -- small enhancements/changes to C server + -- website + +Anatoly Vorobey + -- C server + -- memory allocator design + -- revised setuid code + +Evan Martin + -- automake/autoconf support + -- Python client + -- portability work to build on OS X + +Ryan + -- PHP client + +Jamie McCarthy + -- Perl client fixes: Makefile.PL, stats, doc updates + +Lisa Marie Seelye + -- packaging for Gentoo Linux + -- initial setuid code + +Sean Chittenden + -- packaging for FreeBSD + +Stuart Herbert + -- fix for: memcached's php client can run in an infinite loop + http://bugs.gentoo.org/show_bug.cgi?id=25385 + +Brion Vibber + -- debugging abstraction in PHP client + -- debugging the failure of daemon mode on FreeBSD + +Brad Whitaker + -- compression support for the Perl API + +Richard Russo + -- Java API + +Ryan T. Dean + -- Second PHP client with correct parsing (based on Perl client) + -- autoconf fixes for mallinfo.arena on BSD (don't just check malloc.h) Index: /tags/server/1.2.1/README =================================================================== --- /tags/server/1.2.1/README (revision 257) +++ /tags/server/1.2.1/README (revision 257) @@ -0,0 +1,22 @@ +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). + +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/server/1.2.1/BUILD =================================================================== --- /tags/server/1.2.1/BUILD (revision 257) +++ /tags/server/1.2.1/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/server/1.2.1/devtools/svn-tarballs.pl =================================================================== --- /tags/server/1.2.1/devtools/svn-tarballs.pl (revision 323) +++ /tags/server/1.2.1/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/server/1.2.1/devtools/clean-whitespace.pl =================================================================== --- /tags/server/1.2.1/devtools/clean-whitespace.pl (revision 332) +++ /tags/server/1.2.1/devtools/clean-whitespace.pl (revision 332) @@ -0,0 +1,18 @@ +#!/usr/bin/perl +use strict; +use FindBin qw($Bin); +chdir "$Bin/.." or die; +my @files = (glob("*.h"), glob("*.c")); +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/server/1.2.1/items.c =================================================================== --- /tags/server/1.2.1/items.c (revision 450) +++ /tags/server/1.2.1/items.c (revision 450) @@ -0,0 +1,357 @@ +/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */ +/* $Id$ */ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "memcached.h" + +/* + * 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 +static item *heads[LARGEST_ID]; +static item *tails[LARGEST_ID]; +unsigned int sizes[LARGEST_ID]; + +void item_init(void) { + int i; + for(i=0; i0 + * search up from tail an item with refcount==0 and unlink it; give up after 50 + * tries + */ + + if (id > LARGEST_ID) return 0; + if (tails[id]==0) return 0; + + for (search = tails[id]; tries>0 && search; tries--, search=search->prev) { + if (search->refcount==0) { + item_unlink(search); + break; + } + } + it = slabs_alloc(ntotal); + if (it==0) return 0; + } + + 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 = 0; + it->it_flags = 0; + it->nkey = nkey; + it->nbytes = nbytes; + strcpy(ITEM_key(it), key); + it->exptime = exptime; + memcpy(ITEM_suffix(it), suffix, nsuffix); + it->nsuffix = nsuffix; + return it; +} + +void item_free(item *it) { + unsigned int ntotal = ITEM_ntotal(it); + 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 */ + it->slabs_clsid = 0; + it->it_flags |= ITEM_SLABBED; + slabs_free(it, ntotal); +} + +/* + * Returns true if an item will fit in the cache (its size does not exceed + * the maximum for a cache entry.) + */ +int item_size_ok(char *key, size_t nkey, int flags, int nbytes) { + char prefix[40]; + int nsuffix; + + return slabs_clsid(item_make_header(key, nkey + 1, flags, nbytes, + prefix, &nsuffix)) != 0; +} + +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; +} + +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 item_link(item *it) { + assert((it->it_flags & (ITEM_LINKED|ITEM_SLABBED)) == 0); + assert(it->nbytes < 1048576); + it->it_flags |= ITEM_LINKED; + it->time = current_time; + assoc_insert(it); + + stats.curr_bytes += ITEM_ntotal(it); + stats.curr_items += 1; + stats.total_items += 1; + + item_link_q(it); + + return 1; +} + +void item_unlink(item *it) { + if (it->it_flags & ITEM_LINKED) { + it->it_flags &= ~ITEM_LINKED; + stats.curr_bytes -= ITEM_ntotal(it); + stats.curr_items -= 1; + assoc_delete(ITEM_key(it), it->nkey); + item_unlink_q(it); + } + if (it->refcount == 0) item_free(it); +} + +void item_remove(item *it) { + assert((it->it_flags & ITEM_SLABBED) == 0); + if (it->refcount) it->refcount--; + assert((it->it_flags & ITEM_DELETED) == 0 || it->refcount); + if (it->refcount == 0 && (it->it_flags & ITEM_LINKED) == 0) { + item_free(it); + } +} + +void item_update(item *it) { + if (it->time < current_time - ITEM_UPDATE_INTERVAL) { + assert((it->it_flags & ITEM_SLABBED) == 0); + + item_unlink_q(it); + it->time = current_time; + item_link_q(it); + } +} + +int item_replace(item *it, item *new_it) { + assert((it->it_flags & ITEM_SLABBED) == 0); + + item_unlink(it); + return item_link(new_it); +} + +char *item_cachedump(unsigned int slabs_clsid, unsigned int limit, unsigned int *bytes) { + + int memlimit = 2*1024*1024; + char *buffer; + int bufcurr; + item *it; + int len; + int shown = 0; + char temp[512]; + + if (slabs_clsid > LARGEST_ID) return 0; + it = heads[slabs_clsid]; + + buffer = malloc(memlimit); + if (buffer == 0) return 0; + bufcurr = 0; + + while (it && (!limit || shown < limit)) { + len = sprintf(temp, "ITEM %s [%u b; %lu s]\r\n", ITEM_key(it), it->nbytes - 2, it->time + stats.started); + if (bufcurr + len + 6 > memlimit) /* 6 is END\r\n\0 */ + break; + strcpy(buffer + bufcurr, temp); + bufcurr+=len; + shown++; + it = it->next; + } + + strcpy(buffer+bufcurr, "END\r\n"); + bufcurr+=5; + + *bytes = bufcurr; + return buffer; +} + +void item_stats(char *buffer, int buflen) { + int i; + char *bufcurr = buffer; + rel_time_t now = current_time; + + if (buflen < 4096) { + strcpy(buffer, "SERVER_ERROR out of memory"); + return; + } + + for (i=0; itime); + } + strcpy(bufcurr, "END"); + return; +} + +/* dumps out a list of objects of each size, with granularity of 32 bytes */ +char* item_stats_sizes(int *bytes) { + int num_buckets = 32768; /* max 1MB object, divided into 32 bytes size buckets */ + unsigned int *histogram = (unsigned int*) malloc(num_buckets * sizeof(int)); + char *buf = (char*) malloc(1024*1024*2*sizeof(char)); + int i; + + if (histogram == 0 || buf == 0) { + if (histogram) free(histogram); + if (buf) free(buf); + return 0; + } + + /* build the histogram */ + memset(histogram, 0, num_buckets * sizeof(int)); + for (i=0; inext; + } + } + + /* write the buffer */ + *bytes = 0; + for (i=0; itime >= settings.oldest_live) { + next = iter->next; + if ((iter->it_flags & ITEM_SLABBED) == 0) { + item_unlink(iter); + } + } else { + /* We've hit the first old item. Continue to the next queue. */ + break; + } + } + } +} Index: /tags/server/1.2.1/t/bogus-commands.t =================================================================== --- /tags/server/1.2.1/t/bogus-commands.t (revision 348) +++ /tags/server/1.2.1/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/server/1.2.1/t/00-startup.t =================================================================== --- /tags/server/1.2.1/t/00-startup.t (revision 325) +++ /tags/server/1.2.1/t/00-startup.t (revision 325) @@ -0,0 +1,11 @@ +#!/usr/bin/perl + +use strict; +use Test::More tests => 1; +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); + +ok($server, "started the server"); Index: /tags/server/1.2.1/t/flush-all.t =================================================================== --- /tags/server/1.2.1/t/flush-all.t (revision 436) +++ /tags/server/1.2.1/t/flush-all.t (revision 436) @@ -0,0 +1,35 @@ +#!/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; +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); + +# 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/server/1.2.1/t/udp.t =================================================================== --- /tags/server/1.2.1/t/udp.t (revision 378) +++ /tags/server/1.2.1/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/server/1.2.1/t/64bit.t =================================================================== --- /tags/server/1.2.1/t/64bit.t (revision 392) +++ /tags/server/1.2.1/t/64bit.t (revision 392) @@ -0,0 +1,44 @@ +#!/usr/bin/perl + +use strict; +use Test::More; +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +$ENV{T_MEMD_INITIAL_MALLOC} = "4294967328"; # 2**32 + 32 , just over 4GB +$ENV{T_MEMD_SLABS_ALLOC} = 0; # don't preallocate slabs + +my $server = new_memcached("-m 4098 -M"); +my $sock = $server->sock; + +my ($stats, $slabs) = @_; + +$stats = mem_stats($sock); + +if ($stats->{'pointer_size'} eq "32") { + plan skip_all => 'Skipping 64-bit tests on 32-bit build'; + exit 0; +} else { + plan tests => 6; +} + +is($stats->{'pointer_size'}, 64, "is 64 bit"); +is($stats->{'limit_maxbytes'}, "4297064448", "max bytes is 4098 MB"); + +$slabs = mem_stats($sock, 'slabs'); +is($slabs->{'total_malloced'}, "4294967328", "expected (faked) value of total_malloced"); +is($slabs->{'active_slabs'}, 0, "no active slabs"); + +my $hit_limit = 0; +for (1..5) { + my $size = 400 * 1024; + my $data = "a" x $size; + print $sock "set big$_ 0 0 $size\r\n$data\r\n"; + my $res = <$sock>; + $hit_limit = 1 if $res ne "STORED\r\n"; +} +ok($hit_limit, "hit size limit"); + +$slabs = mem_stats($sock, 'slabs'); +is($slabs->{'active_slabs'}, 1, "1 active slab"); Index: /tags/server/1.2.1/t/incrdecr.t =================================================================== --- /tags/server/1.2.1/t/incrdecr.t (revision 357) +++ /tags/server/1.2.1/t/incrdecr.t (revision 357) @@ -0,0 +1,44 @@ +#!/usr/bin/perl + +use strict; +use Test::More tests => 13; +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); +my $sock = $server->sock; + +print $sock "set num 0 0 1\r\n1\r\n"; +is(scalar <$sock>, "STORED\r\n", "stored num"); +mem_get_is($sock, "num", 1, "stored 1"); + +print $sock "incr num 1\r\n"; +is(scalar <$sock>, "2\r\n", "+ 1 = 2"); +mem_get_is($sock, "num", 2); + +print $sock "incr num 8\r\n"; +is(scalar <$sock>, "10\r\n", "+ 8 = 10"); +mem_get_is($sock, "num", 10); + +print $sock "decr num 1\r\n"; +is(scalar <$sock>, "9\r\n", "- 1 = 9"); + +print $sock "decr num 9\r\n"; +is(scalar <$sock>, "0\r\n", "- 9 = 0"); + +print $sock "decr num 5\r\n"; +is(scalar <$sock>, "0\r\n", "- 5 = 0"); + +print $sock "decr bogus 5\r\n"; +is(scalar <$sock>, "NOT_FOUND\r\n", "can't decr bogus key"); + +print $sock "decr incr 5\r\n"; +is(scalar <$sock>, "NOT_FOUND\r\n", "can't incr bogus key"); + +print $sock "set text 0 0 2\r\nhi\r\n"; +is(scalar <$sock>, "STORED\r\n", "stored text"); +print $sock "incr text 1\r\n"; +is(scalar <$sock>, "1\r\n", "hi - 1 = 0"); + + Index: /tags/server/1.2.1/t/managed-buckets.t =================================================================== --- /tags/server/1.2.1/t/managed-buckets.t (revision 351) +++ /tags/server/1.2.1/t/managed-buckets.t (revision 351) @@ -0,0 +1,11 @@ +#!/usr/bin/perl + +use strict; +use Test::More skip_all => "Tests not written."; # tests => 1 +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); +my $sock = $server->sock; + Index: /tags/server/1.2.1/t/slab-reassign.t =================================================================== --- /tags/server/1.2.1/t/slab-reassign.t (revision 351) +++ /tags/server/1.2.1/t/slab-reassign.t (revision 351) @@ -0,0 +1,11 @@ +#!/usr/bin/perl + +use strict; +use Test::More skip_all => "Tests not written."; # tests => 1 +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); +my $sock = $server->sock; + Index: /tags/server/1.2.1/t/getset.t =================================================================== --- /tags/server/1.2.1/t/getset.t (revision 340) +++ /tags/server/1.2.1/t/getset.t (revision 340) @@ -0,0 +1,51 @@ +#!/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; + +# 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"); + +# add bar (and should get it) +print $sock "add bar 0 0 6\r\nbarval\r\n"; +is(scalar <$sock>, "STORED\r\n", "stored barval"); +mem_get_is($sock, "bar", "barval"); + +# add foo (but shouldn't get new value) +print $sock "add foo 0 0 5\r\nfoov2\r\n"; +is(scalar <$sock>, "NOT_STORED\r\n", "not stored"); +mem_get_is($sock, "foo", "fooval"); + +# replace bar (should work) +print $sock "replace bar 0 0 6\r\nbarva2\r\n"; +is(scalar <$sock>, "STORED\r\n", "replaced barval 2"); + +# replace notexist (shouldn't work) +print $sock "replace notexist 0 0 6\r\nbarva2\r\n"; +is(scalar <$sock>, "NOT_STORED\r\n", "didn't replace notexist"); + +# delete foo. +print $sock "delete foo\r\n"; +is(scalar <$sock>, "DELETED\r\n", "deleted foo"); + +# delete foo again. not found this time. +print $sock "delete foo\r\n"; +is(scalar <$sock>, "NOT_FOUND\r\n", "deleted foo, but not found"); + +# pipeling is okay +print $sock "set foo 0 0 6\r\nfooval\r\ndelete foo\r\nset foo 0 0 6\r\nfooval\r\ndelete foo\r\n"; +is(scalar <$sock>, "STORED\r\n", "pipeline set"); +is(scalar <$sock>, "DELETED\r\n", "pipeline delete"); +is(scalar <$sock>, "STORED\r\n", "pipeline set"); +is(scalar <$sock>, "DELETED\r\n", "pipeline delete"); + + + Index: /tags/server/1.2.1/t/unixsocket.t =================================================================== --- /tags/server/1.2.1/t/unixsocket.t (revision 351) +++ /tags/server/1.2.1/t/unixsocket.t (revision 351) @@ -0,0 +1,11 @@ +#!/usr/bin/perl + +use strict; +use Test::More skip_all => "Tests not written."; # tests => 1 +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); +my $sock = $server->sock; + Index: /tags/server/1.2.1/t/flags.t =================================================================== --- /tags/server/1.2.1/t/flags.t (revision 342) +++ /tags/server/1.2.1/t/flags.t (revision 342) @@ -0,0 +1,18 @@ +#!/usr/bin/perl + +use strict; +use Test::More tests => 6; +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); +my $sock = $server->sock; + +# set foo (and should get it) +for my $flags (0, 123, 2**16-1) { + print $sock "set foo $flags 0 6\r\nfooval\r\n"; + is(scalar <$sock>, "STORED\r\n", "stored foo"); + mem_get_is({ sock => $sock, + flags => $flags }, "foo", "fooval", "got flags $flags back"); +} Index: /tags/server/1.2.1/t/stats.t =================================================================== --- /tags/server/1.2.1/t/stats.t (revision 351) +++ /tags/server/1.2.1/t/stats.t (revision 351) @@ -0,0 +1,11 @@ +#!/usr/bin/perl + +use strict; +use Test::More skip_all => "Tests not written."; # tests => 1 +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); +my $sock = $server->sock; + Index: /tags/server/1.2.1/t/multiversioning.t =================================================================== --- /tags/server/1.2.1/t/multiversioning.t (revision 349) +++ /tags/server/1.2.1/t/multiversioning.t (revision 349) @@ -0,0 +1,46 @@ +#!/usr/bin/perl + +use strict; +use Test::More tests => 13; +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); +my $sock = $server->sock; +my $sock2 = $server->new_sock; + +ok($sock != $sock2, "have two different connections open"); + +# set large value +my $size = 256 * 1024; # 256 kB +my $bigval = "0123456789abcdef" x ($size / 16); +$bigval =~ s/^0/\[/; $bigval =~ s/f$/\]/; +my $bigval2 = uc($bigval); + +print $sock "set big 0 0 $size\r\n$bigval\r\n"; +is(scalar <$sock>, "STORED\r\n", "stored foo"); +mem_get_is($sock, "big", $bigval, "big value got correctly"); + +print $sock "get big\r\n"; +my $buf; +is(read($sock, $buf, $size / 2), $size / 2, "read half the answer back"); +like($buf, qr/VALUE big/, "buf has big value header in it"); +like($buf, qr/abcdef/, "buf has some data in it"); +unlike($buf, qr/abcde\]/, "buf doesn't yet close"); + +# sock2 interrupts (maybe sock1 is slow) and deletes stuff: +print $sock2 "delete big\r\n"; +is(scalar <$sock2>, "DELETED\r\n", "deleted big from sock2 while sock1's still reading it"); +mem_get_is($sock2, "big", undef, "nothing from sock2 now. gone from namespace."); +print $sock2 "set big 0 0 $size\r\n$bigval2\r\n"; +is(scalar <$sock2>, "STORED\r\n", "stored big w/ val2"); +mem_get_is($sock2, "big", $bigval2, "big value2 got correctly"); + +# sock1 resumes reading... +$buf .= <$sock>; +$buf .= <$sock>; +like($buf, qr/abcde\]/, "buf now closes"); + +# and if sock1 reads again, it's the uppercase version: +mem_get_is($sock, "big", $bigval2, "big value2 got correctly from sock1"); Index: /tags/server/1.2.1/t/stress-memcached.pl =================================================================== --- /tags/server/1.2.1/t/stress-memcached.pl (revision 406) +++ /tags/server/1.2.1/t/stress-memcached.pl (revision 406) @@ -0,0 +1,101 @@ +#!/usr/bin/perl +# + +use strict; +use lib '../../api/perl/lib'; +use Cache::Memcached; +use Time::HiRes qw(time); + +unless (@ARGV == 2) { + die "Usage: stress-memcached.pl ip:port threads\n"; +} + +my $host = shift; +my $threads = shift; + +my $memc = new Cache::Memcached; +$memc->set_servers([$host]); + +unless ($memc->set("foo", "bar") && + $memc->get("foo") eq "bar") { + die "memcached not running at $host ?\n"; +} +$memc->disconnect_all(); + + +my $running = 0; +while (1) { + if ($running < $threads) { + my $cpid = fork(); + if ($cpid) { + $running++; + #print "Launched $cpid. Running $running threads.\n"; + } else { + stress(); + exit 0; + } + } else { + wait(); + $running--; + } +} + +sub stress { + undef $memc; + $memc = new Cache::Memcached; + $memc->set_servers([$host]); + + my ($t1, $t2); + my $start = sub { $t1 = time(); }; + my $stop = sub { + my $op = shift; + $t2 = time(); + my $td = sprintf("%0.3f", $t2 - $t1); + if ($td > 0.25) { print "Took $td seconds for: $op\n"; } + }; + + my $max = rand(50); + my $sets = 0; + + for (my $i = 0; $i < $max; $i++) { + my $key = key($i); + my $set = $memc->set($key, $key); + $sets++ if $set; + } + + for (1..int(rand(500))) { + my $rand = int(rand($max)); + my $key = key($rand); + my $meth = int(rand(3)); + my $exp = int(rand(3)); + undef $exp unless $exp; + $start->(); + if ($meth == 0) { + $memc->add($key, $key, $exp); + $stop->("add"); + } elsif ($meth == 1) { + $memc->delete($key); + $stop->("delete"); + } else { + $memc->set($key, $key, $exp); + $stop->("set"); + } + $rand = int(rand($max)); + $key = key($rand); + $start->(); + my $v = $memc->get($key); + $stop->("get"); + if ($v && $v ne $key) { die "Bogus: $v for key $rand\n"; } + } + + $start->(); + my $multi = $memc->get_multi(map { key(int(rand($max))) } (1..$max)); + $stop->("get_multi"); +} + +sub key { + my $n = shift; + $_ = sprintf("%04d", $n); + if ($n % 2) { $_ .= "a"x20; } + $_; +} Index: /tags/server/1.2.1/t/binary-get.t =================================================================== --- /tags/server/1.2.1/t/binary-get.t (revision 351) +++ /tags/server/1.2.1/t/binary-get.t (revision 351) @@ -0,0 +1,11 @@ +#!/usr/bin/perl + +use strict; +use Test::More skip_all => "Tests not written."; # tests => 1 +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); +my $sock = $server->sock; + Index: /tags/server/1.2.1/t/lru.t =================================================================== --- /tags/server/1.2.1/t/lru.t (revision 351) +++ /tags/server/1.2.1/t/lru.t (revision 351) @@ -0,0 +1,11 @@ +#!/usr/bin/perl + +use strict; +use Test::More skip_all => "Tests not written."; # tests => 1 +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); +my $sock = $server->sock; + Index: /tags/server/1.2.1/t/lib/MemcachedTest.pm =================================================================== --- /tags/server/1.2.1/t/lib/MemcachedTest.pm (revision 381) +++ /tags/server/1.2.1/t/lib/MemcachedTest.pm (revision 381) @@ -0,0 +1,153 @@ +package MemcachedTest; +use strict; +use IO::Socket::INET; +use Exporter 'import'; +use FindBin qw($Bin); +use Carp qw(croak); +use vars qw(@EXPORT); + +@EXPORT = qw(new_memcached sleep mem_get_is mem_stats free_port); + +sub sleep { + my $n = shift; + select undef, undef, undef, $n; +} + +sub mem_stats { + my ($sock, $type) = @_; + $type = $type ? " $type" : ""; + print $sock "stats$type\r\n"; + my $stats = {}; + while (<$sock>) { + last if /^(\.|END)/; + /^STAT (\S+) (\d+)/; + #print " slabs: $_"; + $stats->{$1} = $2; + } + return $stats; +} + +sub mem_get_is { + # works on single-line values only. no newlines in value. + my ($sock_opts, $key, $val, $msg) = @_; + my $opts = ref $sock_opts eq "HASH" ? $sock_opts : {}; + my $sock = ref $sock_opts eq "HASH" ? $opts->{sock} : $sock_opts; + + my $expect_flags = $opts->{flags} || 0; + my $dval = defined $val ? "'$val'" : ""; + $msg ||= "$key == $dval"; + + print $sock "get $key\r\n"; + if (! defined $val) { + my $line = scalar <$sock>; + if ($line =~ /^VALUE/) { + $line .= scalar(<$sock>) . scalar(<$sock>); + } + Test::More::is($line, "END\r\n", $msg); + } else { + my $len = length($val); + my $body = scalar(<$sock>); + my $expected = "VALUE $key $expect_flags $len\r\n$val\r\nEND\r\n"; + if (!$body || $body =~ /^END/) { + Test::More::is($body, $expected, $msg); + return; + } + $body .= scalar(<$sock>) . scalar(<$sock>); + Test::More::is($body, $expected, $msg); + } +} + +sub free_port { + my $type = shift || "tcp"; + my $sock; + my $port; + while (!$sock) { + $port = int(rand(20000)) + 30000; + $sock = IO::Socket::INET->new(LocalAddr => '127.0.0.1', + LocalPort => $port, + Proto => $type, + ReuseAddr => 1); + } + return $port; +} + +sub supports_udp { + my $output = `$Bin/../memcached-debug -h`; + return 0 if $output =~ /^memcached 1\.1\./; + return 1; +} + +sub new_memcached { + my $args = shift || ""; + my $port = free_port(); + my $udpport = free_port("udp"); + $args .= " -p $port"; + if (supports_udp()) { + $args .= " -U $udpport"; + } + if ($< == 0) { + $args .= " -u root"; + } + my $childpid = fork(); + + my $exe = "$Bin/../memcached-debug"; + croak("memcached binary doesn't exist. Haven't run 'make' ?\n") unless -e $exe; + croak("memcached binary not executable\n") unless -x _; + + unless ($childpid) { + exec "$exe $args"; + exit; # never gets here. + } + + for (1..20) { + my $conn = IO::Socket::INET->new(PeerAddr => "127.0.0.1:$port"); + if ($conn) { + return Memcached::Handle->new(pid => $childpid, + conn => $conn, + udpport => $udpport, + port => $port); + } + select undef, undef, undef, 0.10; + } + croak("Failed to startup/connect to memcached server."); + +} + +############################################################################ +package Memcached::Handle; +sub new { + my ($class, %params) = @_; + return bless \%params, $class; +} + +sub DESTROY { + my $self = shift; + kill 9, $self->{pid}; +} + +sub port { $_[0]{port} } +sub udpport { $_[0]{udpport} } + +sub sock { + my $self = shift; + return $self->{conn} if $self->{conn} && getpeername($self->{conn}); + return $self->new_sock; +} + +sub new_sock { + my $self = shift; + return IO::Socket::INET->new(PeerAddr => "127.0.0.1:$self->{port}"); +} + +sub new_udp_sock { + my $self = shift; + return IO::Socket::INET->new(PeerAddr => '127.0.0.1', + PeerPort => $self->{udpport}, + Proto => 'udp', + LocalAddr => '127.0.0.1', + LocalPort => MemcachedTest::free_port('udp'), + ); + +} + +1; Index: /tags/server/1.2.1/t/expirations.t =================================================================== --- /tags/server/1.2.1/t/expirations.t (revision 390) +++ /tags/server/1.2.1/t/expirations.t (revision 390) @@ -0,0 +1,49 @@ +#!/usr/bin/perl + +use strict; +use Test::More tests => 8; +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); +my $sock = $server->sock; +my $expire; + +sub wait_for_early_second { + my $have_hires = eval "use Time::HiRes (); 1"; + if ($have_hires) { + my $tsh = Time::HiRes::time(); + my $ts = int($tsh); + return if ($tsh - $ts) < 0.5; + } + + my $ts = int(time()); + while (1) { + my $t = int(time()); + return if $t != $ts; + select undef, undef, undef, 0.10; # 1/10th of a second sleeps until time changes. + } +} + +wait_for_early_second(); + +print $sock "set foo 0 1 6\r\nfooval\r\n"; +is(scalar <$sock>, "STORED\r\n", "stored foo"); + +mem_get_is($sock, "foo", "fooval"); +sleep(1.5); +mem_get_is($sock, "foo", undef); + +$expire = time() - 1; +print $sock "set foo 0 $expire 6\r\nfooval\r\n"; +is(scalar <$sock>, "STORED\r\n", "stored foo"); +mem_get_is($sock, "foo", undef, "already expired"); + +$expire = time() + 1; +print $sock "set foo 0 $expire 6\r\nfoov+1\r\n"; +is(scalar <$sock>, "STORED\r\n", "stored foo"); +mem_get_is($sock, "foo", "foov+1"); +sleep(2.2); +mem_get_is($sock, "foo", undef, "now expired"); + Index: /tags/server/1.2.1/t/delete-window.t =================================================================== --- /tags/server/1.2.1/t/delete-window.t (revision 340) +++ /tags/server/1.2.1/t/delete-window.t (revision 340) @@ -0,0 +1,68 @@ +#!/usr/bin/perl + +use strict; +use Test::More tests => 20; +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +my $server = new_memcached(); +my $sock = $server->sock; +my $line = sub { return scalar <$sock> }; + +# immediate set/deletes +print $sock "set foo 0 0 6\r\nfooval\r\ndelete foo\r\nset foo 0 0 6\r\nfooval\r\ndelete foo\r\n"; +is($line->(), "STORED\r\n", "pipeline set"); +is($line->(), "DELETED\r\n", "pipeline delete"); +is($line->(), "STORED\r\n", "pipeline set"); +is($line->(), "DELETED\r\n", "pipeline delete"); + +# not found test +print $sock "delete foo\r\n"; +is($line->(), "NOT_FOUND\r\n", "thing not found to delete"); + +# test the cool-down window (see protocol doc) whereby add/replace commands can't +# work n seconds after deleting. +print $sock "set foo 0 0 3\r\nbar\r\n"; +is($line->(), "STORED\r\n", "stored foo"); +print $sock "delete foo 1\r\n"; +is($line->(), "DELETED\r\n", "deleted with 1 second window"); +mem_get_is($sock, "foo", undef); +print $sock "add foo 0 0 7\r\nfoo-add\r\n"; +is($line->(), "NOT_STORED\r\n", "didn't add foo"); +print $sock "replace foo 0 0 11\r\nfoo-replace\r\n"; +is($line->(), "NOT_STORED\r\n", "didn't replace foo"); +print $sock "set foo 0 0 7\r\nfoo-set\r\n"; +is($line->(), "STORED\r\n", "stored foo-set"); + +# add can work after expiration time +print $sock "set foo 0 0 3\r\nbar\r\n"; +is($line->(), "STORED\r\n", "stored foo"); +print $sock "delete foo 1\r\n"; +is($line->(), "DELETED\r\n", "deleted with 1 second window"); +sleep(1.2); +print $sock "add foo 0 0 7\r\nfoo-add\r\n"; +is($line->(), "STORED\r\n", "stored foo-add"); + +mem_get_is($sock, "foo", "foo-add", "foo == 'foo-add' (before deleter)"); + +# test 'baz' with set, delete w/ timer, set, wait 5.2 seconds (for 5 +# second deleter event), then get to see which we get. +print $sock "set baz 0 0 4\r\nval1\r\n"; +is($line->(), "STORED\r\n", "stored baz = val1"); +print $sock "delete baz 1\r\n"; +is($line->(), "DELETED\r\n", "deleted with 1 second window"); +print $sock "set baz 0 0 4\r\nval2\r\n"; +is($line->(), "STORED\r\n", "stored baz = val2"); + +diag("waiting 5 seconds for the deleter event..."); +sleep(5.2); + +# follow-up on 1st test: +mem_get_is($sock, "foo", "foo-add", "foo == 'foo-add' (after deleter)"); + +# and follow-up on 2nd test: +mem_get_is($sock, "baz", "val2", "baz=='val2'"); + + + Index: /tags/server/1.2.1/t/daemonize.t =================================================================== --- /tags/server/1.2.1/t/daemonize.t (revision 356) +++ /tags/server/1.2.1/t/daemonize.t (revision 356) @@ -0,0 +1,33 @@ +#!/usr/bin/perl + +use strict; +use Test::More tests => 7; +use FindBin qw($Bin); +use lib "$Bin/lib"; +use MemcachedTest; + +use File::Temp qw(tempfile); + +my (undef, $tmpfn) = tempfile(); + +my $server = new_memcached("-d -P $tmpfn"); +my $sock = $server->sock; +sleep 0.5; + +ok(-e $tmpfn, "pid file exists"); +ok(-s $tmpfn, "pid file has length"); + +open (my $fh, $tmpfn) or die; +my $readpid = do { local $/; <$fh>; }; +chomp $readpid; +close ($fh); + +ok(kill(0, $readpid), "process is still running"); + +my $stats = mem_stats($sock); +is($stats->{pid}, $readpid, "memcached reports same pid as file"); + +ok($server->new_sock, "opened new socket"); +ok(kill(9, $readpid), "sent KILL signal"); +sleep 0.5; +ok(! $server->new_sock, "failed to open new socket"); Index: /tags/server/1.2.1/configure.ac =================================================================== --- /tags/server/1.2.1/configure.ac (revision 453) +++ /tags/server/1.2.1/configure.ac (revision 453) @@ -0,0 +1,161 @@ +AC_PREREQ(2.52) +AC_INIT(memcached, 1.2.1, 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 +AC_PROG_INSTALL + +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_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 + +AC_CHECK_FUNCS(mlockall) + +AC_CONFIG_FILES(Makefile doc/Makefile) +AC_OUTPUT Index: /tags/server/1.2.1/doc/protocol.txt =================================================================== --- /tags/server/1.2.1/doc/protocol.txt (revision 357) +++ /tags/server/1.2.1/doc/protocol.txt (revision 357) @@ -0,0 +1,437 @@ +Protocol +-------- + +Clients of memcached communicate with server through TCP connections. +(A UDP interface is also available; details are below under "UDP +protocol.") A given running memcached server listens on some +(configurable) port; clients connect to that port, send commands to +the server, read responses, and eventually close the connection. + +There is no need to send any command to end the session. A client may +just close the connection at any moment it no longer needs it. Note, +however, that clients are encouraged to cache their connections rather +than reopen them every time they need to store or retrieve data. This +is because memcached is especially designed to work very efficiently +with a very large number (many hundreds, more than a thousand if +necessary) of open connections. Caching connections will eliminate the +overhead associated with establishing a TCP connection (the overhead +of preparing for a new connection on the server side is insignificant +compared to this). + +There are two kinds of data sent in the memcache protocol: text lines +and unstructured data. Text lines are used for commands from clients +and responses from servers. Unstructured data is sent when a client +wants to store or retrieve data. The server will transmit back +unstructured data in exactly the same way it received it, as a byte +stream. The server doesn't care about byte order issues in +unstructured data and isn't aware of them. There are no limitations on +characters that may appear in unstructured data; however, the reader +of such data (either a client or a server) will always know, from a +preceding text line, the exact length of the data block being +transmitted. + +Text lines are always terminated by \r\n. Unstructured data is _also_ +terminated by \r\n, even though \r, \n or any other 8-bit characters +may also appear inside the data. Therefore, when a client retrieves +data from a server, it must use the length of the data block (which it +will be provided with) to determine where the data block ends, and not +the fact that \r\n follows the end of the data block, even though it +does. + +Keys +---- + +Data stored by memcached is identified with the help of a key. A key +is a text string which should uniquely identify the data for clients +that are interested in storing and retrieving it. Currently the +length limit of a key is set at 250 characters (of course, normally +clients wouldn't need to use such long keys); the key must not include +control characters or whitespace. + +Commands +-------- + +There are three types of commands. + +Storage commands (there are three: "set", "add" and "replace") ask the +server to store some data identified by a key. The client sends a +command line, and then a data block; after that the client expects one +line of response, which will indicate success or faulure. + +Retrieval commands (there is only one: "get") ask the server to +retrieve data corresponding to a set of keys (one or more keys in one +request). The client sends a command line, which includes all the +requested keys; after that for each item the server finds it sends to +the client one response line with information about the item, and one +data block with the item's data; this continues until the server +finished with the "END" response line. + +All other commands don't involve unstructured data. In all of them, +the client sends one command line, and expects (depending on the +command) either one line of response, or several lines of response +ending with "END" on the last line. + +A command line always starts with the name of the command, followed by +parameters (if any) delimited by whitespace. Command names are +lower-case and are case-sensitive. + +Expiration times +---------------- + +Some commands involve a client sending some kind of expiration time +(relative to an item or to an operation requested by the client) to +the server. In all such cases, the actual value sent may either be +Unix time (number of seconds since January 1, 1970, as a 32-bit +value), or a number of seconds starting from current time. In the +latter case, this number of seconds may not exceed 60*60*24*30 (number +of seconds in 30 days); if the number sent by a client is larger than +that, the server will consider it to be real Unix time value rather +than an offset from current time. + + +Error strings +------------- + +Each command sent by a client may be answered with an error string +from the server. These error strings come in three types: + +- "ERROR\r\n" + + means the client sent a nonexistent command name. + +- "CLIENT_ERROR \r\n" + + means some sort of client error in the input line, i.e. the input + doesn't conform to the protocol in some way. is a + human-readable error string. + +- "SERVER_ERROR \r\n" + + means some sort of server error prevents the server from carrying + out the command. is a human-readable error string. In cases + of severe server errors, which make it impossible to continue + serving the client (this shouldn't normally happen), the server will + close the connection after sending the error line. This is the only + case in which the server closes a connection to a client. + + +In the descriptions of individual commands below, these error lines +are not again specifically mentioned, but clients must allow for their +possibility. + + +Storage commands +---------------- + +First, the client sends a command line which looks like this: + + \r\n + +- is "set", "add" or "replace" + + "set" means "store this data". + + "add" means "store this data, but only if the server *doesn't* already + hold data for this key". + + "replace" means "store this data, but only if the server *does* + already hold data for this key". + +- is the key under which the client asks to store the data + +- is an arbitrary 16-bit unsigned integer (written out in + decimal) that the server stores along with the data and sends back + when the item is retrieved. Clients may use this as a bit field to + store data-specific information; this field is opaque to the server. + +- is expiration time. If it's 0, the item never expires + (although it may be deleted from the cache to make place for other + items). If it's non-zero (either Unix time or offset in seconds from + current time), it is guaranteed that clients will not be able to + retrieve this item after the expiration time arrives (measured by + server time). + +- is the number of bytes in the data block to follow, *not* + including the delimiting \r\n. may be zero (in which case + it's followed by an empty data block). + +After this line, the client sends the data block: + +\r\n + +- is a chunk of arbitrary 8-bit data of length + from the previous line. + +After sending the command line and the data blockm the client awaits +the reply, which may be: + +- "STORED\r\n", to indicate success. + +- "NOT_STORED\r\n" to indicate the data was not stored, but not +because of an error. This normally means that either that the +condition for an "add" or a "replace" command wasn't met, or that the +item is in a delete queue (see the "delete" command below). + + +Retrieval command: +------------------ + +The retrieval command looks like this: + +get *\r\n + +- * means one or more key strings separated by whitespace. + +After this command, the client expects zero or more items, each of +which is received as a text line followed by a data block. After all +the items have been transmitted, the server sends the string + +"END\r\n" + +to indicate the end of response. + +Each item sent by the server looks like this: + +VALUE \r\n +\r\n + +- is the key for the item being sent + +- is the flags value set by the storage command + +- is the length of the data block to follow, *not* including + its delimiting \r\n + +- is the data for this item. + +If some of the keys appearing in a retrieval request are not sent back +by the server in the item list this means that the server does not +hold items with such keys (because they were never stored, or stored +but deleted to make space for more items, or expired, or explicitly +deleted by a client). + + + +Deletion +-------- + +The command "delete" allows for explicit deletion of items: + +delete