File Coverage

blib/lib/Cache/Memcached/libmemcached.pm

Criterion	Covered	Total	%
statement	7	9	77.7
branch			n/a
condition			n/a
subroutine	3	3	100.0
pod			n/a
total	10	12	83.3

package Cache::Memcached::libmemcached;

require bytes;
use strict;
use warnings;

use Memcached::libmemcached 1.001701, qw(
    MEMCACHED_CALLBACK_PREFIX_KEY
    MEMCACHED_PREFIX_KEY_MAX_SIZE
);
use base qw(Memcached::libmemcached);

use Carp qw(croak carp);
use Scalar::Util qw(weaken);
use Storable ();

our $VERSION = '0.04001';

use constant HAVE_ZLIB    => eval { require Compress::Zlib } && !$@;
use constant F_STORABLE   => 1;
use constant F_COMPRESS   => 2;
use constant OPTIMIZE     => $ENV{PERL_LIBMEMCACHED_OPTIMIZE} ? 1 : 0;

my %behavior;

BEGIN
{
    # Make sure to load bytes.pm if HAVE_ZLIB is enabled
    if (HAVE_ZLIB) {
        require bytes;
    }

    # accessors
    foreach my $field (qw(compress_enable compress_threshold compress_savings)) {
        eval sprintf(<<"        EOSUB", $field, $field, $field, $field);
            sub set_%s { \$_[0]->{%s} = \$_[1] }
            sub get_%s { \$_[0]->{%s} }
        EOSUB
        die if $@;
    }
    # for Cache::Memcached compatibility
    sub enable_compress { shift->set_compress_enable(@_) }

    # XXX this should be done via subclasses
    if (OPTIMIZE) {
        # If the optimize flag is enabled, we do not support master key
        # generation, cause we really care about the speed.
        foreach my $method (qw(get set add replace prepend append cas delete)) {
            eval <<"            EOSUB";
                sub $method {
                    shift->SUPER::memcached_${method}(\@_)
                }
            EOSUB
            die if $@;
        }
    } else {
        # Regular case.
        # Mental note. We only do this cause while we're faster than
        # Cache::Memcached::Fast, *even* when the above optimization isn't
        # toggled.
        foreach my $method (qw(get set add replace prepend append cas delete)) {
            eval <<"            EOSUB";
                sub $method {
                    my \$self = shift;
                    my \$key  = shift;
                    return \$self->SUPER::memcached_${method}(\$key, \@_)
                        unless ref \$key;
                    (my \$master_key, \$key) = @\$key;
                    if (\$master_key) {
                        \$self->SUPER::memcached_${method}_by_key(\$master_key, \$key, \@_);
                    } else {
                        \$self->SUPER::memcached_${method}(\$key, \@_);
                    }
                }
            EOSUB
            die if $@;
        }
    }

    # Create get_*/is_*/set_* methods for some libmemcached behaviors.
    # We only do this for some because there are many and it's easy for
    # the user to use memcached_behavior_set() etc directly.
    #
    %behavior = (
        # non-boolean behaviors that are renamed (to be more descriptive)
        distribution_method => [ 0, 'distribution' ],
        hashing_algorithm   => [ 0, 'hash' ],
        # boolean behaviors that are not renamed:
        no_block        => [ 1 ],
        binary_protocol => [ 1 ],
    );

    while ( my ($method, $field_info) = each %behavior ) {
        my $is_bool = $field_info->[0];
        my $field   = $field_info->[1] || $method;

        my $behavior = "Memcached::libmemcached::MEMCACHED_BEHAVIOR_\U$field";
        warn "$behavior doesn't exist\n" # sanity check
            unless do { no strict 'refs'; defined &$behavior };

        my ($set, $get) = ("set_$method", "get_$method");
        $get = "is_$method" if $is_bool;
        my $code = "sub $set { \$_[0]->memcached_behavior_set($behavior(), \$_[1]) }\n"
                 . "sub $get { \$_[0]->memcached_behavior_get($behavior()) }";
        eval $code;
        die "$@ while executing $code" if $@;
    }

}

sub import
{
    my $class = shift;
    Memcached::libmemcached->export_to_level(1, undef, @_) ;
}

sub new
{
    my $class = shift;
    my %args  = %{ shift || {} };

    my $self = $class->SUPER::new();

    $self->trace_level(delete $args{debug}) if exists $args{debug};

    $self->namespace(delete $args{namespace})
        if exists $args{namespace};

    $self->{compress_threshold} = delete $args{compress_threshold};
    # Add support for Cache::Memcache::Fast's compress_ratio
    $self->{compress_savingsS}  = delete $args{compress_savings} || 0.20;
    $self->{compress_enable}    =
        exists $args{compress_enable} ? delete $args{compress_enable} : 1;

    # servers
    $args{servers} || croak "No servers specified";
    $self->set_servers(delete $args{servers});

    # old-style behavior options (see behavior_ block below)
    foreach my $option (qw(no_block hashing_algorithm distribution_method binary_protocol)) {
        my $behavior = $behavior{$option}->[1] || $option;
        $args{"behavior_$behavior"} = delete $args{$option} if exists $args{$option};
    }

    # allow any libmemcached behavior to be set via args to new()
    for my $name (grep { /^behavior_/ } keys %args) {
        my $value = delete $args{$name};
        my $behavior = "Memcached::libmemcached::MEMCACHED_\U$name";
        no strict 'refs';
        if (not defined &$behavior) {
            carp "$name ($behavior) isn't available"; # sanity check
            next;
        }
        $self->memcached_behavior_set(&$behavior(), $value);
    }

    delete $args{readonly};
    delete $args{no_rehash};

    carp "Unrecognised options: @{[ sort keys %args ]}"
        if %args;

    # Set compression/serialization callbacks
    $self->set_callback_coderefs(
        # Closures so we have reference to $self
        $self->_mk_callbacks()
    );

    # behavior options
    foreach my $option (qw(no_block hashing_algorithm distribution_method binary_protocol)) {
        my $method = "set_$option";
        $self->$method( $args{$option} ) if exists $args{$option};
    }

    return $self;
}

sub namespace {
    my $self = shift;

    my $old_namespace = $self->memcached_callback_get(MEMCACHED_CALLBACK_PREFIX_KEY);
    if (@_) {
        my $namespace = shift;
        $self->memcached_callback_set(MEMCACHED_CALLBACK_PREFIX_KEY, $namespace)
            or carp $self->errstr;
    }

    return $old_namespace;
}

sub set_servers
{
    my $self = shift;
    my $servers = shift || [];

    # $self->{servers} = []; # for compatibility with Cache::Memcached

    # XXX should delete any existing servers from libmemcached
    foreach my $server (@$servers) {
        $self->server_add($server);
    }
}

sub server_add
{
    my $self = shift;
    my $server = shift
        or Carp::confess("server not specified");

    my $weight = 0;
    if (ref $server eq 'ARRAY') {
        my @ary = @$server;
        $server = shift @ary;
        $weight = shift @ary || 0 if @ary;
    }
    elsif (ref $server eq 'HASH') { # Cache::Memcached::Fast
        my $h = $server;
        $server = $h->{address};
        $weight = $h->{weight} if exists $h->{weight};
        # noreply is not supported
    }

    if ($server =~ /^([^:]+):([^:]+)$/) {
        my ($hostname, $port) = ($1, $2);
        $self->memcached_server_add_with_weight($hostname, $port, $weight);
    } else {
        $self->memcached_server_add_unix_socket_with_weight( $server, $weight );
    }

    # for compatibility with Cache::Memcached
    # push @{$self->{servers}}, $server;
}


sub _mk_callbacks
{
    my $self = shift;

    weaken($self);
    my $inflate = sub {
        my ($key, $flags) = @_;
        if ($flags & F_COMPRESS) {
            if (! HAVE_ZLIB) {
                croak("Data for $key is compressed, but we have no Compress::Zlib");
            }
            $_ = Compress::Zlib::memGunzip($_);
        }

        if ($flags & F_STORABLE) {
            $_ = Storable::thaw($_);
        }
        return ();
    };

    my $deflate = sub {
        # Check if we have a complex structure
        if (ref $_) {
            $_ = Storable::nfreeze($_);
            $_[1] |= F_STORABLE;
        }

        # Check if we need compression
        if (HAVE_ZLIB && $self->{compress_enable} && $self->{compress_threshold}) {
            # Find the byte length
            my $length = bytes::length($_);
            if ($length > $self->{compress_threshold}) {
                my $tmp = Compress::Zlib::memGzip($_);
                if (bytes::length($tmp) / $length < 1 - $self->{compress_savingsS}) {
                    $_ = $tmp;
                    $_[1] |= F_COMPRESS;
                }
            }
        }
        return ();
    };
    return ($deflate, $inflate);
}

sub incr
{
    my $self = shift;
    my $key  = shift;
    my $offset = shift || 1;
    my $val = 0;
    $self->memcached_increment($key, $offset, $val) || return undef;
    return $val;
}

sub decr
{
    my $self = shift;
    my $key  = shift;
    my $offset = shift || 1;
    my $val = 0;
    $self->memcached_decrement($key, $offset, $val) || return undef;
    return $val;
}


sub flush_all
{
    $_[0]->memcached_flush(0);
}

*remove = \&delete;

sub disconnect_all {
    $_[0]->memcached_quit();
}


sub server_versions {
    my $self = shift;
    my %versions;
    # XXX not optimal, libmemcached knows these values without having to send a stats request
    $self->walk_stats('', sub {
        my ($key, $value, $hostport) = @_;
        $versions{$hostport} = $value if $key eq 'version';
        return;
    });
    return \%versions;
}


sub stats
{
    my $self = shift;
    my ($stats_args) = @_;

    # http://github.com/memcached/memcached/blob/master/doc/protocol.txt
    $stats_args = [ $stats_args ]
        if $stats_args and not ref $stats_args;
    $stats_args ||= [ '' ];

    # stats keys that aren't matched by the prefix and suffix regexes below
    # but which we want to accumulate in totals
    my %total_misc_keys = map { ($_ => 1) } qw(
        bytes evictions
        connection_structures curr_connections total_connections
    );

    my %h;
    for my $type (@$stats_args) {

        my $code = sub {
            my ($key, $value, $hostport) = @_;

            # XXX - This is hardcoded in the callback cause r139 in perl-memcached
            # removed the magic of "misc"
            $type ||= 'misc';
            $h{hosts}{$hostport}{$type}{$key} = $value;
            #warn "$_ ($key, $value, $hostport, $type)\n";

            # accumulate overall totals for some items
            if ($type eq 'misc') {
                if ($total_misc_keys{$key}
                or $key =~ /^(?:cmd|bytes)_/ # prefixes
                or $key =~ /_(?:hits|misses|errors|yields|badval|items|read|written)$/ # suffixes
                ) {
                    $h{total}{$key} += $value;
                }
            }
            elsif ($type eq 'malloc' or $type eq 'sizes') {
                $h{total}{"${type}_$key"} += $value;
            }
            return;
        };

        $self->walk_stats($type, $code);
    }

    return \%h;
}

# for compatability with Cache::Memcached and Cache::Memcached::Managed 0.20:
# https://rt.cpan.org/Ticket/Display.html?id=62512
# sub sock_to_host { undef }
# sub get_sock { undef }
# sub forget_dead_hosts { undef }

1;

__END__