view release on metacpan or  search on metacpan

lib/Cache/FastMmap.pm  view on Meta::CPAN

      $Self->{deserialize} = $serializer->[1];
    } elsif ($serializer eq 'storable') {
      eval "require Storable;"
        || die "Could not load serialization package: Storable : $@";
      $Self->{serialize}   = Storable->can("freeze");
      $Self->{deserialize} = Storable->can("thaw");
    } elsif ($serializer eq 'sereal') {
      eval "require Sereal::Encoder; require Sereal::Decoder;"
        || die "Could not load serialization package: Sereal : $@";
      my $SerealEnc = Sereal::Encoder->new();
      my $SerealDec = Sereal::Decoder->new();
      $Self->{serialize} = sub { $SerealEnc->encode(@_); };
      $Self->{deserialize} = sub { $SerealDec->decode(@_); };
    } elsif ($serializer eq 'json') {
      eval "require JSON;"
        || die "Could not load serialization package: JSON : $@";
      my $JSON = JSON->new->utf8->allow_nonref;
      $Self->{serialize}   = sub { $JSON->encode(${$_[0]}); };
      $Self->{deserialize} = sub { \$JSON->decode($_[0]); };
    } else {
      die "Unrecognized value >$serializer< for `serializer` parameter";
    }
  }

  # Compress stored values?
  my $compressor = $Args{compressor};
  $compressor = ($Args{compress} ? 'zlib' : 0) if !defined $compressor;

  my %known_compressors = (
    zlib   => 'Compress::Zlib',
    lz4    => 'Compress::LZ4',
    snappy => 'Compress::Snappy',
  );

  if ( $compressor ) {
    if (ref $compressor eq 'ARRAY') {
      $Self->{compress}   = $compressor->[0];
      $Self->{uncompress} = $compressor->[1];
    } elsif (my $compressor_module = $known_compressors{$compressor}) {
      eval "require $compressor_module;"
        || die "Could not load compression package: $compressor_module : $@";

      # LZ4 and Snappy use same API
      if ($compressor_module eq 'Compress::LZ4' || $compressor_module eq 'Compress::Snappy') {
        $Self->{compress}   = $compressor_module->can("compress");
        $Self->{uncompress} = $compressor_module->can("uncompress");
      } elsif ($compressor_module eq 'Compress::Zlib') {
        $Self->{compress}   = $compressor_module->can("memGzip");
        # (gunzip from tmp var: https://rt.cpan.org/Ticket/Display.html?id=72945)
        my $uncompress = $compressor_module->can("memGunzip");
        $Self->{uncompress} = sub { &$uncompress(my $Tmp = shift) };
      }
    } else {
      die "Unrecognized value >$compressor< for `compressor` parameter";
    }
  }

  # If using empty_on_exit, need to track used caches
  my $empty_on_exit = $Self->{empty_on_exit} = int($Args{empty_on_exit} || 0);

  # Need Scalar::Util::weaken to track open caches
  if ($empty_on_exit) {
    eval "use Scalar::Util qw(weaken); 1;"
      || die "Could not load Scalar::Util module: $@";
  }

  # Work out expiry time in seconds
  my $expire_time = $Self->{expire_time} = parse_expire_time($Args{expire_time});

  # Function rounds to the nearest power of 2
  sub RoundPow2 { return int(2 ** int(log($_[0])/log(2)) + 0.1); }

  # Work out cache size
  my ($cache_size, $num_pages, $page_size);

  my %Sizes = (k => 1024, m => 1024*1024);
  if ($cache_size = $Args{cache_size}) {
    $cache_size *= $Sizes{lc($1)} if $cache_size =~ s/([km])$//i;

    if ($num_pages = $Args{num_pages}) {
      $page_size = RoundPow2($cache_size / $num_pages);
      $page_size = 4096 if $page_size < 4096;

    } else {
      $page_size = $Args{page_size} || 65536;
      $page_size *= $Sizes{lc($1)} if $page_size =~ s/([km])$//i;
      $page_size = 4096 if $page_size < 4096;

      # Increase num_pages till we exceed 
      $num_pages = 89;
      if ($num_pages * $page_size <= $cache_size) {
        while ($num_pages * $page_size <= $cache_size) {
          $num_pages = $num_pages * 2 + 1;
        }
      } else {
        while ($num_pages * $page_size > $cache_size) {
          $num_pages = int(($num_pages-1) / 2);
        }
        $num_pages = $num_pages * 2 + 1;
      }

    }

  } else {
    ($num_pages, $page_size) = @Args{qw(num_pages page_size)};
    $num_pages ||= 89;
    $page_size ||= 65536;
    $page_size *= $Sizes{lc($1)} if $page_size =~ s/([km])$//i;
    $page_size = RoundPow2($page_size);
  }

  $cache_size = $num_pages * $page_size;
  @$Self{qw(cache_size num_pages page_size)}
    = ($cache_size, $num_pages, $page_size);

  # Number of slots to start in each page
  my $start_slots = int($Args{start_slots} || 0) || 89;

  # Save read through/write back/write through details
  my $write_back = ($Args{write_action} || 'write_through') eq 'write_back';
  @$Self{qw(context read_cb write_cb delete_cb)}
    = @Args{qw(context read_cb write_cb delete_cb)};
  @$Self{qw(cache_not_found allow_recursive write_back)}
    = (@Args{qw(cache_not_found allow_recursive)}, $write_back);
  @$Self{qw(unlink_on_exit enable_stats)}
    = (@Args{qw(unlink_on_exit)}, $enable_stats);

  # Save pid
  $Self->{pid} = $$;

  # Initialise C cache code
  my $Cache = fc_new();

  $Self->{Cache} = $Cache;

  # Setup cache parameters
  fc_set_param($Cache, 'init_file', $init_file);
  fc_set_param($Cache, 'test_file', $test_file);
  fc_set_param($Cache, 'page_size', $page_size);
  fc_set_param($Cache, 'num_pages', $num_pages);
  fc_set_param($Cache, 'expire_time', $expire_time);
  fc_set_param($Cache, 'share_file', $share_file);
  fc_set_param($Cache, 'permissions', $permissions) if defined $permissions;
  fc_set_param($Cache, 'start_slots', $start_slots);
  fc_set_param($Cache, 'catch_deadlocks', $catch_deadlocks);
  fc_set_param($Cache, 'enable_stats', $enable_stats);

  # And initialise it
  fc_init($Cache);

  # Track cache if need to empty on exit
  weaken($LiveCaches{"$Self"} = $Self)
    if $empty_on_exit;

  # All done, return PERL hash ref as class
  return $Self;
}

=item I<get($Key, [ \%Options ])>

Search cache for given Key. Returns undef if not found. If
I<read_cb> specified and not found, calls the callback to try
and find the value for the key, and if found (or 'cache_not_found'
is set), stores it into the cache and returns the found value.

I<%Options> is optional, and is used by get_and_set() to control
the locking behaviour. For now, you should probably ignore it
unless you read the code to understand how it works

=cut
sub get {
  my ($Self, $Cache) = ($_[0], $_[0]->{Cache});

  my $SkipUnlock = $_[2] && $_[2]->{skip_unlock};
  my $Locked = 0;

  # Hash value, lock page, read result
  my ($HashPage, $HashSlot) = fc_hash($Cache, $_[1]);
  fc_lock($Cache, $HashPage);
  $Locked = 1;

  my ($Val, $Flags, $Found, $ExpireOn);
  my $Err;
  eval {
    ($Val, $Flags, $Found, $ExpireOn) = fc_read($Cache, $HashSlot, $_[1]);

    # Value not found, check underlying data store
    if (!$Found && (my $read_cb = $Self->{read_cb})) {

      # Callback to read from underlying data store
      # (unlock page first if we allow recursive calls)
      if ($Self->{allow_recursive}) {
        fc_unlock($Cache);
        $Val = eval { $read_cb->($Self->{context}, $_[1]); };
        my $CbErr = $@;
        fc_lock($Cache, $HashPage);
        die $CbErr if $CbErr;
      } else {
        $Val = $read_cb->($Self->{context}, $_[1]);
      }

      # If we found it, or want to cache not-found, store back into our cache
      if (defined $Val || $Self->{cache_not_found}) {

        # Are we doing writeback's? If so, need to mark as dirty in cache
        my $write_back = $Self->{write_back};

        $Val = $Self->{serialize}(\$Val) if $Self->{serialize};
        $Val = $Self->{compress}($Val) if $Self->{compress};

        # Get key/value len (we've got 'use bytes'), and do expunge check to
        #  create space if needed