lib/DBM/Deep/Engine.pm

   1 package DBM::Deep::Engine;
   2
   3 use 5.6.0;
   4
   5 use strict;
   6
   7 our $VERSION = q(0.99_03);
   8
   9 use Scalar::Util ();
  10
  11 # File-wide notes:
  12 # * Every method in here assumes that the storage has been appropriately
  13 #   safeguarded. This can be anything from flock() to some sort of manual
  14 #   mutex. But, it's the caller's responsability to make sure that this has
  15 #   been done.
  16
  17 # Setup file and tag signatures.  These should never change.
  18 sub SIG_FILE     () { 'DPDB' }
  19 sub SIG_HEADER   () { 'h'    }
  20 sub SIG_INTERNAL () { 'i'    }
  21 sub SIG_HASH     () { 'H'    }
  22 sub SIG_ARRAY    () { 'A'    }
  23 sub SIG_NULL     () { 'N'    }
  24 sub SIG_DATA     () { 'D'    }
  25 sub SIG_INDEX    () { 'I'    }
  26 sub SIG_BLIST    () { 'B'    }
  27 sub SIG_FREE     () { 'F'    }
  28 sub SIG_KEYS     () { 'K'    }
  29 sub SIG_SIZE     () {  1     }
  30 sub STALE_SIZE   () {  1     }
  31
  32 # Please refer to the pack() documentation for further information
  33 my %StP = (
  34     1 => 'C', # Unsigned char value (no order specified, presumably ASCII)
  35     2 => 'n', # Unsigned short in "network" (big-endian) order
  36     4 => 'N', # Unsigned long in "network" (big-endian) order
  37     8 => 'Q', # Usigned quad (no order specified, presumably machine-dependent)
  38 );
  39
  40 ################################################################################
  41
  42 sub new {
  43     my $class = shift;
  44     my ($args) = @_;
  45
  46     my $self = bless {
  47         byte_size   => 4,
  48
  49         digest      => undef,
  50         hash_size   => 16,  # In bytes
  51         hash_chars  => 256, # Number of chars the algorithm uses per byte
  52         max_buckets => 16,
  53         num_txns    => 2,   # HEAD plus 1 additional transaction for importing
  54         trans_id    => 0,   # Default to the HEAD
  55
  56         entries => {}, # This is the list of entries for transactions
  57         storage => undef,
  58     }, $class;
  59
  60     if ( defined $args->{pack_size} ) {
  61         if ( lc $args->{pack_size} eq 'small' ) {
  62             $args->{byte_size} = 2;
  63         }
  64         elsif ( lc $args->{pack_size} eq 'medium' ) {
  65             $args->{byte_size} = 4;
  66         }
  67         elsif ( lc $args->{pack_size} eq 'large' ) {
  68             $args->{byte_size} = 8;
  69         }
  70         else {
  71             DBM::Deep->_throw_error( "Unknown pack_size value: '$args->{pack_size}'" );
  72         }
  73     }
  74
  75     # Grab the parameters we want to use
  76     foreach my $param ( keys %$self ) {
  77         next unless exists $args->{$param};
  78         $self->{$param} = $args->{$param};
  79     }
  80
  81     ##
  82     # Number of buckets per blist before another level of indexing is
  83     # done. Increase this value for slightly greater speed, but larger database
  84     # files. DO NOT decrease this value below 16, due to risk of recursive
  85     # reindex overrun.
  86     ##
  87     if ( $self->{max_buckets} < 16 ) {
  88         warn "Floor of max_buckets is 16. Setting it to 16 from '$self->{max_buckets}'\n";
  89         $self->{max_buckets} = 16;
  90     }
  91
  92     if ( !$self->{digest} ) {
  93         require Digest::MD5;
  94         $self->{digest} = \&Digest::MD5::md5;
  95     }
  96
  97     return $self;
  98 }
  99
 100 ################################################################################
 101
 102 sub read_value {
 103     my $self = shift;
 104     my ($obj, $key) = @_;
 105
 106     # This will be a Reference sector
 107     my $sector = $self->_load_sector( $obj->_base_offset )
 108         or return;
 109
 110     if ( $sector->staleness != $obj->_staleness ) {
 111         return;
 112     }
 113
 114     my $key_md5 = $self->_apply_digest( $key );
 115
 116     my $value_sector = $sector->get_data_for({
 117         key_md5    => $key_md5,
 118         allow_head => 1,
 119     });
 120
 121     unless ( $value_sector ) {
 122         $value_sector = DBM::Deep::Engine::Sector::Null->new({
 123             engine => $self,
 124             data   => undef,
 125         });
 126
 127         $sector->write_data({
 128             key_md5 => $key_md5,
 129             key     => $key,
 130             value   => $value_sector,
 131         });
 132     }
 133
 134     return $value_sector->data;
 135 }
 136
 137 sub get_classname {
 138     my $self = shift;
 139     my ($obj) = @_;
 140
 141     # This will be a Reference sector
 142     my $sector = $self->_load_sector( $obj->_base_offset )
 143         or DBM::Deep->_throw_error( "How did get_classname fail (no sector for '$obj')?!" );
 144
 145     if ( $sector->staleness != $obj->_staleness ) {
 146         return;
 147     }
 148
 149     return $sector->get_classname;
 150 }
 151
 152 sub key_exists {
 153     my $self = shift;
 154     my ($obj, $key) = @_;
 155
 156     # This will be a Reference sector
 157     my $sector = $self->_load_sector( $obj->_base_offset )
 158         or return '';
 159
 160     if ( $sector->staleness != $obj->_staleness ) {
 161         return '';
 162     }
 163
 164     my $data = $sector->get_data_for({
 165         key_md5    => $self->_apply_digest( $key ),
 166         allow_head => 1,
 167     });
 168
 169     # exists() returns 1 or '' for true/false.
 170     return $data ? 1 : '';
 171 }
 172
 173 sub delete_key {
 174     my $self = shift;
 175     my ($obj, $key) = @_;
 176
 177     my $sector = $self->_load_sector( $obj->_base_offset )
 178         or return;
 179
 180     if ( $sector->staleness != $obj->_staleness ) {
 181         return;
 182     }
 183
 184     return $sector->delete_key({
 185         key_md5    => $self->_apply_digest( $key ),
 186         allow_head => 0,
 187     });
 188 }
 189
 190 sub write_value {
 191     my $self = shift;
 192     my ($obj, $key, $value) = @_;
 193
 194     my $r = Scalar::Util::reftype( $value ) || '';
 195     {
 196         last if $r eq '';
 197         last if $r eq 'HASH';
 198         last if $r eq 'ARRAY';
 199
 200         DBM::Deep->_throw_error(
 201             "Storage of references of type '$r' is not supported."
 202         );
 203     }
 204
 205     my ($class, $type);
 206     if ( !defined $value ) {
 207         $class = 'DBM::Deep::Engine::Sector::Null';
 208     }
 209     elsif ( $r eq 'ARRAY' || $r eq 'HASH' ) {
 210         if ( $r eq 'ARRAY' && tied(@$value) ) {
 211             DBM::Deep->_throw_error( "Cannot store something that is tied." );
 212         }
 213         if ( $r eq 'HASH' && tied(%$value) ) {
 214             DBM::Deep->_throw_error( "Cannot store something that is tied." );
 215         }
 216         $class = 'DBM::Deep::Engine::Sector::Reference';
 217         $type = substr( $r, 0, 1 );
 218     }
 219     else {
 220         $class = 'DBM::Deep::Engine::Sector::Scalar';
 221     }
 222
 223     # This will be a Reference sector
 224     my $sector = $self->_load_sector( $obj->_base_offset )
 225         or DBM::Deep->_throw_error( "Cannot write to a deleted spot in DBM::Deep." );
 226
 227     if ( $sector->staleness != $obj->_staleness ) {
 228         DBM::Deep->_throw_error( "Cannot write to a deleted spot in DBM::Deep.n" );
 229     }
 230
 231     # Create this after loading the reference sector in case something bad happens.
 232     # This way, we won't allocate value sector(s) needlessly.
 233     my $value_sector = $class->new({
 234         engine => $self,
 235         data   => $value,
 236         type   => $type,
 237     });
 238
 239     $sector->write_data({
 240         key     => $key,
 241         key_md5 => $self->_apply_digest( $key ),
 242         value   => $value_sector,
 243     });
 244
 245     # This code is to make sure we write all the values in the $value to the disk
 246     # and to make sure all changes to $value after the assignment are reflected
 247     # on disk. This may be counter-intuitive at first, but it is correct dwimmery.
 248     #   NOTE - simply tying $value won't perform a STORE on each value. Hence, the
 249     # copy to a temp value.
 250     if ( $r eq 'ARRAY' ) {
 251         my @temp = @$value;
 252         tie @$value, 'DBM::Deep', {
 253             base_offset => $value_sector->offset,
 254             staleness   => $value_sector->staleness,
 255             storage     => $self->storage,
 256             engine      => $self,
 257         };
 258         @$value = @temp;
 259         bless $value, 'DBM::Deep::Array' unless Scalar::Util::blessed( $value );
 260     }
 261     elsif ( $r eq 'HASH' ) {
 262         my %temp = %$value;
 263         tie %$value, 'DBM::Deep', {
 264             base_offset => $value_sector->offset,
 265             staleness   => $value_sector->staleness,
 266             storage     => $self->storage,
 267             engine      => $self,
 268         };
 269
 270         %$value = %temp;
 271         bless $value, 'DBM::Deep::Hash' unless Scalar::Util::blessed( $value );
 272     }
 273
 274     return 1;
 275 }
 276
 277 # XXX Add staleness here
 278 sub get_next_key {
 279     my $self = shift;
 280     my ($obj, $prev_key) = @_;
 281
 282     # XXX Need to add logic about resetting the iterator if any key in the reference has changed
 283     unless ( $prev_key ) {
 284         $obj->{iterator} = DBM::Deep::Iterator->new({
 285             base_offset => $obj->_base_offset,
 286             engine      => $self,
 287         });
 288     }
 289
 290     return $obj->{iterator}->get_next_key( $obj );
 291 }
 292
 293 ################################################################################
 294
 295 sub setup_fh {
 296     my $self = shift;
 297     my ($obj) = @_;
 298
 299     # We're opening the file.
 300     unless ( $obj->_base_offset ) {
 301         my $bytes_read = $self->_read_file_header;
 302
 303         # Creating a new file
 304         unless ( $bytes_read ) {
 305             $self->_write_file_header;
 306
 307             # 1) Create Array/Hash entry
 308             my $initial_reference = DBM::Deep::Engine::Sector::Reference->new({
 309                 engine => $self,
 310                 type   => $obj->_type,
 311             });
 312             $obj->{base_offset} = $initial_reference->offset;
 313             $obj->{staleness} = $initial_reference->staleness;
 314
 315             $self->storage->flush;
 316         }
 317         # Reading from an existing file
 318         else {
 319             $obj->{base_offset} = $bytes_read;
 320             my $initial_reference = DBM::Deep::Engine::Sector::Reference->new({
 321                 engine => $self,
 322                 offset => $obj->_base_offset,
 323             });
 324             unless ( $initial_reference ) {
 325                 DBM::Deep->_throw_error("Corrupted file, no master index record");
 326             }
 327
 328             unless ($obj->_type eq $initial_reference->type) {
 329                 DBM::Deep->_throw_error("File type mismatch");
 330             }
 331
 332             $obj->{staleness} = $initial_reference->staleness;
 333         }
 334     }
 335
 336     return 1;
 337 }
 338
 339 sub begin_work {
 340     my $self = shift;
 341     my ($obj) = @_;
 342
 343     if ( $self->trans_id ) {
 344         DBM::Deep->_throw_error( "Cannot begin_work within an active transaction" );
 345     }
 346
 347     my @slots = $self->read_txn_slots;
 348     for my $i ( 1 .. @slots ) {
 349         next if $slots[$i];
 350         $slots[$i] = 1;
 351         $self->set_trans_id( $i );
 352         last;
 353     }
 354     $self->write_txn_slots( @slots );
 355
 356     if ( !$self->trans_id ) {
 357         DBM::Deep->_throw_error( "Cannot begin_work - no available transactions" );
 358     }
 359
 360     return;
 361 }
 362
 363 sub rollback {
 364     my $self = shift;
 365     my ($obj) = @_;
 366
 367     if ( !$self->trans_id ) {
 368         DBM::Deep->_throw_error( "Cannot rollback without an active transaction" );
 369     }
 370
 371     # Each entry is the file location for a bucket that has a modification for
 372     # this transaction. The entries need to be expunged.
 373     foreach my $entry (@{ $self->get_entries } ) {
 374         # Remove the entry here
 375         my $read_loc = $entry
 376           + $self->hash_size
 377           + $self->byte_size
 378           + $self->trans_id * ( $self->byte_size + 4 );
 379
 380         my $data_loc = $self->storage->read_at( $read_loc, $self->byte_size );
 381         $data_loc = unpack( $StP{$self->byte_size}, $data_loc );
 382         $self->storage->print_at( $read_loc, pack( $StP{$self->byte_size}, 0 ) );
 383
 384         if ( $data_loc > 1 ) {
 385             $self->_load_sector( $data_loc )->free;
 386         }
 387     }
 388
 389     $self->clear_entries;
 390
 391     my @slots = $self->read_txn_slots;
 392     $slots[$self->trans_id] = 0;
 393     $self->write_txn_slots( @slots );
 394     $self->inc_txn_staleness_counter( $self->trans_id );
 395     $self->set_trans_id( 0 );
 396
 397     return 1;
 398 }
 399
 400 sub commit {
 401     my $self = shift;
 402     my ($obj) = @_;
 403
 404     if ( !$self->trans_id ) {
 405         DBM::Deep->_throw_error( "Cannot commit without an active transaction" );
 406     }
 407
 408     foreach my $entry (@{ $self->get_entries } ) {
 409         # Overwrite the entry in head with the entry in trans_id
 410         my $base = $entry
 411           + $self->hash_size
 412           + $self->byte_size;
 413
 414         my $head_loc = $self->storage->read_at( $base, $self->byte_size );
 415         $head_loc = unpack( $StP{$self->byte_size}, $head_loc );
 416         my $trans_loc = $self->storage->read_at(
 417             $base + $self->trans_id * ( $self->byte_size + 4 ), $self->byte_size,
 418         );
 419
 420         $self->storage->print_at( $base, $trans_loc );
 421         $self->storage->print_at(
 422             $base + $self->trans_id * ( $self->byte_size + 4 ),
 423             pack( $StP{$self->byte_size} . ' N', (0) x 2 ),
 424         );
 425
 426         if ( $head_loc > 1 ) {
 427             $self->_load_sector( $head_loc )->free;
 428         }
 429     }
 430
 431     $self->clear_entries;
 432
 433     my @slots = $self->read_txn_slots;
 434     $slots[$self->trans_id] = 0;
 435     $self->write_txn_slots( @slots );
 436     $self->inc_txn_staleness_counter( $self->trans_id );
 437     $self->set_trans_id( 0 );
 438
 439     return 1;
 440 }
 441
 442 sub read_txn_slots {
 443     my $self = shift;
 444     return split '', unpack( 'b32',
 445         $self->storage->read_at(
 446             $self->trans_loc, 4,
 447         )
 448     );
 449 }
 450
 451 sub write_txn_slots {
 452     my $self = shift;
 453     $self->storage->print_at( $self->trans_loc,
 454         pack( 'b32', join('', @_) ),
 455     );
 456 }
 457
 458 sub get_running_txn_ids {
 459     my $self = shift;
 460     my @transactions = $self->read_txn_slots;
 461     my @trans_ids = grep { $transactions[$_] } 0 .. $#transactions;
 462 }
 463
 464 sub get_txn_staleness_counter {
 465     my $self = shift;
 466     my ($trans_id) = @_;
 467
 468     # Hardcode staleness of 0 for the HEAD
 469     return 0 unless $trans_id;
 470
 471     my $x = unpack( 'N',
 472         $self->storage->read_at(
 473             $self->trans_loc + 4 * $trans_id,
 474             4,
 475         )
 476     );
 477     return $x;
 478 }
 479
 480 sub inc_txn_staleness_counter {
 481     my $self = shift;
 482     my ($trans_id) = @_;
 483
 484     # Hardcode staleness of 0 for the HEAD
 485     return unless $trans_id;
 486
 487     $self->storage->print_at(
 488         $self->trans_loc + 4 * $trans_id,
 489         pack( 'N', $self->get_txn_staleness_counter( $trans_id ) + 1 ),
 490     );
 491 }
 492
 493 sub get_entries {
 494     my $self = shift;
 495     return [ keys %{ $self->{entries}{$self->trans_id} ||= {} } ];
 496 }
 497
 498 sub add_entry {
 499     my $self = shift;
 500     my ($trans_id, $loc) = @_;
 501
 502     $self->{entries}{$trans_id} ||= {};
 503     $self->{entries}{$trans_id}{$loc} = undef;
 504 }
 505
 506 # If the buckets are being relocated because of a reindexing, the entries
 507 # mechanism needs to be made aware of it.
 508 sub reindex_entry {
 509     my $self = shift;
 510     my ($old_loc, $new_loc) = @_;
 511
 512     TRANS:
 513     while ( my ($trans_id, $locs) = each %{ $self->{entries} } ) {
 514         foreach my $orig_loc ( keys %{ $locs } ) {
 515             if ( $orig_loc == $old_loc ) {
 516                 delete $locs->{orig_loc};
 517                 $locs->{$new_loc} = undef;
 518                 next TRANS;
 519             }
 520         }
 521     }
 522 }
 523
 524 sub clear_entries {
 525     my $self = shift;
 526     delete $self->{entries}{$self->trans_id};
 527 }
 528
 529 ################################################################################
 530
 531 {
 532     my $header_fixed = length( SIG_FILE ) + 1 + 4 + 4;
 533
 534     sub _write_file_header {
 535         my $self = shift;
 536
 537         my $header_var = 1 + 1 + 1 + 4 + 4 * $self->num_txns + 3 * $self->byte_size;
 538
 539         my $loc = $self->storage->request_space( $header_fixed + $header_var );
 540
 541         $self->storage->print_at( $loc,
 542             SIG_FILE,
 543             SIG_HEADER,
 544             pack('N', 1),           # header version - at this point, we're at 9 bytes
 545             pack('N', $header_var), # header size
 546             # --- Above is $header_fixed. Below is $header_var
 547             pack('C', $self->byte_size),
 548             pack('C', $self->max_buckets),
 549             pack('C', $self->num_txns),
 550             pack('N', 0 ),                   # Transaction activeness bitfield
 551             pack('N' . $self->num_txns, 0 x $self->num_txns ), # Transaction staleness counters
 552             pack($StP{$self->byte_size}, 0), # Start of free chain (blist size)
 553             pack($StP{$self->byte_size}, 0), # Start of free chain (data size)
 554             pack($StP{$self->byte_size}, 0), # Start of free chain (index size)
 555         );
 556
 557         $self->set_trans_loc( $header_fixed + 3 );
 558         $self->set_chains_loc( $header_fixed + 3 + 4 + 4 * $self->num_txns );
 559
 560         return;
 561     }
 562
 563     sub _read_file_header {
 564         my $self = shift;
 565
 566         my $buffer = $self->storage->read_at( 0, $header_fixed );
 567         return unless length($buffer);
 568
 569         my ($file_signature, $sig_header, $header_version, $size) = unpack(
 570             'A4 A N N', $buffer
 571         );
 572
 573         unless ( $file_signature eq SIG_FILE ) {
 574             $self->storage->close;
 575             DBM::Deep->_throw_error( "Signature not found -- file is not a Deep DB" );
 576         }
 577
 578         unless ( $sig_header eq SIG_HEADER ) {
 579             $self->storage->close;
 580             DBM::Deep->_throw_error( "Old file version found." );
 581         }
 582
 583         my $buffer2 = $self->storage->read_at( undef, $size );
 584         my @values = unpack( 'C C C', $buffer2 );
 585
 586         if ( @values != 3 || grep { !defined } @values ) {
 587             $self->storage->close;
 588             DBM::Deep->_throw_error("Corrupted file - bad header");
 589         }
 590
 591         $self->set_trans_loc( $header_fixed + scalar(@values) );
 592         $self->set_chains_loc( $header_fixed + scalar(@values) + 4 + 4 * $self->num_txns );
 593
 594         #XXX Add warnings if values weren't set right
 595         @{$self}{qw(byte_size max_buckets num_txns)} = @values;
 596
 597         my $header_var = scalar(@values) + 4 + 4 * $self->num_txns + 3 * $self->byte_size;
 598         unless ( $size == $header_var ) {
 599             $self->storage->close;
 600             DBM::Deep->_throw_error( "Unexpected size found ($size <-> $header_var)." );
 601         }
 602
 603         return length($buffer) + length($buffer2);
 604     }
 605 }
 606
 607 sub _load_sector {
 608     my $self = shift;
 609     my ($offset) = @_;
 610
 611     # Add a catch for offset of 0 or 1
 612     return if $offset <= 1;
 613
 614     my $type = $self->storage->read_at( $offset, 1 );
 615     return if $type eq chr(0);
 616
 617     if ( $type eq $self->SIG_ARRAY || $type eq $self->SIG_HASH ) {
 618         return DBM::Deep::Engine::Sector::Reference->new({
 619             engine => $self,
 620             type   => $type,
 621             offset => $offset,
 622         });
 623     }
 624     # XXX Don't we need key_md5 here?
 625     elsif ( $type eq $self->SIG_BLIST ) {
 626         return DBM::Deep::Engine::Sector::BucketList->new({
 627             engine => $self,
 628             type   => $type,
 629             offset => $offset,
 630         });
 631     }
 632     elsif ( $type eq $self->SIG_INDEX ) {
 633         return DBM::Deep::Engine::Sector::Index->new({
 634             engine => $self,
 635             type   => $type,
 636             offset => $offset,
 637         });
 638     }
 639     elsif ( $type eq $self->SIG_NULL ) {
 640         return DBM::Deep::Engine::Sector::Null->new({
 641             engine => $self,
 642             type   => $type,
 643             offset => $offset,
 644         });
 645     }
 646     elsif ( $type eq $self->SIG_DATA ) {
 647         return DBM::Deep::Engine::Sector::Scalar->new({
 648             engine => $self,
 649             type   => $type,
 650             offset => $offset,
 651         });
 652     }
 653     # This was deleted from under us, so just return and let the caller figure it out.
 654     elsif ( $type eq $self->SIG_FREE ) {
 655         return;
 656     }
 657
 658     DBM::Deep->_throw_error( "'$offset': Don't know what to do with type '$type'" );
 659 }
 660
 661 sub _apply_digest {
 662     my $self = shift;
 663     return $self->{digest}->(@_);
 664 }
 665
 666 sub _add_free_blist_sector { shift->_add_free_sector( 0, @_ ) }
 667 sub _add_free_data_sector { shift->_add_free_sector( 1, @_ ) }
 668 sub _add_free_index_sector { shift->_add_free_sector( 2, @_ ) }
 669
 670 sub _add_free_sector {
 671     my $self = shift;
 672     my ($multiple, $offset, $size) = @_;
 673
 674     my $chains_offset = $multiple * $self->byte_size;
 675
 676     my $storage = $self->storage;
 677
 678     # Increment staleness.
 679     # XXX Can this increment+modulo be done by "&= 0x1" ?
 680     my $staleness = unpack( $StP{STALE_SIZE()}, $storage->read_at( $offset + SIG_SIZE, STALE_SIZE ) );
 681     $staleness = ($staleness + 1 ) % ( 2 ** ( 8 * STALE_SIZE ) );
 682     $storage->print_at( $offset + SIG_SIZE, pack( $StP{STALE_SIZE()}, $staleness ) );
 683
 684     my $old_head = $storage->read_at( $self->chains_loc + $chains_offset, $self->byte_size );
 685
 686     $storage->print_at( $self->chains_loc + $chains_offset,
 687         pack( $StP{$self->byte_size}, $offset ),
 688     );
 689
 690     # Record the old head in the new sector after the signature and staleness counter
 691     $storage->print_at( $offset + SIG_SIZE + STALE_SIZE, $old_head );
 692 }
 693
 694 sub _request_blist_sector { shift->_request_sector( 0, @_ ) }
 695 sub _request_data_sector { shift->_request_sector( 1, @_ ) }
 696 sub _request_index_sector { shift->_request_sector( 2, @_ ) }
 697
 698 sub _request_sector {
 699     my $self = shift;
 700     my ($multiple, $size) = @_;
 701
 702     my $chains_offset = $multiple * $self->byte_size;
 703
 704     my $old_head = $self->storage->read_at( $self->chains_loc + $chains_offset, $self->byte_size );
 705     my $loc = unpack( $StP{$self->byte_size}, $old_head );
 706
 707     # We don't have any free sectors of the right size, so allocate a new one.
 708     unless ( $loc ) {
 709         my $offset = $self->storage->request_space( $size );
 710
 711         # Zero out the new sector. This also guarantees correct increases
 712         # in the filesize.
 713         $self->storage->print_at( $offset, chr(0) x $size );
 714
 715         return $offset;
 716     }
 717
 718     # Read the new head after the signature and the staleness counter
 719     my $new_head = $self->storage->read_at( $loc + SIG_SIZE + STALE_SIZE, $self->byte_size );
 720     $self->storage->print_at( $self->chains_loc + $chains_offset, $new_head );
 721
 722     return $loc;
 723 }
 724
 725 ################################################################################
 726
 727 sub storage     { $_[0]{storage} }
 728 sub byte_size   { $_[0]{byte_size} }
 729 sub hash_size   { $_[0]{hash_size} }
 730 sub hash_chars  { $_[0]{hash_chars} }
 731 sub num_txns    { $_[0]{num_txns} }
 732 sub max_buckets { $_[0]{max_buckets} }
 733 sub blank_md5   { chr(0) x $_[0]->hash_size }
 734
 735 sub trans_id     { $_[0]{trans_id} }
 736 sub set_trans_id { $_[0]{trans_id} = $_[1] }
 737
 738 sub trans_loc     { $_[0]{trans_loc} }
 739 sub set_trans_loc { $_[0]{trans_loc} = $_[1] }
 740
 741 sub chains_loc     { $_[0]{chains_loc} }
 742 sub set_chains_loc { $_[0]{chains_loc} = $_[1] }
 743
 744 ################################################################################
 745
 746 package DBM::Deep::Iterator;
 747
 748 sub new {
 749     my $class = shift;
 750     my ($args) = @_;
 751
 752     my $self = bless {
 753         breadcrumbs => [],
 754         engine      => $args->{engine},
 755         base_offset => $args->{base_offset},
 756     }, $class;
 757
 758     Scalar::Util::weaken( $self->{engine} );
 759
 760     return $self;
 761 }
 762
 763 sub reset { $_[0]{breadcrumbs} = [] }
 764
 765 sub get_sector_iterator {
 766     my $self = shift;
 767     my ($loc) = @_;
 768
 769     my $sector = $self->{engine}->_load_sector( $loc )
 770         or return;
 771
 772     if ( $sector->isa( 'DBM::Deep::Engine::Sector::Index' ) ) {
 773         return DBM::Deep::Iterator::Index->new({
 774             iterator => $self,
 775             sector   => $sector,
 776         });
 777     }
 778     elsif ( $sector->isa( 'DBM::Deep::Engine::Sector::BucketList' ) ) {
 779         return DBM::Deep::Iterator::BucketList->new({
 780             iterator => $self,
 781             sector   => $sector,
 782         });
 783     }
 784
 785     DBM::Deep->_throw_error( "get_sector_iterator(): Why did $loc make a $sector?" );
 786 }
 787
 788 sub get_next_key {
 789     my $self = shift;
 790     my ($obj) = @_;
 791
 792     my $crumbs = $self->{breadcrumbs};
 793     my $e = $self->{engine};
 794
 795     unless ( @$crumbs ) {
 796         # This will be a Reference sector
 797         my $sector = $e->_load_sector( $self->{base_offset} )
 798             # If no sector is found, thist must have been deleted from under us.
 799             or return;
 800
 801         if ( $sector->staleness != $obj->_staleness ) {
 802             return;
 803         }
 804
 805         my $loc = $sector->get_blist_loc
 806             or return;
 807
 808         push @$crumbs, $self->get_sector_iterator( $loc );
 809     }
 810
 811     FIND_NEXT_KEY: {
 812         # We're at the end.
 813         unless ( @$crumbs ) {
 814             $self->reset;
 815             return;
 816         }
 817
 818         my $iterator = $crumbs->[-1];
 819
 820         # This level is done.
 821         if ( $iterator->at_end ) {
 822             pop @$crumbs;
 823             redo FIND_NEXT_KEY;
 824         }
 825
 826         if ( $iterator->isa( 'DBM::Deep::Iterator::Index' ) ) {
 827             # If we don't have any more, it will be caught at the
 828             # prior check.
 829             if ( my $next = $iterator->get_next_iterator ) {
 830                 push @$crumbs, $next;
 831             }
 832             redo FIND_NEXT_KEY;
 833         }
 834
 835         unless ( $iterator->isa( 'DBM::Deep::Iterator::BucketList' ) ) {
 836             DBM::Deep->_throw_error(
 837                 "Should have a bucketlist iterator here - instead have $iterator"
 838             );
 839         }
 840
 841         # At this point, we have a BucketList iterator
 842         my $key = $iterator->get_next_key;
 843         if ( defined $key ) {
 844             return $key;
 845         }
 846         #XXX else { $iterator->set_to_end() } ?
 847
 848         # We hit the end of the bucketlist iterator, so redo
 849         redo FIND_NEXT_KEY;
 850     }
 851
 852     DBM::Deep->_throw_error( "get_next_key(): How did we get here?" );
 853 }
 854
 855 package DBM::Deep::Iterator::Index;
 856
 857 sub new {
 858     my $self = bless $_[1] => $_[0];
 859     $self->{curr_index} = 0;
 860     return $self;
 861 }
 862
 863 sub at_end {
 864     my $self = shift;
 865     return $self->{curr_index} >= $self->{iterator}{engine}->hash_chars;
 866 }
 867
 868 sub get_next_iterator {
 869     my $self = shift;
 870
 871     my $loc;
 872     while ( !$loc ) {
 873         return if $self->at_end;
 874         $loc = $self->{sector}->get_entry( $self->{curr_index}++ );
 875     }
 876
 877     return $self->{iterator}->get_sector_iterator( $loc );
 878 }
 879
 880 package DBM::Deep::Iterator::BucketList;
 881
 882 sub new {
 883     my $self = bless $_[1] => $_[0];
 884     $self->{curr_index} = 0;
 885     return $self;
 886 }
 887
 888 sub at_end {
 889     my $self = shift;
 890     return $self->{curr_index} >= $self->{iterator}{engine}->max_buckets;
 891 }
 892
 893 sub get_next_key {
 894     my $self = shift;
 895
 896     return if $self->at_end;
 897
 898     my $idx = $self->{curr_index}++;
 899
 900     my $data_loc = $self->{sector}->get_data_location_for({
 901         allow_head => 1,
 902         idx        => $idx,
 903     }) or return;
 904
 905     #XXX Do we want to add corruption checks here?
 906     return $self->{sector}->get_key_for( $idx )->data;
 907 }
 908
 909 package DBM::Deep::Engine::Sector;
 910
 911 sub new {
 912     my $self = bless $_[1], $_[0];
 913     Scalar::Util::weaken( $self->{engine} );
 914     $self->_init;
 915     return $self;
 916 }
 917
 918 #sub _init {}
 919 #sub clone { DBM::Deep->_throw_error( "Must be implemented in the child class" ); }
 920
 921 sub engine { $_[0]{engine} }
 922 sub offset { $_[0]{offset} }
 923 sub type   { $_[0]{type} }
 924
 925 sub base_size {
 926    my $self = shift;
 927    return $self->engine->SIG_SIZE + $self->engine->STALE_SIZE;
 928 }
 929
 930 sub free {
 931     my $self = shift;
 932
 933     my $e = $self->engine;
 934
 935     $e->storage->print_at( $self->offset, $e->SIG_FREE );
 936     # Skip staleness counter
 937     $e->storage->print_at( $self->offset + $self->base_size,
 938         chr(0) x ($self->size - $self->base_size),
 939     );
 940
 941     my $free_meth = $self->free_meth;
 942     $e->$free_meth( $self->offset, $self->size );
 943
 944     return;
 945 }
 946
 947 package DBM::Deep::Engine::Sector::Data;
 948
 949 our @ISA = qw( DBM::Deep::Engine::Sector );
 950
 951 # This is in bytes
 952 sub size { return 256 }
 953 sub free_meth { return '_add_free_data_sector' }
 954
 955 sub clone {
 956     my $self = shift;
 957     return ref($self)->new({
 958         engine => $self->engine,
 959         data   => $self->data,
 960         type   => $self->type,
 961     });
 962 }
 963
 964 package DBM::Deep::Engine::Sector::Scalar;
 965
 966 our @ISA = qw( DBM::Deep::Engine::Sector::Data );
 967
 968 sub free {
 969     my $self = shift;
 970
 971     my $chain_loc = $self->chain_loc;
 972
 973     $self->SUPER::free();
 974
 975     if ( $chain_loc ) {
 976         $self->engine->_load_sector( $chain_loc )->free;
 977     }
 978
 979     return;
 980 }
 981
 982 sub type { $_[0]{engine}->SIG_DATA }
 983 sub _init {
 984     my $self = shift;
 985
 986     my $engine = $self->engine;
 987
 988     unless ( $self->offset ) {
 989         my $data_section = $self->size - $self->base_size - 1 * $engine->byte_size - 1;
 990
 991         $self->{offset} = $engine->_request_data_sector( $self->size );
 992
 993         my $data = delete $self->{data};
 994         my $dlen = length $data;
 995         my $continue = 1;
 996         my $curr_offset = $self->offset;
 997         while ( $continue ) {
 998
 999             my $next_offset = 0;
1000
1001             my ($leftover, $this_len, $chunk);
1002             if ( $dlen > $data_section ) {
1003                 $leftover = 0;
1004                 $this_len = $data_section;
1005                 $chunk = substr( $data, 0, $this_len );
1006
1007                 $dlen -= $data_section;
1008                 $next_offset = $engine->_request_data_sector( $self->size );
1009                 $data = substr( $data, $this_len );
1010             }
1011             else {
1012                 $leftover = $data_section - $dlen;
1013                 $this_len = $dlen;
1014                 $chunk = $data;
1015
1016                 $continue = 0;
1017             }
1018
1019             $engine->storage->print_at( $curr_offset, $self->type ); # Sector type
1020             # Skip staleness
1021             $engine->storage->print_at( $curr_offset + $self->base_size,
1022                 pack( $StP{$engine->byte_size}, $next_offset ),  # Chain loc
1023                 pack( $StP{1}, $this_len ),                      # Data length
1024                 $chunk,                                          # Data to be stored in this sector
1025                 chr(0) x $leftover,                              # Zero-fill the rest
1026             );
1027
1028             $curr_offset = $next_offset;
1029         }
1030
1031         return;
1032     }
1033 }
1034
1035 sub data_length {
1036     my $self = shift;
1037
1038     my $buffer = $self->engine->storage->read_at(
1039         $self->offset + $self->base_size + $self->engine->byte_size, 1
1040     );
1041
1042     return unpack( $StP{1}, $buffer );
1043 }
1044
1045 sub chain_loc {
1046     my $self = shift;
1047     my $chain_loc = $self->engine->storage->read_at(
1048         $self->offset + $self->base_size, $self->engine->byte_size,
1049     );
1050     return unpack( $StP{$self->engine->byte_size}, $chain_loc );
1051 }
1052
1053 sub data {
1054     my $self = shift;
1055
1056     my $data;
1057     while ( 1 ) {
1058         my $chain_loc = $self->chain_loc;
1059
1060         $data .= $self->engine->storage->read_at(
1061             $self->offset + $self->base_size + $self->engine->byte_size + 1, $self->data_length,
1062         );
1063
1064         last unless $chain_loc;
1065
1066         $self = $self->engine->_load_sector( $chain_loc );
1067     }
1068
1069     return $data;
1070 }
1071
1072 package DBM::Deep::Engine::Sector::Null;
1073
1074 our @ISA = qw( DBM::Deep::Engine::Sector::Data );
1075
1076 sub type { $_[0]{engine}->SIG_NULL }
1077 sub data_length { 0 }
1078 sub data { return }
1079
1080 sub _init {
1081     my $self = shift;
1082
1083     my $engine = $self->engine;
1084
1085     unless ( $self->offset ) {
1086         my $leftover = $self->size - $self->base_size - 1 * $engine->byte_size - 1;
1087
1088         $self->{offset} = $engine->_request_data_sector( $self->size );
1089         $engine->storage->print_at( $self->offset, $self->type ); # Sector type
1090         # Skip staleness counter
1091         $engine->storage->print_at( $self->offset + $self->base_size,
1092             pack( $StP{$engine->byte_size}, 0 ),  # Chain loc
1093             pack( $StP{1}, $self->data_length ),  # Data length
1094             chr(0) x $leftover,                   # Zero-fill the rest
1095         );
1096
1097         return;
1098     }
1099 }
1100
1101 package DBM::Deep::Engine::Sector::Reference;
1102
1103 our @ISA = qw( DBM::Deep::Engine::Sector::Data );
1104
1105 sub _init {
1106     my $self = shift;
1107
1108     my $e = $self->engine;
1109
1110     unless ( $self->offset ) {
1111         my $classname = Scalar::Util::blessed( delete $self->{data} );
1112         my $leftover = $self->size - $self->base_size - 2 * $e->byte_size;
1113
1114         my $class_offset = 0;
1115         if ( defined $classname ) {
1116             my $class_sector = DBM::Deep::Engine::Sector::Scalar->new({
1117                 engine => $e,
1118                 data   => $classname,
1119             });
1120             $class_offset = $class_sector->offset;
1121         }
1122
1123         $self->{offset} = $e->_request_data_sector( $self->size );
1124         $e->storage->print_at( $self->offset, $self->type ); # Sector type
1125         # Skip staleness counter
1126         $e->storage->print_at( $self->offset + $self->base_size,
1127             pack( $StP{$e->byte_size}, 0 ),             # Index/BList loc
1128             pack( $StP{$e->byte_size}, $class_offset ), # Classname loc
1129             chr(0) x $leftover,                         # Zero-fill the rest
1130         );
1131     }
1132     else {
1133         $self->{type} = $e->storage->read_at( $self->offset, 1 );
1134     }
1135
1136     $self->{staleness} = unpack(
1137         $StP{$e->STALE_SIZE},
1138         $e->storage->read_at( $self->offset + $e->SIG_SIZE, $e->STALE_SIZE ),
1139     );
1140
1141     return;
1142 }
1143
1144 sub free {
1145     my $self = shift;
1146
1147     my $blist_loc = $self->get_blist_loc;
1148     $self->engine->_load_sector( $blist_loc )->free if $blist_loc;
1149
1150     my $class_loc = $self->get_class_offset;
1151     $self->engine->_load_sector( $class_loc )->free if $class_loc;
1152
1153     $self->SUPER::free();
1154 }
1155
1156 sub staleness { $_[0]{staleness} }
1157
1158 sub get_data_for {
1159     my $self = shift;
1160     my ($args) = @_;
1161
1162     # Assume that the head is not allowed unless otherwise specified.
1163     $args->{allow_head} = 0 unless exists $args->{allow_head};
1164
1165     # Assume we don't create a new blist location unless otherwise specified.
1166     $args->{create} = 0 unless exists $args->{create};
1167
1168     my $blist = $self->get_bucket_list({
1169         key_md5 => $args->{key_md5},
1170         key => $args->{key},
1171         create  => $args->{create},
1172     });
1173     return unless $blist && $blist->{found};
1174
1175     # At this point, $blist knows where the md5 is. What it -doesn't- know yet
1176     # is whether or not this transaction has this key. That's part of the next
1177     # function call.
1178     my $location = $blist->get_data_location_for({
1179         allow_head => $args->{allow_head},
1180     }) or return;
1181
1182     return $self->engine->_load_sector( $location );
1183 }
1184
1185 sub write_data {
1186     my $self = shift;
1187     my ($args) = @_;
1188
1189     my $blist = $self->get_bucket_list({
1190         key_md5 => $args->{key_md5},
1191         key => $args->{key},
1192         create  => 1,
1193     }) or DBM::Deep->_throw_error( "How did write_data fail (no blist)?!" );
1194
1195     # Handle any transactional bookkeeping.
1196     if ( $self->engine->trans_id ) {
1197         if ( ! $blist->has_md5 ) {
1198             $blist->mark_deleted({
1199                 trans_id => 0,
1200             });
1201         }
1202     }
1203     else {
1204         my @trans_ids = $self->engine->get_running_txn_ids;
1205         if ( $blist->has_md5 ) {
1206             if ( @trans_ids ) {
1207                 my $old_value = $blist->get_data_for;
1208                 foreach my $other_trans_id ( @trans_ids ) {
1209                     next if $blist->get_data_location_for({
1210                         trans_id   => $other_trans_id,
1211                         allow_head => 0,
1212                     });
1213                     $blist->write_md5({
1214                         trans_id => $other_trans_id,
1215                         key      => $args->{key},
1216                         key_md5  => $args->{key_md5},
1217                         value    => $old_value->clone,
1218                     });
1219                 }
1220             }
1221         }
1222         else {
1223             if ( @trans_ids ) {
1224                 foreach my $other_trans_id ( @trans_ids ) {
1225                     #XXX This doesn't seem to possible to ever happen . . .
1226                     next if $blist->get_data_location_for({ trans_id => $other_trans_id, allow_head => 0 });
1227                     $blist->mark_deleted({
1228                         trans_id => $other_trans_id,
1229                     });
1230                 }
1231             }
1232         }
1233     }
1234
1235     #XXX Is this safe to do transactionally?
1236     # Free the place we're about to write to.
1237     if ( $blist->get_data_location_for({ allow_head => 0 }) ) {
1238         $blist->get_data_for({ allow_head => 0 })->free;
1239     }
1240
1241     $blist->write_md5({
1242         key      => $args->{key},
1243         key_md5  => $args->{key_md5},
1244         value    => $args->{value},
1245     });
1246 }
1247
1248 sub delete_key {
1249     my $self = shift;
1250     my ($args) = @_;
1251
1252     # XXX What should happen if this fails?
1253     my $blist = $self->get_bucket_list({
1254         key_md5 => $args->{key_md5},
1255     }) or DBM::Deep->_throw_error( "How did delete_key fail (no blist)?!" );
1256
1257     # Save the location so that we can free the data
1258     my $location = $blist->get_data_location_for({
1259         allow_head => 0,
1260     });
1261     my $old_value = $location && $self->engine->_load_sector( $location );
1262
1263     if ( $self->engine->trans_id == 0 ) {
1264         my @trans_ids = $self->engine->get_running_txn_ids;
1265         if ( @trans_ids ) {
1266             foreach my $other_trans_id ( @trans_ids ) {
1267                 next if $blist->get_data_location_for({ trans_id => $other_trans_id, allow_head => 0 });
1268                 $blist->write_md5({
1269                     trans_id => $other_trans_id,
1270                     key      => $args->{key},
1271                     key_md5  => $args->{key_md5},
1272                     value    => $old_value->clone,
1273                 });
1274             }
1275         }
1276     }
1277
1278     $blist->mark_deleted( $args );
1279
1280     my $data;
1281     if ( $old_value ) {
1282         $data = $old_value->data;
1283         $old_value->free;
1284     }
1285
1286     return $data;
1287 }
1288
1289 sub get_blist_loc {
1290     my $self = shift;
1291
1292     my $e = $self->engine;
1293     my $blist_loc = $e->storage->read_at( $self->offset + $self->base_size, $e->byte_size );
1294     return unpack( $StP{$e->byte_size}, $blist_loc );
1295 }
1296
1297 sub get_bucket_list {
1298     my $self = shift;
1299     my ($args) = @_;
1300     $args ||= {};
1301
1302     # XXX Add in check here for recycling?
1303
1304     my $engine = $self->engine;
1305
1306     my $blist_loc = $self->get_blist_loc;
1307
1308     # There's no index or blist yet
1309     unless ( $blist_loc ) {
1310         return unless $args->{create};
1311
1312         my $blist = DBM::Deep::Engine::Sector::BucketList->new({
1313             engine  => $engine,
1314             key_md5 => $args->{key_md5},
1315         });
1316
1317         $engine->storage->print_at( $self->offset + $self->base_size,
1318             pack( $StP{$engine->byte_size}, $blist->offset ),
1319         );
1320
1321         return $blist;
1322     }
1323
1324     my $sector = $engine->_load_sector( $blist_loc )
1325         or DBM::Deep->_throw_error( "Cannot read sector at $blist_loc in get_bucket_list()" );
1326     my $i = 0;
1327     my $last_sector = undef;
1328     while ( $sector->isa( 'DBM::Deep::Engine::Sector::Index' ) ) {
1329         $blist_loc = $sector->get_entry( ord( substr( $args->{key_md5}, $i++, 1 ) ) );
1330         $last_sector = $sector;
1331         if ( $blist_loc ) {
1332             $sector = $engine->_load_sector( $blist_loc )
1333                 or DBM::Deep->_throw_error( "Cannot read sector at $blist_loc in get_bucket_list()" );
1334         }
1335         else {
1336             $sector = undef;
1337             last;
1338         }
1339     }
1340
1341     # This means we went through the Index sector(s) and found an empty slot
1342     unless ( $sector ) {
1343         return unless $args->{create};
1344
1345         DBM::Deep->_throw_error( "No last_sector when attempting to build a new entry" )
1346             unless $last_sector;
1347
1348         my $blist = DBM::Deep::Engine::Sector::BucketList->new({
1349             engine  => $engine,
1350             key_md5 => $args->{key_md5},
1351         });
1352
1353         $last_sector->set_entry( ord( substr( $args->{key_md5}, $i - 1, 1 ) ) => $blist->offset );
1354
1355         return $blist;
1356     }
1357
1358     $sector->find_md5( $args->{key_md5} );
1359
1360     # See whether or not we need to reindex the bucketlist
1361     if ( !$sector->has_md5 && $args->{create} && $sector->{idx} == -1 ) {
1362         my $new_index = DBM::Deep::Engine::Sector::Index->new({
1363             engine => $engine,
1364         });
1365
1366         my %blist_cache;
1367         #XXX q.v. the comments for this function.
1368         foreach my $entry ( $sector->chopped_up ) {
1369             my ($spot, $md5) = @{$entry};
1370             my $idx = ord( substr( $md5, $i, 1 ) );
1371
1372             # XXX This is inefficient
1373             my $blist = $blist_cache{$idx}
1374                 ||= DBM::Deep::Engine::Sector::BucketList->new({
1375                     engine => $engine,
1376                 });
1377
1378             $new_index->set_entry( $idx => $blist->offset );
1379
1380             my $new_spot = $blist->write_at_next_open( $md5 );
1381             $engine->reindex_entry( $spot => $new_spot );
1382         }
1383
1384         # Handle the new item separately.
1385         {
1386             my $idx = ord( substr( $args->{key_md5}, $i, 1 ) );
1387             my $blist = $blist_cache{$idx}
1388                 ||= DBM::Deep::Engine::Sector::BucketList->new({
1389                     engine => $engine,
1390                 });
1391
1392             $new_index->set_entry( $idx => $blist->offset );
1393
1394             #XXX THIS IS HACKY!
1395             $blist->find_md5( $args->{key_md5} );
1396             $blist->write_md5({
1397                 key     => $args->{key},
1398                 key_md5 => $args->{key_md5},
1399                 value   => DBM::Deep::Engine::Sector::Null->new({
1400                     engine => $engine,
1401                     data   => undef,
1402                 }),
1403             });
1404         }
1405
1406         if ( $last_sector ) {
1407             $last_sector->set_entry(
1408                 ord( substr( $args->{key_md5}, $i - 1, 1 ) ),
1409                 $new_index->offset,
1410             );
1411         } else {
1412             $engine->storage->print_at( $self->offset + $self->base_size,
1413                 pack( $StP{$engine->byte_size}, $new_index->offset ),
1414             );
1415         }
1416
1417         $sector->free;
1418
1419         $sector = $blist_cache{ ord( substr( $args->{key_md5}, $i, 1 ) ) };
1420         $sector->find_md5( $args->{key_md5} );
1421     }
1422
1423     return $sector;
1424 }
1425
1426 sub get_class_offset {
1427     my $self = shift;
1428
1429     my $e = $self->engine;
1430     return unpack(
1431         $StP{$e->byte_size},
1432         $e->storage->read_at(
1433             $self->offset + $self->base_size + 1 * $e->byte_size, $e->byte_size,
1434         ),
1435     );
1436 }
1437
1438 sub get_classname {
1439     my $self = shift;
1440
1441     my $class_offset = $self->get_class_offset;
1442
1443     return unless $class_offset;
1444
1445     return $self->engine->_load_sector( $class_offset )->data;
1446 }
1447
1448 #XXX Add singleton handling here
1449 sub data {
1450     my $self = shift;
1451
1452     my $new_obj = DBM::Deep->new({
1453         type        => $self->type,
1454         base_offset => $self->offset,
1455         staleness   => $self->staleness,
1456         storage     => $self->engine->storage,
1457         engine      => $self->engine,
1458     });
1459
1460     if ( $self->engine->storage->{autobless} ) {
1461         my $classname = $self->get_classname;
1462         if ( defined $classname ) {
1463             bless $new_obj, $classname;
1464         }
1465     }
1466
1467     return $new_obj;
1468 }
1469
1470 package DBM::Deep::Engine::Sector::BucketList;
1471
1472 our @ISA = qw( DBM::Deep::Engine::Sector );
1473
1474 sub _init {
1475     my $self = shift;
1476
1477     my $engine = $self->engine;
1478
1479     unless ( $self->offset ) {
1480         my $leftover = $self->size - $self->base_size;
1481
1482         $self->{offset} = $engine->_request_blist_sector( $self->size );
1483         $engine->storage->print_at( $self->offset, $engine->SIG_BLIST ); # Sector type
1484         # Skip staleness counter
1485         $engine->storage->print_at( $self->offset + $self->base_size,
1486             chr(0) x $leftover, # Zero-fill the data
1487         );
1488     }
1489
1490     if ( $self->{key_md5} ) {
1491         $self->find_md5;
1492     }
1493
1494     return $self;
1495 }
1496
1497 sub size {
1498     my $self = shift;
1499     unless ( $self->{size} ) {
1500         my $e = $self->engine;
1501         # Base + numbuckets * bucketsize
1502         $self->{size} = $self->base_size + $e->max_buckets * $self->bucket_size;
1503     }
1504     return $self->{size};
1505 }
1506
1507 sub free_meth { return '_add_free_blist_sector' }
1508
1509 sub bucket_size {
1510     my $self = shift;
1511     unless ( $self->{bucket_size} ) {
1512         my $e = $self->engine;
1513         # Key + head (location) + transactions (location + staleness-counter)
1514         my $location_size = $e->byte_size + $e->num_txns * ( $e->byte_size + 4 );
1515         $self->{bucket_size} = $e->hash_size + $location_size;
1516     }
1517     return $self->{bucket_size};
1518 }
1519
1520 # XXX This is such a poor hack. I need to rethink this code.
1521 sub chopped_up {
1522     my $self = shift;
1523
1524     my $e = $self->engine;
1525
1526     my @buckets;
1527     foreach my $idx ( 0 .. $e->max_buckets - 1 ) {
1528         my $spot = $self->offset + $self->base_size + $idx * $self->bucket_size;
1529         my $md5 = $e->storage->read_at( $spot, $e->hash_size );
1530
1531         #XXX If we're chopping, why would we ever have the blank_md5?
1532         last if $md5 eq $e->blank_md5;
1533
1534         my $rest = $e->storage->read_at( undef, $self->bucket_size - $e->hash_size );
1535         push @buckets, [ $spot, $md5 . $rest ];
1536     }
1537
1538     return @buckets;
1539 }
1540
1541 sub write_at_next_open {
1542     my $self = shift;
1543     my ($entry) = @_;
1544
1545     #XXX This is such a hack!
1546     $self->{_next_open} = 0 unless exists $self->{_next_open};
1547
1548     my $spot = $self->offset + $self->base_size + $self->{_next_open}++ * $self->bucket_size;
1549     $self->engine->storage->print_at( $spot, $entry );
1550
1551     return $spot;
1552 }
1553
1554 sub has_md5 {
1555     my $self = shift;
1556     unless ( exists $self->{found} ) {
1557         $self->find_md5;
1558     }
1559     return $self->{found};
1560 }
1561
1562 sub find_md5 {
1563     my $self = shift;
1564
1565     $self->{found} = undef;
1566     $self->{idx}   = -1;
1567
1568     if ( @_ ) {
1569         $self->{key_md5} = shift;
1570     }
1571
1572     # If we don't have an MD5, then what are we supposed to do?
1573     unless ( exists $self->{key_md5} ) {
1574         DBM::Deep->_throw_error( "Cannot find_md5 without a key_md5 set" );
1575     }
1576
1577     my $e = $self->engine;
1578     foreach my $idx ( 0 .. $e->max_buckets - 1 ) {
1579         my $potential = $e->storage->read_at(
1580             $self->offset + $self->base_size + $idx * $self->bucket_size, $e->hash_size,
1581         );
1582
1583         if ( $potential eq $e->blank_md5 ) {
1584             $self->{idx} = $idx;
1585             return;
1586         }
1587
1588         if ( $potential eq $self->{key_md5} ) {
1589             $self->{found} = 1;
1590             $self->{idx} = $idx;
1591             return;
1592         }
1593     }
1594
1595     return;
1596 }
1597
1598 sub write_md5 {
1599     my $self = shift;
1600     my ($args) = @_;
1601
1602     DBM::Deep->_throw_error( "write_md5: no key" ) unless exists $args->{key};
1603     DBM::Deep->_throw_error( "write_md5: no key_md5" ) unless exists $args->{key_md5};
1604     DBM::Deep->_throw_error( "write_md5: no value" ) unless exists $args->{value};
1605
1606     my $engine = $self->engine;
1607
1608     $args->{trans_id} = $engine->trans_id unless exists $args->{trans_id};
1609
1610     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
1611     $engine->add_entry( $args->{trans_id}, $spot );
1612
1613     unless ($self->{found}) {
1614         my $key_sector = DBM::Deep::Engine::Sector::Scalar->new({
1615             engine => $engine,
1616             data   => $args->{key},
1617         });
1618
1619         $engine->storage->print_at( $spot,
1620             $args->{key_md5},
1621             pack( $StP{$engine->byte_size}, $key_sector->offset ),
1622         );
1623     }
1624
1625     my $loc = $spot
1626       + $engine->hash_size
1627       + $engine->byte_size
1628       + $args->{trans_id} * ( $engine->byte_size + 4 );
1629
1630     $engine->storage->print_at( $loc,
1631         pack( $StP{$engine->byte_size}, $args->{value}->offset ),
1632         pack( 'N', $engine->get_txn_staleness_counter( $args->{trans_id} ) ),
1633     );
1634 }
1635
1636 sub mark_deleted {
1637     my $self = shift;
1638     my ($args) = @_;
1639     $args ||= {};
1640
1641     my $engine = $self->engine;
1642
1643     $args->{trans_id} = $engine->trans_id unless exists $args->{trans_id};
1644
1645     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
1646     $engine->add_entry( $args->{trans_id}, $spot );
1647
1648     my $loc = $spot
1649       + $engine->hash_size
1650       + $engine->byte_size
1651       + $args->{trans_id} * ( $engine->byte_size + 4 );
1652
1653     $engine->storage->print_at( $loc,
1654         pack( $StP{$engine->byte_size}, 1 ), # 1 is the marker for deleted
1655         pack( 'N', $engine->get_txn_staleness_counter( $args->{trans_id} ) ),
1656     );
1657 }
1658
1659 =pod
1660 # This has been commented out until it is used in order to not bring coverage stats down.
1661 sub delete_md5 {
1662     my $self = shift;
1663     my ($args) = @_;
1664
1665     my $engine = $self->engine;
1666     return undef unless $self->{found};
1667
1668     # Save the location so that we can free the data
1669     my $location = $self->get_data_location_for({
1670         allow_head => 0,
1671     });
1672     my $key_sector = $self->get_key_for;
1673
1674     #XXX This isn't going to work right and you know it! This eradicates data
1675     # that we're not ready to eradicate just yet.
1676     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
1677     $engine->storage->print_at( $spot,
1678         $engine->storage->read_at(
1679             $spot + $self->bucket_size,
1680             $self->bucket_size * ( $engine->num_txns - $self->{idx} - 1 ),
1681         ),
1682         chr(0) x $self->bucket_size,
1683     );
1684
1685     $key_sector->free;
1686
1687     my $data_sector = $self->engine->_load_sector( $location );
1688     my $data = $data_sector->data;
1689     $data_sector->free;
1690
1691     return $data;
1692 }
1693 =cut
1694
1695 sub get_data_location_for {
1696     my $self = shift;
1697     my ($args) = @_;
1698     $args ||= {};
1699
1700     $args->{allow_head} = 0 unless exists $args->{allow_head};
1701     $args->{trans_id}   = $self->engine->trans_id unless exists $args->{trans_id};
1702     $args->{idx}        = $self->{idx} unless exists $args->{idx};
1703
1704     my $e = $self->engine;
1705
1706     my $spot = $self->offset + $self->base_size
1707       + $args->{idx} * $self->bucket_size
1708       + $e->hash_size
1709       + $e->byte_size
1710       + $args->{trans_id} * ( $e->byte_size + 4 );
1711
1712     my $buffer = $e->storage->read_at(
1713         $spot,
1714         $e->byte_size + 4,
1715     );
1716     my ($loc, $staleness) = unpack( $StP{$e->byte_size} . ' N', $buffer );
1717
1718     # We have found an entry that is old, so get rid of it
1719     if ( $staleness != (my $s = $e->get_txn_staleness_counter( $args->{trans_id} ) ) ) {
1720         $e->storage->print_at(
1721             $spot,
1722             pack( $StP{$e->byte_size} . ' N', (0) x 2 ),
1723         );
1724         $loc = 0;
1725     }
1726
1727     # If we're in a transaction and we never wrote to this location, try the
1728     # HEAD instead.
1729     if ( $args->{trans_id} && !$loc && $args->{allow_head} ) {
1730         return $self->get_data_location_for({
1731             trans_id   => 0,
1732             allow_head => 1,
1733             idx        => $args->{idx},
1734         });
1735     }
1736     return $loc <= 1 ? 0 : $loc;
1737 }
1738
1739 sub get_data_for {
1740     my $self = shift;
1741     my ($args) = @_;
1742     $args ||= {};
1743
1744     return unless $self->{found};
1745     my $location = $self->get_data_location_for({
1746         allow_head => $args->{allow_head},
1747     });
1748     return $self->engine->_load_sector( $location );
1749 }
1750
1751 sub get_key_for {
1752     my $self = shift;
1753     my ($idx) = @_;
1754     $idx = $self->{idx} unless defined $idx;
1755
1756     if ( $idx >= $self->engine->max_buckets ) {
1757         DBM::Deep->_throw_error( "get_key_for(): Attempting to retrieve $idx" );
1758     }
1759
1760     my $location = $self->engine->storage->read_at(
1761         $self->offset + $self->base_size + $idx * $self->bucket_size + $self->engine->hash_size,
1762         $self->engine->byte_size,
1763     );
1764     $location = unpack( $StP{$self->engine->byte_size}, $location );
1765     DBM::Deep->_throw_error( "get_key_for: No location?" ) unless $location;
1766
1767     return $self->engine->_load_sector( $location );
1768 }
1769
1770 package DBM::Deep::Engine::Sector::Index;
1771
1772 our @ISA = qw( DBM::Deep::Engine::Sector );
1773
1774 sub _init {
1775     my $self = shift;
1776
1777     my $engine = $self->engine;
1778
1779     unless ( $self->offset ) {
1780         my $leftover = $self->size - $self->base_size;
1781
1782         $self->{offset} = $engine->_request_index_sector( $self->size );
1783         $engine->storage->print_at( $self->offset, $engine->SIG_INDEX ); # Sector type
1784         # Skip staleness counter
1785         $engine->storage->print_at( $self->offset + $self->base_size,
1786             chr(0) x $leftover, # Zero-fill the rest
1787         );
1788     }
1789
1790     return $self;
1791 }
1792
1793 sub size {
1794     my $self = shift;
1795     unless ( $self->{size} ) {
1796         my $e = $self->engine;
1797         $self->{size} = $self->base_size + $e->byte_size * $e->hash_chars;
1798     }
1799     return $self->{size};
1800 }
1801
1802 sub free_meth { return '_add_free_index_sector' }
1803
1804 sub free {
1805     my $self = shift;
1806     my $e = $self->engine;
1807
1808     for my $i ( 0 .. $e->hash_chars - 1 ) {
1809         my $l = $self->get_entry( $i ) or next;
1810         $e->_load_sector( $l )->free;
1811     }
1812
1813     $self->SUPER::free();
1814 }
1815
1816 sub _loc_for {
1817     my $self = shift;
1818     my ($idx) = @_;
1819     return $self->offset + $self->base_size + $idx * $self->engine->byte_size;
1820 }
1821
1822 sub get_entry {
1823     my $self = shift;
1824     my ($idx) = @_;
1825
1826     my $e = $self->engine;
1827
1828     DBM::Deep->_throw_error( "get_entry: Out of range ($idx)" )
1829         if $idx < 0 || $idx >= $e->hash_chars;
1830
1831     return unpack(
1832         $StP{$e->byte_size},
1833         $e->storage->read_at( $self->_loc_for( $idx ), $e->byte_size ),
1834     );
1835 }
1836
1837 sub set_entry {
1838     my $self = shift;
1839     my ($idx, $loc) = @_;
1840
1841     my $e = $self->engine;
1842
1843     DBM::Deep->_throw_error( "set_entry: Out of range ($idx)" )
1844         if $idx < 0 || $idx >= $e->hash_chars;
1845
1846     $self->engine->storage->print_at(
1847         $self->_loc_for( $idx ),
1848         pack( $StP{$e->byte_size}, $loc ),
1849     );
1850 }
1851
1852 1;
1853 __END__