Cache abstraction layer with Redis support

lib/MogileFS/Cache.pm

@@ -0,0 +1,100 @@
+package MogileFS::Cache;
+use strict;
+use MogileFS::Util qw(error);
+
+my $have_memc_module = eval "use Cache::Memcached; 1;";
+my $have_redis_module = eval "use Redis; 1;";
+
+sub new {
+    my ($class) = @_;
+    return $class->new_from_config;
+}
+
+sub new_from_config {
+    my ($class) = @_;
+    return undef unless ($have_memc_module || $have_redis_module);
+
+    my ($type, $servers, $cache_ttl);
+
+    if ($servers = MogileFS::Config->server_setting_cached('memcache_servers')) {
+        $type = 'memcache';
+        $cache_ttl = MogileFS::Config->server_setting_cached('memcache_ttl') || 3600;
+    } else {
+        $servers = MogileFS->config('cache_servers');
+        $type = MogileFS->config('cache_type');
+        $cache_ttl = MogileFS->config('cache_ttl') || 3600;
+    }
+
+    my $subclass;
+    if ($type eq 'none' || $type eq '') {
+        return undef;
+    } elsif ($type eq 'memcache' && $have_memc_module) {
+        $subclass = 'MogileFS::Cache::Memcache';
+    } elsif ($type eq 'redis' && $have_redis_module) {
+        $subclass = 'MogileFS::Cache::Redis';
+    } else {
+        error("Cache type not supported: $type");
+        return undef;
+    }
+    unless (eval "use $subclass; 1") {
+        error("Error loading $subclass: $@");
+        return undef;
+    }
+
+    my $self = bless {
+        servers => $servers,
+        ttl     => $cache_ttl,
+    }, $subclass;
+    $self->init;
+    return $self;
+}
+
+sub init { 1 }
+
+sub refresh {
+    my $self = shift;
+    return $self;
+}
+
+sub set {
+    my ($self, $key, $value) = @_;
+    die "set not implemented for $self";
+}
+
+sub get {
+    my ($self, $key) = @_;
+    die "get not implemented for $self";
+}
+
+sub delete {
+    my ($self, $key) = @_;
+    die "delete not implemented for $self";
+}
+
+sub hash {
+    my ($self, $key) = @_;
+    if ($key->{type} eq 'fid') {
+        return 'mogf:' . $key->{domain} . ':' . $key->{key};
+    } elsif ($key->{type} eq 'devid') {
+        return 'mogd:' . $key->{fid};
+    }
+    return $key;
+}
+
+1;
+
+__END__
+
+=head1 NAME
+
+MogileFS::Cache - path cache provider.  base class.
+
+=head1 ABOUT
+
+Caches key-path mappings for faster lookup.
+
+=head1 SEE ALSO
+
+L<MogileFS::Cache::Memcache>
+
+

lib/MogileFS/Cache/Memcache.pm

@@ -0,0 +1,64 @@
+package MogileFS::Cache::Memcache;
+use strict;
+use base 'MogileFS::Cache';
+use Cache::Memcached;
+
+# --------------------------------------------------------------------------
+# Memcache mappings are as follows:
+#   mogf:<dmid>:<dkey> -> fidid
+#   mogd:<fidid> -> \@devids (and TODO: invalidate when deletion is run!)
+# --------------------------------------------------------------------------
+
+my $cache;
+
+sub init {
+    my $self = shift;
+    $self->SUPER::init;
+
+    my @servers = grep(!/^$/, grep(s/^\s*|\s*$//g, split(',', $self->{servers})));
+    @servers = ('127.0.0.1:11211') unless @servers;
+
+    $cache = Cache::Memcached->new;
+    $cache->set_servers(\@servers);
+    return $self;
+}
+
+sub refresh {
+    my $self = shift;
+
+    # backwards compatibility with previous cache implementation
+    my @servers = grep(!/^$/, grep(s/^\s*|\s*$//g, split(',', MogileFS::Config->server_setting_cached("memcache_servers"))));
+    return undef unless @servers;
+
+    $cache->set_servers(\@servers);
+    return $self;
+}
+
+sub set {
+    my ($self, $key, $value) = @_;
+    return $cache->set($self->hash($key), $value, $self->{ttl});
+}
+
+sub get {
+    my ($self, $key) = @_;
+    return $cache->get($self->hash($key));
+}
+
+sub delete {
+    my ($self, $key) = @_;
+    return $cache->delete($self->hash($key));
+}
+
+1;
+
+__END__
+
+=head1 NAME
+
+MogileFS::Cache::Memcache - Memcache path cache for MogileFS
+
+=head1 SEE ALSO
+
+L<MogileFS::Cache>
+
+

lib/MogileFS/Cache/Redis.pm

@@ -0,0 +1,83 @@
+package MogileFS::Cache::Redis;
+use strict;
+use base 'MogileFS::Cache';
+use Redis;
+
+# --------------------------------------------------------------------------
+# Redis mappings:
+#   mogf:<dmid>:<dkey> -> fidid
+#   mogd:<fidid> -> \@devids (Redis set)
+# --------------------------------------------------------------------------
+
+my $cache;
+
+sub init {
+    my $self = shift;
+    $self->SUPER::init;
+
+    my $server = $self->{servers};
+    $server =~ s/^\s+//;
+    $server =~ s/\s+$//;
+    $server ||= '127.0.0.1:6379';
+
+    if ($server =~ m!^/!) {
+        $cache = Redis->new(sock => $server, encoding => undef, reconnect => 60);
+    } else {
+        $cache = Redis->new(server => $server, encoding => undef, reconnect => 60);
+    }
+    return $self;
+}
+
+sub refresh {
+    my $self = shift;
+    return $self;
+}
+
+sub set {
+    # We use simple keys and sets since Redis only supports
+    # expiry based on key rather than individual elements of
+    # an advanced data structure like hashes.
+    my ($self, $key, $value) = @_;
+    my $hash = $self->hash($key);
+    my $response;
+    if ($key->{type} eq 'devid') {
+        # requires Redis >= 2.4
+        $response = $cache->sadd($hash, @$value);
+    } else {
+        $response = $cache->set($hash => $value);
+    }
+    # Unlike memcache, setting ttl to 0 will clear the key
+    if ($self->{ttl}) {
+        $cache->expire($hash, $self->{ttl});
+    }
+    return $response;
+}
+
+sub get {
+    my ($self, $key) = @_;
+    my $hash = $self->hash($key);
+    if ($key->{type} eq 'devid') {
+        my $members = $cache->smembers($hash);
+        return @$members ? $members : undef;
+    }
+    return $cache->get($hash);
+}
+
+sub delete {
+    my ($self, $key) = @_;
+    return $cache->del($self->hash($key));
+}
+
+1;
+
+__END__
+
+=head1 NAME
+
+MogileFS::Cache::Redis - Redis path cache for MogileFS
+
+=head1 SEE ALSO
+
+L<MogileFS::Cache>
+
+

lib/MogileFS/Config.pm

@@ -105,6 +105,9 @@ sub load_config {
                              'repl_use_get_port=i' => \$cmdline{repl_use_get_port},
                              'local_network=s' => \$cmdline{local_network},
                              'mogstored_stream_port' => \$cmdline{mogstored_stream_port},
+                             'cache_type' => \$cmdline{cache_type},
+                             'cache_servers' => \$cmdline{cache_servers},
+                             'cache_ttl' => \$cmdline{cache_ttl},
                              );
 
     # warn of old/deprecated options
@@ -166,6 +169,10 @@ sub load_config {
     $node_timeout   = choose_value( 'node_timeout', 2 );
     $conn_timeout   = choose_value( 'conn_timeout', 2 );
 
+    choose_value( 'cache_type', "" );
+    choose_value( 'cache_servers', "" );
+    choose_value( 'cache_ttl', 3600 );
+
     choose_value( 'rebalance_ignore_missing', 0 );
     $repl_use_get_port = choose_value( 'repl_use_get_port', 0 );
     $local_network  = choose_value( 'local_network', '' );
@@ -289,27 +296,6 @@ sub server_setting_cached {
     return $server_settings{$key};
 }
 
-my $memc;
-my $last_memc_server_fetch = 0;
-my $have_memc_module = eval "use Cache::Memcached; 1;";
-sub memcache_client {
-    return undef unless $have_memc_module;
-
-    # only reload the server list every 30 seconds
-    my $now = time();
-    return $memc if $last_memc_server_fetch > $now - 30;
-
-    my @servers = grep(/:\d+$/, split(/\s*,\s*/, MogileFS::Config->server_setting_cached("memcache_servers") || ""));
-    $last_memc_server_fetch = $now;
-
-    return ($memc = undef) unless @servers;
-
-    $memc ||= Cache::Memcached->new;
-    $memc->set_servers(\@servers);
-
-    return $memc;
-}
-
 my $cache_hostname;
 sub hostname {
     return $cache_hostname ||= Sys::Hostname::hostname();

lib/MogileFS/DevFID.pm

@@ -140,8 +140,8 @@ sub add_to_db {
 
     my $sto = Mgd::get_store();
     if ($sto->add_fidid_to_devid($self->{fidid}, $self->{devid})) {
-        if (my $memc = MogileFS::Config->memcache_client) {
-            $memc->delete("mogdevids:$self->{fidid}");
+        if (my $cache = Mgd::get_cache()) {
+            $cache->delete({ type => 'devid', fid => $self->{fidid} });
         }
         return $self->fid->update_devcount(no_lock => $no_lock);
     } else {

lib/MogileFS/FID.pm

@@ -160,13 +160,13 @@ sub mark_unreachable {
 sub delete {
     my $fid = shift;
     my $sto = Mgd::get_store();
-    my $memc = MogileFS::Config->memcache_client;
-    if ($memc) {
+    my $cache = Mgd::get_cache();
+    if ($cache) {
         $fid->_tryload;
     }
     $sto->delete_fidid($fid->id);
-    if ($memc && $fid->{_loaded}) {
-        $memc->delete("mogfid:$fid->{dmid}:$fid->{dkey}");
+    if ($cache && $fid->{_loaded}) {
+        $cache->delete({ type => 'fid', domain => $fid->{dmid}, key => $fid->{dkey} });
     }
 }
 

lib/MogileFS/Server.pm

@@ -58,6 +58,7 @@ use MogileFS::FID;
 use MogileFS::DevFID;
 
 use MogileFS::Store;
+use MogileFS::Cache;
 
 use MogileFS::ReplicationPolicy::MultipleHosts;
 
@@ -246,6 +247,24 @@ sub set_store {
     $store_pid = $$;
 }
 
+# cache abstraction layer
+my ($cache, $cache_pid, $last_cache_refresh);
+sub get_cache {
+    return undef if MogileFS->config('cache_type') eq 'none';
+    my $now = time();
+    if (!$last_cache_refresh && $cache && $cache_pid == $$) {
+        return $cache;
+    } elsif ($last_cache_refresh && $last_cache_refresh > ($now - 30)) {
+        return $cache;
+    }
+    # backwards compatibility with previous cache implementation
+    if (MogileFS->config('cache_type') eq '') {
+        $last_cache_refresh = $now;
+    }
+    $cache_pid = $$;
+    return $cache = $cache ? $cache->refresh() : MogileFS::Cache->new;
+}
+
 sub domain_factory {
     return MogileFS::Factory::Domain->get_factory;
 }