lib/DBM/Deep/Engine.pm

   1 package DBM::Deep::Engine;
   2
   3 use 5.6.0;
   4
   5 use strict;
   6 use warnings;
   7
   8 use Fcntl qw( :DEFAULT :flock :seek );
   9
  10 # File-wide notes:
  11 # * All the local($/,$\); are to protect read() and print() from -l.
  12 # * To add to bucket_size, make sure you modify the following:
  13 #   - calculate_sizes()
  14 #   - _get_key_subloc()
  15 #   - add_bucket() - where the buckets are printed
  16
  17 ##
  18 # Setup file and tag signatures.  These should never change.
  19 ##
  20 sub SIG_FILE     () { 'DPDB' }
  21 sub SIG_HEADER   () { 'h'    }
  22 sub SIG_INTERNAL () { 'i'    }
  23 sub SIG_HASH     () { 'H'    }
  24 sub SIG_ARRAY    () { 'A'    }
  25 sub SIG_NULL     () { 'N'    }
  26 sub SIG_DATA     () { 'D'    }
  27 sub SIG_INDEX    () { 'I'    }
  28 sub SIG_BLIST    () { 'B'    }
  29 sub SIG_FREE     () { 'F'    }
  30 sub SIG_SIZE     () {  1     }
  31
  32 sub new {
  33     my $class = shift;
  34     my ($args) = @_;
  35
  36     my $self = bless {
  37         long_size   => 4,
  38         long_pack   => 'N',
  39         data_size   => 4,
  40         data_pack   => 'N',
  41
  42         digest      => \&Digest::MD5::md5,
  43         hash_size   => 16,
  44
  45         ##
  46         # Maximum number of buckets per list before another level of indexing is
  47         # done. Increase this value for slightly greater speed, but larger database
  48         # files. DO NOT decrease this value below 16, due to risk of recursive
  49         # reindex overrun.
  50         ##
  51         max_buckets => 16,
  52
  53         fileobj => undef,
  54     }, $class;
  55
  56     if ( defined $args->{pack_size} ) {
  57         if ( lc $args->{pack_size} eq 'small' ) {
  58             $args->{long_size} = 2;
  59             $args->{long_pack} = 'n';
  60         }
  61         elsif ( lc $args->{pack_size} eq 'medium' ) {
  62             $args->{long_size} = 4;
  63             $args->{long_pack} = 'N';
  64         }
  65         elsif ( lc $args->{pack_size} eq 'large' ) {
  66             $args->{long_size} = 8;
  67             $args->{long_pack} = 'Q';
  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     if ( $self->{max_buckets} < 16 ) {
  81         warn "Floor of max_buckets is 16. Setting it to 16 from '$self->{max_buckets}'\n";
  82         $self->{max_buckets} = 16;
  83     }
  84
  85     return $self;
  86 }
  87
  88 sub _fileobj { return $_[0]{fileobj} }
  89 sub _fh      { return $_[0]->_fileobj->{fh} }
  90
  91 sub calculate_sizes {
  92     my $self = shift;
  93
  94     #XXX Does this need to be updated with different hashing algorithms?
  95     $self->{index_size}       = (2**8) * $self->{long_size};
  96     $self->{bucket_size}      = $self->{hash_size} + $self->{long_size} * 3;
  97     $self->{bucket_list_size} = $self->{max_buckets} * $self->{bucket_size};
  98
  99     return;
 100 }
 101
 102 sub write_file_header {
 103     my $self = shift;
 104
 105     local($/,$\);
 106
 107     my $fh = $self->_fh;
 108
 109     my $loc = $self->_request_space( length( SIG_FILE ) + 21 );
 110     seek($fh, $loc + $self->_fileobj->{file_offset}, SEEK_SET);
 111     print( $fh
 112         SIG_FILE,
 113         SIG_HEADER,
 114         pack('N', 1),  # header version
 115         pack('N', 12), # header size
 116         pack('N', 0),  # currently running transaction IDs
 117         pack('n', $self->{long_size}),
 118         pack('A', $self->{long_pack}),
 119         pack('n', $self->{data_size}),
 120         pack('A', $self->{data_pack}),
 121         pack('n', $self->{max_buckets}),
 122     );
 123
 124     $self->_fileobj->set_transaction_offset( 13 );
 125
 126     return;
 127 }
 128
 129 sub read_file_header {
 130     my $self = shift;
 131
 132     local($/,$\);
 133
 134     my $fh = $self->_fh;
 135
 136     seek($fh, 0 + $self->_fileobj->{file_offset}, SEEK_SET);
 137     my $buffer;
 138     my $bytes_read = read( $fh, $buffer, length(SIG_FILE) + 9 );
 139
 140     return unless $bytes_read;
 141
 142     my ($file_signature, $sig_header, $header_version, $size) = unpack(
 143         'A4 A N N', $buffer
 144     );
 145
 146     unless ( $file_signature eq SIG_FILE ) {
 147         $self->_fileobj->close;
 148         $self->_throw_error( "Signature not found -- file is not a Deep DB" );
 149     }
 150
 151     unless ( $sig_header eq SIG_HEADER ) {
 152         $self->_fileobj->close;
 153         $self->_throw_error( "Old file version found." );
 154     }
 155
 156     my $buffer2;
 157     $bytes_read += read( $fh, $buffer2, $size );
 158     my ($running_transactions, @values) = unpack( 'N n A n A n', $buffer2 );
 159
 160     $self->_fileobj->set_transaction_offset( 13 );
 161
 162     if ( @values < 5 || grep { !defined } @values ) {
 163         $self->_fileobj->close;
 164         $self->_throw_error("Corrupted file - bad header");
 165     }
 166
 167     #XXX Add warnings if values weren't set right
 168     @{$self}{qw(long_size long_pack data_size data_pack max_buckets)} = @values;
 169
 170     return $bytes_read;
 171 }
 172
 173 sub setup_fh {
 174     my $self = shift;
 175     my ($obj) = @_;
 176
 177     my $fh = $self->_fh;
 178     flock $fh, LOCK_EX;
 179
 180     #XXX The duplication of calculate_sizes needs to go away
 181     unless ( $obj->{base_offset} ) {
 182         my $bytes_read = $self->read_file_header;
 183
 184         $self->calculate_sizes;
 185
 186         ##
 187         # File is empty -- write header and master index
 188         ##
 189         if (!$bytes_read) {
 190             $self->write_file_header;
 191
 192             $obj->{base_offset} = $self->_request_space( $self->tag_size( $self->{index_size} ) );
 193
 194             $self->write_tag(
 195                 $obj->_base_offset, $obj->_type,
 196                 chr(0)x$self->{index_size},
 197             );
 198
 199             # Flush the filehandle
 200             my $old_fh = select $fh;
 201             my $old_af = $|; $| = 1; $| = $old_af;
 202             select $old_fh;
 203         }
 204         else {
 205             $obj->{base_offset} = $bytes_read;
 206
 207             ##
 208             # Get our type from master index header
 209             ##
 210             my $tag = $self->load_tag($obj->_base_offset)
 211                 or $self->_throw_error("Corrupted file, no master index record");
 212
 213             unless ($obj->_type eq $tag->{signature}) {
 214                 $self->_throw_error("File type mismatch");
 215             }
 216         }
 217     }
 218     else {
 219         $self->calculate_sizes;
 220     }
 221
 222     #XXX We have to make sure we don't mess up when autoflush isn't turned on
 223     unless ( $self->_fileobj->{inode} ) {
 224         my @stats = stat($fh);
 225         $self->_fileobj->{inode} = $stats[1];
 226         $self->_fileobj->{end} = $stats[7];
 227     }
 228
 229     flock $fh, LOCK_UN;
 230
 231     return 1;
 232 }
 233
 234 sub tag_size {
 235     my $self = shift;
 236     my ($size) = @_;
 237     return SIG_SIZE + $self->{data_size} + $size;
 238 }
 239
 240 sub write_tag {
 241     ##
 242     # Given offset, signature and content, create tag and write to disk
 243     ##
 244     my $self = shift;
 245     my ($offset, $sig, $content) = @_;
 246     my $size = length( $content );
 247
 248     local($/,$\);
 249
 250     my $fh = $self->_fh;
 251
 252     if ( defined $offset ) {
 253         seek($fh, $offset + $self->_fileobj->{file_offset}, SEEK_SET);
 254     }
 255
 256     print( $fh $sig . pack($self->{data_pack}, $size) . $content );
 257
 258     return unless defined $offset;
 259
 260     return {
 261         signature => $sig,
 262         size => $size,
 263         offset => $offset + SIG_SIZE + $self->{data_size},
 264         content => $content
 265     };
 266 }
 267
 268 sub load_tag {
 269     ##
 270     # Given offset, load single tag and return signature, size and data
 271     ##
 272     my $self = shift;
 273     my ($offset) = @_;
 274
 275     local($/,$\);
 276
 277 #    print join(':',map{$_||''}caller(1)), $/;
 278
 279     my $fh = $self->_fh;
 280
 281     seek($fh, $offset + $self->_fileobj->{file_offset}, SEEK_SET);
 282
 283     #XXX I'm not sure this check will work if autoflush isn't enabled ...
 284     return if eof $fh;
 285
 286     my $b;
 287     read( $fh, $b, SIG_SIZE + $self->{data_size} );
 288     my ($sig, $size) = unpack( "A $self->{data_pack}", $b );
 289
 290     my $buffer;
 291     read( $fh, $buffer, $size);
 292
 293     return {
 294         signature => $sig,
 295         size => $size,
 296         offset => $offset + SIG_SIZE + $self->{data_size},
 297         content => $buffer
 298     };
 299 }
 300
 301 sub _get_dbm_object {
 302     my $item = shift;
 303
 304     my $obj = eval {
 305         local $SIG{__DIE__};
 306         if ($item->isa( 'DBM::Deep' )) {
 307             return $item;
 308         }
 309         return;
 310     };
 311     return $obj if $obj;
 312
 313     my $r = Scalar::Util::reftype( $item ) || '';
 314     if ( $r eq 'HASH' ) {
 315         my $obj = eval {
 316             local $SIG{__DIE__};
 317             my $obj = tied(%$item);
 318             if ($obj->isa( 'DBM::Deep' )) {
 319                 return $obj;
 320             }
 321             return;
 322         };
 323         return $obj if $obj;
 324     }
 325     elsif ( $r eq 'ARRAY' ) {
 326         my $obj = eval {
 327             local $SIG{__DIE__};
 328             my $obj = tied(@$item);
 329             if ($obj->isa( 'DBM::Deep' )) {
 330                 return $obj;
 331             }
 332             return;
 333         };
 334         return $obj if $obj;
 335     }
 336
 337     return;
 338 }
 339
 340 sub _length_needed {
 341     my $self = shift;
 342     my ($value, $key) = @_;
 343
 344     my $is_dbm_deep = eval {
 345         local $SIG{'__DIE__'};
 346         $value->isa( 'DBM::Deep' );
 347     };
 348
 349     my $len = SIG_SIZE + $self->{data_size}
 350             + $self->{data_size} + length( $key );
 351
 352     if ( $is_dbm_deep && $value->_fileobj eq $self->_fileobj ) {
 353         return $len + $self->{long_size};
 354     }
 355
 356     my $r = Scalar::Util::reftype( $value ) || '';
 357     if ( $self->_fileobj->{autobless} ) {
 358         # This is for the bit saying whether or not this thing is blessed.
 359         $len += 1;
 360     }
 361
 362     unless ( $r eq 'HASH' || $r eq 'ARRAY' ) {
 363         if ( defined $value ) {
 364             $len += length( $value );
 365         }
 366         return $len;
 367     }
 368
 369     $len += $self->{index_size};
 370
 371     # if autobless is enabled, must also take into consideration
 372     # the class name as it is stored after the key.
 373     if ( $self->_fileobj->{autobless} ) {
 374         my $c = Scalar::Util::blessed($value);
 375         if ( defined $c && !$is_dbm_deep ) {
 376             $len += $self->{data_size} + length($c);
 377         }
 378     }
 379
 380     return $len;
 381 }
 382
 383 sub add_bucket {
 384     ##
 385     # Adds one key/value pair to bucket list, given offset, MD5 digest of key,
 386     # plain (undigested) key and value.
 387     ##
 388     my $self = shift;
 389     my ($tag, $md5, $plain_key, $value, $deleted) = @_;
 390     $deleted ||= 0;
 391
 392     local($/,$\);
 393
 394     # This verifies that only supported values will be stored.
 395     {
 396         my $r = Scalar::Util::reftype( $value );
 397         last if !defined $r;
 398
 399         last if $r eq 'HASH';
 400         last if $r eq 'ARRAY';
 401
 402         $self->_throw_error(
 403             "Storage of variables of type '$r' is not supported."
 404         );
 405     }
 406
 407     my $location = 0;
 408     my $result = 2;
 409
 410     my $root = $self->_fileobj;
 411     my $fh   = $self->_fh;
 412
 413     my $actual_length = $self->_length_needed( $value, $plain_key );
 414
 415     #ACID - This is a mutation. Must only find the exact transaction
 416     my ($subloc, $offset, $size,$is_deleted) = $self->_find_in_buckets( $tag, $md5, 1 );
 417
 418     my @transactions;
 419     if ( $self->_fileobj->transaction_id == 0 ) {
 420         @transactions = $self->_fileobj->current_transactions;
 421     }
 422
 423 #    $self->_release_space( $size, $subloc );
 424     # Updating a known md5
 425 #XXX This needs updating to use _release_space
 426     if ( $subloc ) {
 427         $result = 1;
 428
 429         if ($actual_length <= $size) {
 430             $location = $subloc;
 431         }
 432         else {
 433             $location = $self->_request_space( $actual_length );
 434             seek(
 435                 $fh,
 436                 $tag->{offset} + $offset
 437               + $self->{hash_size} + $root->{file_offset},
 438                 SEEK_SET,
 439             );
 440             print( $fh pack($self->{long_pack}, $location ) );
 441             print( $fh pack($self->{long_pack}, $actual_length ) );
 442             print( $fh pack('n n', $root->transaction_id, $deleted ) );
 443         }
 444     }
 445     # Adding a new md5
 446     elsif ( defined $offset ) {
 447         $location = $self->_request_space( $actual_length );
 448
 449         seek( $fh, $tag->{offset} + $offset + $root->{file_offset}, SEEK_SET );
 450         print( $fh $md5 . pack($self->{long_pack}, $location ) );
 451         print( $fh pack($self->{long_pack}, $actual_length ) );
 452         print( $fh pack('n n', $root->transaction_id, $deleted ) );
 453
 454         for ( @transactions ) {
 455             my $tag2 = $self->load_tag( $tag->{offset} - SIG_SIZE - $self->{data_size} );
 456             $self->_fileobj->{transaction_id} = $_;
 457             $self->add_bucket( $tag2, $md5, '', '', 1 );
 458             $self->_fileobj->{transaction_id} = 0;
 459         }
 460     }
 461     # If bucket didn't fit into list, split into a new index level
 462     # split_index() will do the _request_space() call
 463     else {
 464         $location = $self->split_index( $md5, $tag );
 465     }
 466
 467     $self->write_value( $location, $plain_key, $value );
 468
 469     return $result;
 470 }
 471
 472 sub write_value {
 473     my $self = shift;
 474     my ($location, $key, $value) = @_;
 475
 476     local($/,$\);
 477
 478     my $fh = $self->_fh;
 479     my $root = $self->_fileobj;
 480
 481     my $dbm_deep_obj = _get_dbm_object( $value );
 482     if ( $dbm_deep_obj && $dbm_deep_obj->_fileobj ne $self->_fileobj ) {
 483         $self->_throw_error( "Cannot cross-reference. Use export() instead" );
 484     }
 485
 486     seek($fh, $location + $root->{file_offset}, SEEK_SET);
 487
 488     ##
 489     # Write signature based on content type, set content length and write
 490     # actual value.
 491     ##
 492     my $r = Scalar::Util::reftype( $value ) || '';
 493     if ( $dbm_deep_obj ) {
 494         $self->write_tag( undef, SIG_INTERNAL,pack($self->{long_pack}, $dbm_deep_obj->_base_offset) );
 495     }
 496     elsif ($r eq 'HASH') {
 497         if ( !$dbm_deep_obj && tied %{$value} ) {
 498             $self->_throw_error( "Cannot store something that is tied" );
 499         }
 500         $self->write_tag( undef, SIG_HASH, chr(0)x$self->{index_size} );
 501     }
 502     elsif ($r eq 'ARRAY') {
 503         if ( !$dbm_deep_obj && tied @{$value} ) {
 504             $self->_throw_error( "Cannot store something that is tied" );
 505         }
 506         $self->write_tag( undef, SIG_ARRAY, chr(0)x$self->{index_size} );
 507     }
 508     elsif (!defined($value)) {
 509         $self->write_tag( undef, SIG_NULL, '' );
 510     }
 511     else {
 512         $self->write_tag( undef, SIG_DATA, $value );
 513     }
 514
 515     ##
 516     # Plain key is stored AFTER value, as keys are typically fetched less often.
 517     ##
 518     print( $fh pack($self->{data_pack}, length($key)) . $key );
 519
 520     # Internal references don't care about autobless
 521     return 1 if $dbm_deep_obj;
 522
 523     ##
 524     # If value is blessed, preserve class name
 525     ##
 526     if ( $root->{autobless} ) {
 527         my $c = Scalar::Util::blessed($value);
 528         if ( defined $c && !$dbm_deep_obj ) {
 529             print( $fh chr(1) );
 530             print( $fh pack($self->{data_pack}, length($c)) . $c );
 531         }
 532         else {
 533             print( $fh chr(0) );
 534         }
 535     }
 536
 537     ##
 538     # Tie the passed in reference so that changes to it are reflected in the
 539     # datafile. The use of $location as the base_offset will act as the
 540     # the linkage between parent and child.
 541     #
 542     # The overall assignment is a hack around the fact that just tying doesn't
 543     # store the values. This may not be the wrong thing to do.
 544     ##
 545     if ($r eq 'HASH') {
 546         my %x = %$value;
 547         tie %$value, 'DBM::Deep', {
 548             base_offset => $location,
 549             fileobj     => $root,
 550         };
 551         %$value = %x;
 552     }
 553     elsif ($r eq 'ARRAY') {
 554         my @x = @$value;
 555         tie @$value, 'DBM::Deep', {
 556             base_offset => $location,
 557             fileobj     => $root,
 558         };
 559         @$value = @x;
 560     }
 561
 562     return 1;
 563 }
 564
 565 sub split_index {
 566     my $self = shift;
 567     my ($md5, $tag) = @_;
 568
 569     local($/,$\);
 570
 571     my $fh = $self->_fh;
 572     my $root = $self->_fileobj;
 573
 574     my $loc = $self->_request_space(
 575         $self->tag_size( $self->{index_size} ),
 576     );
 577
 578     seek($fh, $tag->{ref_loc} + $root->{file_offset}, SEEK_SET);
 579     print( $fh pack($self->{long_pack}, $loc) );
 580
 581     my $index_tag = $self->write_tag(
 582         $loc, SIG_INDEX,
 583         chr(0)x$self->{index_size},
 584     );
 585
 586     my $newtag_loc = $self->_request_space(
 587         $self->tag_size( $self->{bucket_list_size} ),
 588     );
 589
 590     my $keys = $tag->{content}
 591              . $md5 . pack($self->{long_pack}, $newtag_loc)
 592                     . pack($self->{long_pack}, 0)  # size
 593                     . pack($self->{long_pack}, 0); # transaction ID
 594
 595     my @newloc = ();
 596     BUCKET:
 597     for (my $i = 0; $i <= $self->{max_buckets}; $i++) {
 598         my ($key, $old_subloc, $size) = $self->_get_key_subloc( $keys, $i );
 599
 600         die "[INTERNAL ERROR]: No key in split_index()\n" unless $key;
 601         die "[INTERNAL ERROR]: No subloc in split_index()\n" unless $old_subloc;
 602
 603         my $num = ord(substr($key, $tag->{ch} + 1, 1));
 604
 605         if ($newloc[$num]) {
 606             seek($fh, $newloc[$num] + $root->{file_offset}, SEEK_SET);
 607             my $subkeys;
 608             read( $fh, $subkeys, $self->{bucket_list_size});
 609
 610             # This is looking for the first empty spot
 611             my ($subloc, $offset, $size) = $self->_find_in_buckets(
 612                 { content => $subkeys }, '',
 613             );
 614
 615             seek($fh, $newloc[$num] + $offset + $root->{file_offset}, SEEK_SET);
 616             print( $fh $key . pack($self->{long_pack}, $old_subloc) );
 617
 618             next;
 619         }
 620
 621         seek($fh, $index_tag->{offset} + ($num * $self->{long_size}) + $root->{file_offset}, SEEK_SET);
 622
 623         my $loc = $self->_request_space(
 624             $self->tag_size( $self->{bucket_list_size} ),
 625         );
 626
 627         print( $fh pack($self->{long_pack}, $loc) );
 628
 629         my $blist_tag = $self->write_tag(
 630             $loc, SIG_BLIST,
 631             chr(0)x$self->{bucket_list_size},
 632         );
 633
 634         seek($fh, $blist_tag->{offset} + $root->{file_offset}, SEEK_SET);
 635         print( $fh $key . pack($self->{long_pack}, $old_subloc) );
 636
 637         $newloc[$num] = $blist_tag->{offset};
 638     }
 639
 640     $self->_release_space(
 641         $self->tag_size( $self->{bucket_list_size} ),
 642         $tag->{offset} - SIG_SIZE - $self->{data_size},
 643     );
 644
 645     return $newtag_loc;
 646 }
 647
 648 sub read_from_loc {
 649     my $self = shift;
 650     my ($subloc) = @_;
 651
 652     local($/,$\);
 653
 654     my $fh = $self->_fh;
 655
 656     ##
 657     # Found match -- seek to offset and read signature
 658     ##
 659     my $signature;
 660     seek($fh, $subloc + $self->_fileobj->{file_offset}, SEEK_SET);
 661     read( $fh, $signature, SIG_SIZE);
 662
 663     ##
 664     # If value is a hash or array, return new DBM::Deep object with correct offset
 665     ##
 666     if (($signature eq SIG_HASH) || ($signature eq SIG_ARRAY)) {
 667         my $new_obj = DBM::Deep->new({
 668             type => $signature,
 669             base_offset => $subloc,
 670             fileobj     => $self->_fileobj,
 671         });
 672
 673         if ($new_obj->_fileobj->{autobless}) {
 674             ##
 675             # Skip over value and plain key to see if object needs
 676             # to be re-blessed
 677             ##
 678             seek($fh, $self->{data_size} + $self->{index_size}, SEEK_CUR);
 679
 680             my $size;
 681             read( $fh, $size, $self->{data_size});
 682             $size = unpack($self->{data_pack}, $size);
 683             if ($size) { seek($fh, $size, SEEK_CUR); }
 684
 685             my $bless_bit;
 686             read( $fh, $bless_bit, 1);
 687             if (ord($bless_bit)) {
 688                 ##
 689                 # Yes, object needs to be re-blessed
 690                 ##
 691                 my $class_name;
 692                 read( $fh, $size, $self->{data_size});
 693                 $size = unpack($self->{data_pack}, $size);
 694                 if ($size) { read( $fh, $class_name, $size); }
 695                 if ($class_name) { $new_obj = bless( $new_obj, $class_name ); }
 696             }
 697         }
 698
 699         return $new_obj;
 700     }
 701     elsif ( $signature eq SIG_INTERNAL ) {
 702         my $size;
 703         read( $fh, $size, $self->{data_size});
 704         $size = unpack($self->{data_pack}, $size);
 705
 706         if ( $size ) {
 707             my $new_loc;
 708             read( $fh, $new_loc, $size );
 709             $new_loc = unpack( $self->{long_pack}, $new_loc );
 710
 711             return $self->read_from_loc( $new_loc );
 712         }
 713         else {
 714             return;
 715         }
 716     }
 717     ##
 718     # Otherwise return actual value
 719     ##
 720     elsif ( $signature eq SIG_DATA ) {
 721         my $size;
 722         read( $fh, $size, $self->{data_size});
 723         $size = unpack($self->{data_pack}, $size);
 724
 725         my $value = '';
 726         if ($size) { read( $fh, $value, $size); }
 727         return $value;
 728     }
 729
 730     ##
 731     # Key exists, but content is null
 732     ##
 733     return;
 734 }
 735
 736 sub get_bucket_value {
 737     ##
 738     # Fetch single value given tag and MD5 digested key.
 739     ##
 740     my $self = shift;
 741     my ($tag, $md5) = @_;
 742
 743     #ACID - This is a read. Can find exact or HEAD
 744     my ($subloc, $offset, $size,$is_deleted) = $self->_find_in_buckets( $tag, $md5 );
 745     if ( $subloc && !$is_deleted ) {
 746         return $self->read_from_loc( $subloc );
 747     }
 748     return;
 749 }
 750
 751 sub delete_bucket {
 752     ##
 753     # Delete single key/value pair given tag and MD5 digested key.
 754     ##
 755     my $self = shift;
 756     my ($tag, $md5) = @_;
 757
 758     local($/,$\);
 759
 760     #ACID - This is a mutation. Must only find the exact transaction
 761     my ($subloc, $offset, $size) = $self->_find_in_buckets( $tag, $md5, 1 );
 762 #XXX This needs _release_space()
 763     if ( $subloc ) {
 764         my $fh = $self->_fh;
 765         seek($fh, $tag->{offset} + $offset + $self->_fileobj->{file_offset}, SEEK_SET);
 766         print( $fh substr($tag->{content}, $offset + $self->{bucket_size} ) );
 767         print( $fh chr(0) x $self->{bucket_size} );
 768
 769         return 1;
 770     }
 771     return;
 772 }
 773
 774 sub bucket_exists {
 775     ##
 776     # Check existence of single key given tag and MD5 digested key.
 777     ##
 778     my $self = shift;
 779     my ($tag, $md5) = @_;
 780
 781     #ACID - This is a read. Can find exact or HEAD
 782     my ($subloc, $offset, $size, $is_deleted) = $self->_find_in_buckets( $tag, $md5 );
 783     return ($subloc && !$is_deleted) && 1;
 784 }
 785
 786 sub find_bucket_list {
 787     ##
 788     # Locate offset for bucket list, given digested key
 789     ##
 790     my $self = shift;
 791     my ($offset, $md5, $args) = @_;
 792     $args = {} unless $args;
 793
 794     local($/,$\);
 795
 796     ##
 797     # Locate offset for bucket list using digest index system
 798     ##
 799     my $tag = $self->load_tag( $offset )
 800         or $self->_throw_error( "INTERNAL ERROR - Cannot find tag" );
 801
 802     my $ch = 0;
 803     while ($tag->{signature} ne SIG_BLIST) {
 804         my $num = ord substr($md5, $ch, 1);
 805
 806         my $ref_loc = $tag->{offset} + ($num * $self->{long_size});
 807         $tag = $self->index_lookup( $tag, $num );
 808
 809         if (!$tag) {
 810             return if !$args->{create};
 811
 812             my $loc = $self->_request_space(
 813                 $self->tag_size( $self->{bucket_list_size} ),
 814             );
 815
 816             my $fh = $self->_fh;
 817             seek($fh, $ref_loc + $self->_fileobj->{file_offset}, SEEK_SET);
 818             print( $fh pack($self->{long_pack}, $loc) );
 819
 820             $tag = $self->write_tag(
 821                 $loc, SIG_BLIST,
 822                 chr(0)x$self->{bucket_list_size},
 823             );
 824
 825             $tag->{ref_loc} = $ref_loc;
 826             $tag->{ch} = $ch;
 827
 828             last;
 829         }
 830
 831         $tag->{ch} = $ch++;
 832         $tag->{ref_loc} = $ref_loc;
 833     }
 834
 835     return $tag;
 836 }
 837
 838 sub index_lookup {
 839     ##
 840     # Given index tag, lookup single entry in index and return .
 841     ##
 842     my $self = shift;
 843     my ($tag, $index) = @_;
 844
 845     my $location = unpack(
 846         $self->{long_pack},
 847         substr(
 848             $tag->{content},
 849             $index * $self->{long_size},
 850             $self->{long_size},
 851         ),
 852     );
 853
 854     if (!$location) { return; }
 855
 856     return $self->load_tag( $location );
 857 }
 858
 859 sub traverse_index {
 860     ##
 861     # Scan index and recursively step into deeper levels, looking for next key.
 862     ##
 863     my $self = shift;
 864     my ($obj, $offset, $ch, $force_return_next) = @_;
 865
 866     local($/,$\);
 867
 868     my $tag = $self->load_tag( $offset );
 869
 870     my $fh = $self->_fh;
 871
 872     if ($tag->{signature} ne SIG_BLIST) {
 873         my $content = $tag->{content};
 874         my $start = $obj->{return_next} ? 0 : ord(substr($obj->{prev_md5}, $ch, 1));
 875
 876         for (my $idx = $start; $idx < (2**8); $idx++) {
 877             my $subloc = unpack(
 878                 $self->{long_pack},
 879                 substr(
 880                     $content,
 881                     $idx * $self->{long_size},
 882                     $self->{long_size},
 883                 ),
 884             );
 885
 886             if ($subloc) {
 887                 my $result = $self->traverse_index(
 888                     $obj, $subloc, $ch + 1, $force_return_next,
 889                 );
 890
 891                 if (defined($result)) { return $result; }
 892             }
 893         } # index loop
 894
 895         $obj->{return_next} = 1;
 896     } # tag is an index
 897
 898     else {
 899         my $keys = $tag->{content};
 900         if ($force_return_next) { $obj->{return_next} = 1; }
 901
 902         ##
 903         # Iterate through buckets, looking for a key match
 904         ##
 905         for (my $i = 0; $i < $self->{max_buckets}; $i++) {
 906             my ($key, $subloc) = $self->_get_key_subloc( $keys, $i );
 907
 908             # End of bucket list -- return to outer loop
 909             if (!$subloc) {
 910                 $obj->{return_next} = 1;
 911                 last;
 912             }
 913             # Located previous key -- return next one found
 914             elsif ($key eq $obj->{prev_md5}) {
 915                 $obj->{return_next} = 1;
 916                 next;
 917             }
 918             # Seek to bucket location and skip over signature
 919             elsif ($obj->{return_next}) {
 920                 seek($fh, $subloc + $self->_fileobj->{file_offset}, SEEK_SET);
 921
 922                 # Skip over value to get to plain key
 923                 my $sig;
 924                 read( $fh, $sig, SIG_SIZE );
 925
 926                 my $size;
 927                 read( $fh, $size, $self->{data_size});
 928                 $size = unpack($self->{data_pack}, $size);
 929                 if ($size) { seek($fh, $size, SEEK_CUR); }
 930
 931                 # Read in plain key and return as scalar
 932                 my $plain_key;
 933                 read( $fh, $size, $self->{data_size});
 934                 $size = unpack($self->{data_pack}, $size);
 935                 if ($size) { read( $fh, $plain_key, $size); }
 936
 937                 return $plain_key;
 938             }
 939         }
 940
 941         $obj->{return_next} = 1;
 942     } # tag is a bucket list
 943
 944     return;
 945 }
 946
 947 sub get_next_key {
 948     ##
 949     # Locate next key, given digested previous one
 950     ##
 951     my $self = shift;
 952     my ($obj) = @_;
 953
 954     $obj->{prev_md5} = $_[1] ? $_[1] : undef;
 955     $obj->{return_next} = 0;
 956
 957     ##
 958     # If the previous key was not specifed, start at the top and
 959     # return the first one found.
 960     ##
 961     if (!$obj->{prev_md5}) {
 962         $obj->{prev_md5} = chr(0) x $self->{hash_size};
 963         $obj->{return_next} = 1;
 964     }
 965
 966     return $self->traverse_index( $obj, $obj->_base_offset, 0 );
 967 }
 968
 969 # Utilities
 970
 971 sub _get_key_subloc {
 972     my $self = shift;
 973     my ($keys, $idx) = @_;
 974
 975     my ($key, $subloc, $size, $transaction_id, $is_deleted) = unpack(
 976         # This is 'a', not 'A'. Please read the pack() documentation for the
 977         # difference between the two and why it's important.
 978         "a$self->{hash_size} $self->{long_pack}2 n2",
 979         substr(
 980             $keys,
 981             ($idx * $self->{bucket_size}),
 982             $self->{bucket_size},
 983         ),
 984     );
 985
 986     return ($key, $subloc, $size, $transaction_id, $is_deleted);
 987 }
 988
 989 sub _find_in_buckets {
 990     my $self = shift;
 991     my ($tag, $md5, $exact) = @_;
 992
 993     my $trans_id = $self->_fileobj->transaction_id;
 994
 995     my @zero;
 996
 997     BUCKET:
 998     for ( my $i = 0; $i < $self->{max_buckets}; $i++ ) {
 999         my ($key, $subloc, $size, $transaction_id, $is_deleted) = $self->_get_key_subloc(
1000             $tag->{content}, $i,
1001         );
1002
1003         my @rv = ($subloc, $i * $self->{bucket_size}, $size, $is_deleted);
1004
1005         unless ( $subloc ) {
1006             if ( !$exact && @zero and $trans_id ) {
1007                 @rv = ($zero[2], $zero[0] * $self->{bucket_size},$zero[3],$is_deleted);
1008             }
1009             return @rv;
1010         }
1011
1012         next BUCKET if $key ne $md5;
1013
1014         # Save off the HEAD in case we need it.
1015         @zero = ($i,$key,$subloc,$size,$transaction_id,$is_deleted) if $transaction_id == 0;
1016
1017         next BUCKET if $transaction_id != $trans_id;
1018
1019         return @rv;
1020     }
1021
1022     return;
1023 }
1024
1025 sub _request_space {
1026     my $self = shift;
1027     my ($size) = @_;
1028
1029     my $loc = $self->_fileobj->{end};
1030     $self->_fileobj->{end} += $size;
1031
1032     return $loc;
1033 }
1034
1035 sub _release_space {
1036     my $self = shift;
1037     my ($size, $loc) = @_;
1038
1039     local($/,$\);
1040
1041     my $next_loc = 0;
1042
1043     my $fh = $self->_fh;
1044     seek( $fh, $loc + $self->_fileobj->{file_offset}, SEEK_SET );
1045     print( $fh SIG_FREE
1046         . pack($self->{long_pack}, $size )
1047         . pack($self->{long_pack}, $next_loc )
1048     );
1049
1050     return;
1051 }
1052
1053 sub _throw_error {
1054     die "DBM::Deep: $_[1]\n";
1055 }
1056
1057 1;
1058 __END__
1059
1060 # This will be added in later, after more refactoring is done. This is an early
1061 # attempt at refactoring on the physical level instead of the virtual level.
1062 sub _read_at {
1063     my $self = shift;
1064     my ($spot, $amount, $unpack) = @_;
1065
1066     local($/,$\);
1067
1068     my $fh = $self->_fh;
1069     seek( $fh, $spot + $self->_fileobj->{file_offset}, SEEK_SET );
1070
1071     my $buffer;
1072     my $bytes_read = read( $fh, $buffer, $amount );
1073
1074     if ( $unpack ) {
1075         $buffer = unpack( $unpack, $buffer );
1076     }
1077
1078     if ( wantarray ) {
1079         return ($buffer, $bytes_read);
1080     }
1081     else {
1082         return $buffer;
1083     }
1084 }
1085
1086 sub _print_at {
1087     my $self = shift;
1088     my ($spot, $data) = @_;
1089
1090     local($/,$\);
1091
1092     my $fh = $self->_fh;
1093     seek( $fh, $spot, SEEK_SET );
1094     print( $fh $data );
1095
1096     return;
1097 }
1098
1099 sub get_file_version {
1100     my $self = shift;
1101
1102     local($/,$\);
1103
1104     my $fh = $self->_fh;
1105
1106     seek( $fh, 13 + $self->_fileobj->{file_offset}, SEEK_SET );
1107     my $buffer;
1108     my $bytes_read = read( $fh, $buffer, 4 );
1109     unless ( $bytes_read == 4 ) {
1110         $self->_throw_error( "Cannot read file version" );
1111     }
1112
1113     return unpack( 'N', $buffer );
1114 }
1115
1116 sub write_file_version {
1117     my $self = shift;
1118     my ($new_version) = @_;
1119
1120     local($/,$\);
1121
1122     my $fh = $self->_fh;
1123
1124     seek( $fh, 13 + $self->_fileobj->{file_offset}, SEEK_SET );
1125     print( $fh pack( 'N', $new_version ) );
1126
1127     return;
1128 }
1129