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