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