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    => 16, # HEAD plus 15 running txns
  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             die "Unknown pack_size value: '$args->{pack_size}'\n";
  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 die "How did get_classname fail (no sector for '$obj')?!\n";
 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 die "Cannot write to a deleted spot in DBM::Deep.\n";
 226
 227     if ( $sector->staleness != $obj->_staleness ) {
 228         die "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 a 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 a 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 a 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 + 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('N', 0 ),                   # Transaction activeness bitfield
 550             pack('N' . $self->num_txns, 0 x $self->num_txns ), # Transaction staleness counters
 551             pack($StP{$self->byte_size}, 0), # Start of free chain (blist size)
 552             pack($StP{$self->byte_size}, 0), # Start of free chain (data size)
 553             pack($StP{$self->byte_size}, 0), # Start of free chain (index size)
 554         );
 555
 556         $self->set_trans_loc( $header_fixed + 2 );
 557         $self->set_chains_loc( $header_fixed + 2 + 4 + 4 * $self->num_txns );
 558
 559         return;
 560     }
 561
 562     sub _read_file_header {
 563         my $self = shift;
 564
 565         my $buffer = $self->storage->read_at( 0, $header_fixed );
 566         return unless length($buffer);
 567
 568         my ($file_signature, $sig_header, $header_version, $size) = unpack(
 569             'A4 A N N', $buffer
 570         );
 571
 572         unless ( $file_signature eq SIG_FILE ) {
 573             $self->storage->close;
 574             DBM::Deep->_throw_error( "Signature not found -- file is not a Deep DB" );
 575         }
 576
 577         unless ( $sig_header eq SIG_HEADER ) {
 578             $self->storage->close;
 579             DBM::Deep->_throw_error( "Old file version found." );
 580         }
 581
 582         my $buffer2 = $self->storage->read_at( undef, $size );
 583         my @values = unpack( 'C C', $buffer2 );
 584
 585         $self->set_trans_loc( $header_fixed + 2 );
 586         $self->set_chains_loc( $header_fixed + 2 + 4 + 4 * $self->num_txns );
 587
 588         if ( @values < 2 || grep { !defined } @values ) {
 589             $self->storage->close;
 590             DBM::Deep->_throw_error("Corrupted file - bad header");
 591         }
 592
 593         #XXX Add warnings if values weren't set right
 594         @{$self}{qw(byte_size max_buckets)} = @values;
 595
 596         my $header_var = 1 + 1 + 4 + 4 * $self->num_txns + 3 * $self->byte_size;
 597         unless ( $size eq $header_var ) {
 598             $self->storage->close;
 599             DBM::Deep->_throw_error( "Unexpected size found ($size <-> $header_var)." );
 600         }
 601
 602         return length($buffer) + length($buffer2);
 603     }
 604 }
 605
 606 sub _load_sector {
 607     my $self = shift;
 608     my ($offset) = @_;
 609
 610     # Add a catch for offset of 0 or 1
 611     return if $offset <= 1;
 612
 613     my $type = $self->storage->read_at( $offset, 1 );
 614     return if $type eq chr(0);
 615
 616     if ( $type eq $self->SIG_ARRAY || $type eq $self->SIG_HASH ) {
 617         return DBM::Deep::Engine::Sector::Reference->new({
 618             engine => $self,
 619             type   => $type,
 620             offset => $offset,
 621         });
 622     }
 623     # XXX Don't we need key_md5 here?
 624     elsif ( $type eq $self->SIG_BLIST ) {
 625         return DBM::Deep::Engine::Sector::BucketList->new({
 626             engine => $self,
 627             type   => $type,
 628             offset => $offset,
 629         });
 630     }
 631     elsif ( $type eq $self->SIG_INDEX ) {
 632         return DBM::Deep::Engine::Sector::Index->new({
 633             engine => $self,
 634             type   => $type,
 635             offset => $offset,
 636         });
 637     }
 638     elsif ( $type eq $self->SIG_NULL ) {
 639         return DBM::Deep::Engine::Sector::Null->new({
 640             engine => $self,
 641             type   => $type,
 642             offset => $offset,
 643         });
 644     }
 645     elsif ( $type eq $self->SIG_DATA ) {
 646         return DBM::Deep::Engine::Sector::Scalar->new({
 647             engine => $self,
 648             type   => $type,
 649             offset => $offset,
 650         });
 651     }
 652     # This was deleted from under us, so just return and let the caller figure it out.
 653     elsif ( $type eq $self->SIG_FREE ) {
 654         return;
 655     }
 656
 657     die "'$offset': Don't know what to do with type '$type'\n";
 658 }
 659
 660 sub _apply_digest {
 661     my $self = shift;
 662     return $self->{digest}->(@_);
 663 }
 664
 665 sub _add_free_blist_sector { shift->_add_free_sector( 0, @_ ) }
 666 sub _add_free_data_sector { shift->_add_free_sector( 1, @_ ) }
 667 sub _add_free_index_sector { shift->_add_free_sector( 2, @_ ) }
 668
 669 sub _add_free_sector {
 670     my $self = shift;
 671     my ($multiple, $offset, $size) = @_;
 672
 673     my $chains_offset = $multiple * $self->byte_size;
 674
 675     my $storage = $self->storage;
 676
 677     # Increment staleness.
 678     # XXX Can this increment+modulo be done by "&= 0x1" ?
 679     my $staleness = unpack( $StP{STALE_SIZE()}, $storage->read_at( $offset + SIG_SIZE, STALE_SIZE ) );
 680     $staleness = ($staleness + 1 ) % ( 2 ** ( 8 * STALE_SIZE ) );
 681     $storage->print_at( $offset + SIG_SIZE, pack( $StP{STALE_SIZE()}, $staleness ) );
 682
 683     my $old_head = $storage->read_at( $self->chains_loc + $chains_offset, $self->byte_size );
 684
 685     $storage->print_at( $self->chains_loc + $chains_offset,
 686         pack( $StP{$self->byte_size}, $offset ),
 687     );
 688
 689     # Record the old head in the new sector after the signature and staleness counter
 690     $storage->print_at( $offset + SIG_SIZE + STALE_SIZE, $old_head );
 691 }
 692
 693 sub _request_blist_sector { shift->_request_sector( 0, @_ ) }
 694 sub _request_data_sector { shift->_request_sector( 1, @_ ) }
 695 sub _request_index_sector { shift->_request_sector( 2, @_ ) }
 696
 697 sub _request_sector {
 698     my $self = shift;
 699     my ($multiple, $size) = @_;
 700
 701     my $chains_offset = $multiple * $self->byte_size;
 702
 703     my $old_head = $self->storage->read_at( $self->chains_loc + $chains_offset, $self->byte_size );
 704     my $loc = unpack( $StP{$self->byte_size}, $old_head );
 705
 706     # We don't have any free sectors of the right size, so allocate a new one.
 707     unless ( $loc ) {
 708         my $offset = $self->storage->request_space( $size );
 709
 710         # Zero out the new sector. This also guarantees correct increases
 711         # in the filesize.
 712         $self->storage->print_at( $offset, chr(0) x $size );
 713
 714         return $offset;
 715     }
 716
 717     # Read the new head after the signature and the staleness counter
 718     my $new_head = $self->storage->read_at( $loc + SIG_SIZE + STALE_SIZE, $self->byte_size );
 719     $self->storage->print_at( $self->chains_loc + $chains_offset, $new_head );
 720
 721     return $loc;
 722 }
 723
 724 ################################################################################
 725
 726 sub storage     { $_[0]{storage} }
 727 sub byte_size   { $_[0]{byte_size} }
 728 sub hash_size   { $_[0]{hash_size} }
 729 sub hash_chars  { $_[0]{hash_chars} }
 730 sub num_txns    { $_[0]{num_txns} }
 731 sub max_buckets { $_[0]{max_buckets} }
 732 sub blank_md5   { chr(0) x $_[0]->hash_size }
 733
 734 sub trans_id     { $_[0]{trans_id} }
 735 sub set_trans_id { $_[0]{trans_id} = $_[1] }
 736
 737 sub trans_loc     { $_[0]{trans_loc} }
 738 sub set_trans_loc { $_[0]{trans_loc} = $_[1] }
 739
 740 sub chains_loc     { $_[0]{chains_loc} }
 741 sub set_chains_loc { $_[0]{chains_loc} = $_[1] }
 742
 743 ################################################################################
 744
 745 package DBM::Deep::Iterator;
 746
 747 sub new {
 748     my $class = shift;
 749     my ($args) = @_;
 750
 751     my $self = bless {
 752         breadcrumbs => [],
 753         engine      => $args->{engine},
 754         base_offset => $args->{base_offset},
 755     }, $class;
 756
 757     Scalar::Util::weaken( $self->{engine} );
 758
 759     return $self;
 760 }
 761
 762 sub reset { $_[0]{breadcrumbs} = [] }
 763
 764 sub get_sector_iterator {
 765     my $self = shift;
 766     my ($loc) = @_;
 767
 768     my $sector = $self->{engine}->_load_sector( $loc )
 769         or return;
 770
 771     if ( $sector->isa( 'DBM::Deep::Engine::Sector::Index' ) ) {
 772         return DBM::Deep::Iterator::Index->new({
 773             iterator => $self,
 774             sector   => $sector,
 775         });
 776     }
 777     elsif ( $sector->isa( 'DBM::Deep::Engine::Sector::BucketList' ) ) {
 778         return DBM::Deep::Iterator::BucketList->new({
 779             iterator => $self,
 780             sector   => $sector,
 781         });
 782     }
 783     else {
 784         die "Why did $loc make a $sector?";
 785     }
 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
 847         # We hit the end of the bucketlist iterator, so redo
 848         redo FIND_NEXT_KEY;
 849     }
 850
 851     DBM::Deep->_throw_error( "get_next_key(): How did we get here?" );
 852 }
 853
 854 package DBM::Deep::Iterator::Index;
 855
 856 sub new {
 857     my $self = bless $_[1] => $_[0];
 858     $self->{curr_index} = 0;
 859     return $self;
 860 }
 861
 862 sub at_end {
 863     my $self = shift;
 864     return $self->{curr_index} >= $self->{iterator}{engine}->hash_chars;
 865 }
 866
 867 sub get_next_iterator {
 868     my $self = shift;
 869
 870     my $loc;
 871     while ( !$loc ) {
 872         return if $self->at_end;
 873         $loc = $self->{sector}->get_entry( $self->{curr_index}++ );
 874     }
 875
 876     return $self->{iterator}->get_sector_iterator( $loc );
 877 }
 878
 879 package DBM::Deep::Iterator::BucketList;
 880
 881 sub new {
 882     my $self = bless $_[1] => $_[0];
 883     $self->{curr_index} = 0;
 884     return $self;
 885 }
 886
 887 sub at_end {
 888     my $self = shift;
 889     return $self->{curr_index} >= $self->{iterator}{engine}->max_buckets;
 890 }
 891
 892 sub get_next_key {
 893     my $self = shift;
 894
 895     return if $self->at_end;
 896
 897     my $data_loc = $self->{sector}->get_data_location_for({
 898         allow_head => 1,
 899         idx => $self->{curr_index}++,
 900     }) or return;
 901
 902     my $key_sector = $self->{sector}->get_key_for( $self->{curr_index} - 1 );
 903
 904     #XXX Is this check necessary now?
 905     return unless $key_sector;
 906
 907     return $key_sector->data;
 908 }
 909
 910 package DBM::Deep::Engine::Sector;
 911
 912 sub new {
 913     my $self = bless $_[1], $_[0];
 914     Scalar::Util::weaken( $self->{engine} );
 915     $self->_init;
 916     return $self;
 917 }
 918 sub _init {}
 919 sub clone { die "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 die "How did write_data fail (no blist)?!\n";
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                     next if $blist->get_data_location_for({ trans_id => $other_trans_id, allow_head => 0 });
1226                     $blist->mark_deleted({
1227                         trans_id => $other_trans_id,
1228                     });
1229                 }
1230             }
1231         }
1232     }
1233
1234     #XXX Is this safe to do transactionally?
1235     # Free the place we're about to write to.
1236     if ( $blist->get_data_location_for({ allow_head => 0 }) ) {
1237         $blist->get_data_for({ allow_head => 0 })->free;
1238     }
1239
1240     $blist->write_md5({
1241         key      => $args->{key},
1242         key_md5  => $args->{key_md5},
1243         value    => $args->{value},
1244     });
1245 }
1246
1247 sub delete_key {
1248     my $self = shift;
1249     my ($args) = @_;
1250
1251     # XXX What should happen if this fails?
1252     my $blist = $self->get_bucket_list({
1253         key_md5 => $args->{key_md5},
1254     }) or die "How did delete_key fail (no blist)?!\n";
1255
1256     # Save the location so that we can free the data
1257     my $location = $blist->get_data_location_for({
1258         allow_head => 0,
1259     });
1260     my $old_value = $location && $self->engine->_load_sector( $location );
1261
1262     if ( $self->engine->trans_id == 0 ) {
1263         my @trans_ids = $self->engine->get_running_txn_ids;
1264         if ( @trans_ids ) {
1265             foreach my $other_trans_id ( @trans_ids ) {
1266                 next if $blist->get_data_location_for({ trans_id => $other_trans_id, allow_head => 0 });
1267                 $blist->write_md5({
1268                     trans_id => $other_trans_id,
1269                     key      => $args->{key},
1270                     key_md5  => $args->{key_md5},
1271                     value    => $old_value->clone,
1272                 });
1273             }
1274         }
1275     }
1276
1277     $blist->mark_deleted( $args );
1278
1279     my $data;
1280     if ( $old_value ) {
1281         $data = $old_value->data;
1282         $old_value->free;
1283     }
1284
1285     return $data;
1286 }
1287
1288 sub get_blist_loc {
1289     my $self = shift;
1290
1291     my $e = $self->engine;
1292     my $blist_loc = $e->storage->read_at( $self->offset + $self->base_size, $e->byte_size );
1293     return unpack( $StP{$e->byte_size}, $blist_loc );
1294 }
1295
1296 sub get_bucket_list {
1297     my $self = shift;
1298     my ($args) = @_;
1299     $args ||= {};
1300
1301     # XXX Add in check here for recycling?
1302
1303     my $engine = $self->engine;
1304
1305     my $blist_loc = $self->get_blist_loc;
1306
1307     # There's no index or blist yet
1308     unless ( $blist_loc ) {
1309         return unless $args->{create};
1310
1311         my $blist = DBM::Deep::Engine::Sector::BucketList->new({
1312             engine  => $engine,
1313             key_md5 => $args->{key_md5},
1314         });
1315
1316         $engine->storage->print_at( $self->offset + $self->base_size,
1317             pack( $StP{$engine->byte_size}, $blist->offset ),
1318         );
1319
1320         return $blist;
1321     }
1322
1323     # Add searching here through the index layers, if any
1324     my $sector = $engine->_load_sector( $blist_loc )
1325         or die "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 die "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         die "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 sub data {
1449     my $self = shift;
1450
1451     my $new_obj = DBM::Deep->new({
1452         type        => $self->type,
1453         base_offset => $self->offset,
1454         staleness   => $self->staleness,
1455         storage     => $self->engine->storage,
1456         engine      => $self->engine,
1457     });
1458
1459     if ( $self->engine->storage->{autobless} ) {
1460         my $classname = $self->get_classname;
1461         if ( defined $classname ) {
1462             bless $new_obj, $classname;
1463         }
1464     }
1465
1466     return $new_obj;
1467 }
1468
1469 package DBM::Deep::Engine::Sector::BucketList;
1470
1471 our @ISA = qw( DBM::Deep::Engine::Sector );
1472
1473 sub _init {
1474     my $self = shift;
1475
1476     my $engine = $self->engine;
1477
1478     unless ( $self->offset ) {
1479         my $leftover = $self->size - $self->base_size;
1480
1481         $self->{offset} = $engine->_request_blist_sector( $self->size );
1482         $engine->storage->print_at( $self->offset, $engine->SIG_BLIST ); # Sector type
1483         # Skip staleness counter
1484         $engine->storage->print_at( $self->offset + $self->base_size,
1485             chr(0) x $leftover, # Zero-fill the data
1486         );
1487     }
1488
1489     if ( $self->{key_md5} ) {
1490         $self->find_md5;
1491     }
1492
1493     return $self;
1494 }
1495
1496 sub size {
1497     my $self = shift;
1498     unless ( $self->{size} ) {
1499         my $e = $self->engine;
1500         # Base + numbuckets * bucketsize
1501         $self->{size} = $self->base_size + $e->max_buckets * $self->bucket_size;
1502     }
1503     return $self->{size};
1504 }
1505
1506 sub free_meth { return '_add_free_blist_sector' }
1507
1508 sub bucket_size {
1509     my $self = shift;
1510     unless ( $self->{bucket_size} ) {
1511         my $e = $self->engine;
1512         # Key + head (location) + transactions (location + staleness-counter)
1513         my $location_size = $e->byte_size + $e->num_txns * ( $e->byte_size + 4 );
1514         $self->{bucket_size} = $e->hash_size + $location_size;
1515     }
1516     return $self->{bucket_size};
1517 }
1518
1519 # XXX This is such a poor hack. I need to rethink this code.
1520 sub chopped_up {
1521     my $self = shift;
1522
1523     my $e = $self->engine;
1524
1525     my @buckets;
1526     foreach my $idx ( 0 .. $e->max_buckets - 1 ) {
1527         my $spot = $self->offset + $self->base_size + $idx * $self->bucket_size;
1528         my $md5 = $e->storage->read_at( $spot, $e->hash_size );
1529
1530         last if $md5 eq $e->blank_md5;
1531
1532         my $rest = $e->storage->read_at( undef, $self->bucket_size - $e->hash_size );
1533         push @buckets, [ $spot, $md5 . $rest ];
1534     }
1535
1536     return @buckets;
1537 }
1538
1539 sub write_at_next_open {
1540     my $self = shift;
1541     my ($entry) = @_;
1542
1543     #XXX This is such a hack!
1544     $self->{_next_open} = 0 unless exists $self->{_next_open};
1545
1546     my $spot = $self->offset + $self->base_size + $self->{_next_open}++ * $self->bucket_size;
1547     $self->engine->storage->print_at( $spot, $entry );
1548
1549     return $spot;
1550 }
1551
1552 sub has_md5 {
1553     my $self = shift;
1554     unless ( exists $self->{found} ) {
1555         $self->find_md5;
1556     }
1557     return $self->{found};
1558 }
1559
1560 sub find_md5 {
1561     my $self = shift;
1562
1563     $self->{found} = undef;
1564     $self->{idx}   = -1;
1565
1566     if ( @_ ) {
1567         $self->{key_md5} = shift;
1568     }
1569
1570     # If we don't have an MD5, then what are we supposed to do?
1571     unless ( exists $self->{key_md5} ) {
1572         DBM::Deep->_throw_error( "Cannot find_md5 without a key_md5 set" );
1573     }
1574
1575     my $e = $self->engine;
1576     foreach my $idx ( 0 .. $e->max_buckets - 1 ) {
1577         my $potential = $e->storage->read_at(
1578             $self->offset + $self->base_size + $idx * $self->bucket_size, $e->hash_size,
1579         );
1580
1581         if ( $potential eq $e->blank_md5 ) {
1582             $self->{idx} = $idx;
1583             return;
1584         }
1585
1586         if ( $potential eq $self->{key_md5} ) {
1587             $self->{found} = 1;
1588             $self->{idx} = $idx;
1589             return;
1590         }
1591     }
1592
1593     return;
1594 }
1595
1596 sub write_md5 {
1597     my $self = shift;
1598     my ($args) = @_;
1599
1600     DBM::Deep->_throw_error( "write_md5: no key" ) unless exists $args->{key};
1601     DBM::Deep->_throw_error( "write_md5: no key_md5" ) unless exists $args->{key_md5};
1602     DBM::Deep->_throw_error( "write_md5: no value" ) unless exists $args->{value};
1603
1604     my $engine = $self->engine;
1605
1606     $args->{trans_id} = $engine->trans_id unless exists $args->{trans_id};
1607
1608     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
1609     $engine->add_entry( $args->{trans_id}, $spot );
1610
1611     unless ($self->{found}) {
1612         my $key_sector = DBM::Deep::Engine::Sector::Scalar->new({
1613             engine => $engine,
1614             data   => $args->{key},
1615         });
1616
1617         $engine->storage->print_at( $spot,
1618             $args->{key_md5},
1619             pack( $StP{$engine->byte_size}, $key_sector->offset ),
1620         );
1621     }
1622
1623     my $loc = $spot
1624       + $engine->hash_size
1625       + $engine->byte_size
1626       + $args->{trans_id} * ( $engine->byte_size + 4 );
1627
1628     $engine->storage->print_at( $loc,
1629         pack( $StP{$engine->byte_size}, $args->{value}->offset ),
1630         pack( 'N', $engine->get_txn_staleness_counter( $args->{trans_id} ) ),
1631     );
1632 }
1633
1634 sub mark_deleted {
1635     my $self = shift;
1636     my ($args) = @_;
1637     $args ||= {};
1638
1639     my $engine = $self->engine;
1640
1641     $args->{trans_id} = $engine->trans_id unless exists $args->{trans_id};
1642
1643     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
1644     $engine->add_entry( $args->{trans_id}, $spot );
1645
1646     my $loc = $spot
1647       + $engine->hash_size
1648       + $engine->byte_size
1649       + $args->{trans_id} * ( $engine->byte_size + 4 );
1650
1651     $engine->storage->print_at( $loc,
1652         pack( $StP{$engine->byte_size}, 1 ), # 1 is the marker for deleted
1653         pack( 'N', $engine->get_txn_staleness_counter( $args->{trans_id} ) ),
1654     );
1655 }
1656
1657 sub delete_md5 {
1658     my $self = shift;
1659     my ($args) = @_;
1660
1661     my $engine = $self->engine;
1662     return undef unless $self->{found};
1663
1664     # Save the location so that we can free the data
1665     my $location = $self->get_data_location_for({
1666         allow_head => 0,
1667     });
1668     my $key_sector = $self->get_key_for;
1669
1670     #XXX This isn't going to work right and you know it! This eradicates data
1671     # that we're not ready to eradicate just yet.
1672     my $spot = $self->offset + $self->base_size + $self->{idx} * $self->bucket_size;
1673     $engine->storage->print_at( $spot,
1674         $engine->storage->read_at(
1675             $spot + $self->bucket_size,
1676             $self->bucket_size * ( $engine->num_txns - $self->{idx} - 1 ),
1677         ),
1678         chr(0) x $self->bucket_size,
1679     );
1680
1681     $key_sector->free;
1682
1683     my $data_sector = $self->engine->_load_sector( $location );
1684     my $data = $data_sector->data;
1685     $data_sector->free;
1686
1687     return $data;
1688 }
1689
1690 sub get_data_location_for {
1691     my $self = shift;
1692     my ($args) = @_;
1693     $args ||= {};
1694
1695     $args->{allow_head} = 0 unless exists $args->{allow_head};
1696     $args->{trans_id}   = $self->engine->trans_id unless exists $args->{trans_id};
1697     $args->{idx}        = $self->{idx} unless exists $args->{idx};
1698
1699     my $e = $self->engine;
1700
1701     my $spot = $self->offset + $self->base_size
1702       + $args->{idx} * $self->bucket_size
1703       + $e->hash_size
1704       + $e->byte_size
1705       + $args->{trans_id} * ( $e->byte_size + 4 );
1706
1707     my $buffer = $e->storage->read_at(
1708         $spot,
1709         $e->byte_size + 4,
1710     );
1711     my ($loc, $staleness) = unpack( $StP{$e->byte_size} . ' N', $buffer );
1712
1713     # We have found an entry that is old, so get rid of it
1714     if ( $staleness != (my $s = $e->get_txn_staleness_counter( $args->{trans_id} ) ) ) {
1715         $e->storage->print_at(
1716             $spot,
1717             pack( $StP{$e->byte_size} . ' N', (0) x 2 ),
1718         );
1719         $loc = 0;
1720     }
1721
1722     # If we're in a transaction and we never wrote to this location, try the
1723     # HEAD instead.
1724     if ( $args->{trans_id} && !$loc && $args->{allow_head} ) {
1725         return $self->get_data_location_for({
1726             trans_id   => 0,
1727             allow_head => 1,
1728             idx        => $args->{idx},
1729         });
1730     }
1731     return $loc <= 1 ? 0 : $loc;
1732 }
1733
1734 sub get_data_for {
1735     my $self = shift;
1736     my ($args) = @_;
1737     $args ||= {};
1738
1739     return unless $self->{found};
1740     my $location = $self->get_data_location_for({
1741         allow_head => $args->{allow_head},
1742     });
1743     return $self->engine->_load_sector( $location );
1744 }
1745
1746 sub get_key_for {
1747     my $self = shift;
1748     my ($idx) = @_;
1749     $idx = $self->{idx} unless defined $idx;
1750
1751     if ( $idx >= $self->engine->max_buckets ) {
1752         DBM::Deep->_throw_error( "get_key_for(): Attempting to retrieve $idx" );
1753     }
1754
1755     my $location = $self->engine->storage->read_at(
1756         $self->offset + $self->base_size + $idx * $self->bucket_size + $self->engine->hash_size,
1757         $self->engine->byte_size,
1758     );
1759     $location = unpack( $StP{$self->engine->byte_size}, $location );
1760     return unless $location;
1761     return $self->engine->_load_sector( $location );
1762 }
1763
1764 package DBM::Deep::Engine::Sector::Index;
1765
1766 our @ISA = qw( DBM::Deep::Engine::Sector );
1767
1768 sub _init {
1769     my $self = shift;
1770
1771     my $engine = $self->engine;
1772
1773     unless ( $self->offset ) {
1774         my $leftover = $self->size - $self->base_size;
1775
1776         $self->{offset} = $engine->_request_index_sector( $self->size );
1777         $engine->storage->print_at( $self->offset, $engine->SIG_INDEX ); # Sector type
1778         # Skip staleness counter
1779         $engine->storage->print_at( $self->offset + $self->base_size,
1780             chr(0) x $leftover, # Zero-fill the rest
1781         );
1782     }
1783
1784     return $self;
1785 }
1786
1787 sub size {
1788     my $self = shift;
1789     unless ( $self->{size} ) {
1790         my $e = $self->engine;
1791         $self->{size} = $self->base_size + $e->byte_size * $e->hash_chars;
1792     }
1793     return $self->{size};
1794 }
1795
1796 sub free_meth { return '_add_free_index_sector' }
1797
1798 sub free {
1799     my $self = shift;
1800     my $e = $self->engine;
1801
1802     for my $i ( 0 .. $e->hash_chars - 1 ) {
1803         my $l = $self->location_for( $i ) or next;
1804         $e->_load_sector( $l )->free;
1805     }
1806
1807     $self->SUPER::free();
1808 }
1809
1810 sub _loc_for {
1811     my $self = shift;
1812     my ($idx) = @_;
1813     return $self->offset + $self->base_size + $idx * $self->engine->byte_size;
1814 }
1815
1816 sub get_entry {
1817     my $self = shift;
1818     my ($idx) = @_;
1819
1820     my $e = $self->engine;
1821
1822     die "get_entry: Out of range ($idx)"
1823         if $idx < 0 || $idx >= $e->hash_chars;
1824
1825     return unpack(
1826         $StP{$e->byte_size},
1827         $e->storage->read_at( $self->_loc_for( $idx ), $e->byte_size ),
1828     );
1829 }
1830
1831 sub set_entry {
1832     my $self = shift;
1833     my ($idx, $loc) = @_;
1834
1835     my $e = $self->engine;
1836
1837     die "set_entry: Out of range ($idx)"
1838         if $idx < 0 || $idx >= $e->hash_chars;
1839
1840     $self->engine->storage->print_at(
1841         $self->_loc_for( $idx ),
1842         pack( $StP{$e->byte_size}, $loc ),
1843     );
1844 }
1845
1846 1;
1847 __END__