lib/DBM/Deep/Engine3.pm

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