lib/DBM/Deep/File.pm

   1 package DBM::Deep::File;
   2
   3 use 5.6.0;
   4
   5 use strict;
   6 use warnings;
   7
   8 our $VERSION = q(0.99_01);
   9
  10 use Fcntl qw( :DEFAULT :flock :seek );
  11
  12 sub new {
  13     my $class = shift;
  14     my ($args) = @_;
  15
  16     my $self = bless {
  17         audit_fh           => undef,
  18         audit_file         => undef,
  19         autobless          => 1,
  20         autoflush          => undef,
  21         end                => 0,
  22         fh                 => undef,
  23         file               => undef,
  24         file_offset        => 0,
  25         locking            => undef,
  26         locked             => 0,
  27         filter_store_key   => undef,
  28         filter_store_value => undef,
  29         filter_fetch_key   => undef,
  30         filter_fetch_value => undef,
  31
  32         # These are values that are not expected to be passed in through
  33         # $args. They are here for documentation purposes.
  34         transaction_id     => 0,
  35         transaction_offset => 0,
  36         transaction_audit  => undef,
  37         base_db_obj        => undef,
  38     }, $class;
  39
  40     # Grab the parameters we want to use
  41     foreach my $param ( keys %$self ) {
  42         next unless exists $args->{$param};
  43         $self->{$param} = $args->{$param};
  44     }
  45
  46     if ( $self->{fh} && !$self->{file_offset} ) {
  47         $self->{file_offset} = tell( $self->{fh} );
  48     }
  49
  50     $self->open unless $self->{fh};
  51
  52     if ( $self->{audit_file} && !$self->{audit_fh} ) {
  53         my $flags = O_WRONLY | O_APPEND | O_CREAT;
  54
  55         my $fh;
  56         sysopen( $fh, $self->{audit_file}, $flags )
  57             or die "Cannot open audit file '$self->{audit_file}' for read/write: $!";
  58
  59         # Set the audit_fh to autoflush
  60         my $old = select $fh;
  61         $|=1;
  62         select $old;
  63
  64         $self->{audit_fh} = $fh;
  65     }
  66
  67
  68     return $self;
  69 }
  70
  71 sub set_db {
  72     my $self = shift;
  73
  74     unless ( $self->{base_db_obj} ) {
  75         $self->{base_db_obj} = shift;
  76         Scalar::Util::weaken( $self->{base_db_obj} );
  77     }
  78
  79     return;
  80 }
  81
  82 sub open {
  83     my $self = shift;
  84
  85     # Adding O_BINARY should remove the need for the binmode below. However,
  86     # I'm not going to remove it because I don't have the Win32 chops to be
  87     # absolutely certain everything will be ok.
  88     my $flags = O_RDWR | O_CREAT | O_BINARY;
  89
  90     my $fh;
  91     sysopen( $fh, $self->{file}, $flags )
  92         or die "DBM::Deep: Cannot sysopen file '$self->{file}': $!\n";
  93     $self->{fh} = $fh;
  94
  95     # Even though we use O_BINARY, better be safe than sorry.
  96     binmode $fh;
  97
  98     if ($self->{autoflush}) {
  99         my $old = select $fh;
 100         $|=1;
 101         select $old;
 102     }
 103
 104     return 1;
 105 }
 106
 107 sub close {
 108     my $self = shift;
 109
 110     if ( $self->{fh} ) {
 111         close $self->{fh};
 112         $self->{fh} = undef;
 113     }
 114
 115     return 1;
 116 }
 117
 118 sub set_inode {
 119     my $self = shift;
 120
 121     unless ( $self->{inode} ) {
 122         my @stats = stat($self->{fh});
 123         $self->{inode} = $stats[1];
 124         $self->{end} = $stats[7];
 125     }
 126
 127     return 1;
 128 }
 129
 130 sub print_at {
 131     my $self = shift;
 132     my $loc  = shift;
 133
 134     local ($/,$\);
 135
 136     my $fh = $self->{fh};
 137     if ( defined $loc ) {
 138         seek( $fh, $loc + $self->{file_offset}, SEEK_SET );
 139     }
 140
 141     print( $fh @_ );
 142
 143     return 1;
 144 }
 145
 146 sub read_at {
 147     my $self = shift;
 148     my ($loc, $size) = @_;
 149
 150     local ($/,$\);
 151
 152     my $fh = $self->{fh};
 153     if ( defined $loc ) {
 154         seek( $fh, $loc + $self->{file_offset}, SEEK_SET );
 155     }
 156
 157     my $buffer;
 158     read( $fh, $buffer, $size);
 159
 160     return $buffer;
 161 }
 162
 163 sub increment_pointer {
 164     my $self = shift;
 165     my ($size) = @_;
 166
 167     if ( defined $size ) {
 168         seek( $self->{fh}, $size, SEEK_CUR );
 169     }
 170
 171     return 1;
 172 }
 173
 174 sub DESTROY {
 175     my $self = shift;
 176     return unless $self;
 177
 178     $self->close;
 179
 180     return;
 181 }
 182
 183 sub request_space {
 184     my $self = shift;
 185     my ($size) = @_;
 186
 187     #XXX Do I need to reset $self->{end} here? I need a testcase
 188     my $loc = $self->{end};
 189     $self->{end} += $size;
 190
 191     return $loc;
 192 }
 193
 194 #sub release_space {
 195 #    my $self = shift;
 196 #    my ($size, $loc) = @_;
 197 #
 198 #    local($/,$\);
 199 #
 200 #    my $next_loc = 0;
 201 #
 202 #    my $fh = $self->{fh};
 203 #    seek( $fh, $loc + $self->{file_offset}, SEEK_SET );
 204 #    print( $fh SIG_FREE
 205 #        . pack($self->{long_pack}, $size )
 206 #        . pack($self->{long_pack}, $next_loc )
 207 #    );
 208 #
 209 #    return;
 210 #}
 211
 212 ##
 213 # If db locking is set, flock() the db file.  If called multiple
 214 # times before unlock(), then the same number of unlocks() must
 215 # be called before the lock is released.
 216 ##
 217 sub lock {
 218     my $self = shift;
 219     my ($obj, $type) = @_;
 220
 221     #XXX This may not always be the correct thing to do
 222     $obj = $self->{base_db_obj} unless defined $obj;
 223
 224     $type = LOCK_EX unless defined $type;
 225
 226     if (!defined($self->{fh})) { return; }
 227
 228     if ($self->{locking}) {
 229         if (!$self->{locked}) {
 230             flock($self->{fh}, $type);
 231
 232             # refresh end counter in case file has changed size
 233             my @stats = stat($self->{fh});
 234             $self->{end} = $stats[7];
 235
 236             # double-check file inode, in case another process
 237             # has optimize()d our file while we were waiting.
 238             if ($stats[1] != $self->{inode}) {
 239                 $self->close;
 240                 $self->open;
 241
 242                 #XXX This needs work
 243                 $obj->{engine}->setup_fh( $obj );
 244
 245                 flock($self->{fh}, $type); # re-lock
 246
 247                 # This may not be necessary after re-opening
 248                 $self->{end} = (stat($self->{fh}))[7]; # re-end
 249             }
 250         }
 251         $self->{locked}++;
 252
 253         return 1;
 254     }
 255
 256     return;
 257 }
 258
 259 ##
 260 # If db locking is set, unlock the db file.  See note in lock()
 261 # regarding calling lock() multiple times.
 262 ##
 263 sub unlock {
 264     my $self = shift;
 265
 266     if (!defined($self->{fh})) { return; }
 267
 268     if ($self->{locking} && $self->{locked} > 0) {
 269         $self->{locked}--;
 270         if (!$self->{locked}) { flock($self->{fh}, LOCK_UN); }
 271
 272         return 1;
 273     }
 274
 275     return;
 276 }
 277
 278 sub set_transaction_offset {
 279     my $self = shift;
 280     $self->{transaction_offset} = shift;
 281 }
 282
 283 sub audit {
 284     my $self = shift;
 285     my ($string) = @_;
 286
 287     if ( my $afh = $self->{audit_fh} ) {
 288         flock( $afh, LOCK_EX );
 289
 290         if ( $string =~ /^#/ ) {
 291             print( $afh "$string " . localtime(time) . "\n" );
 292         }
 293         else {
 294             print( $afh "$string # " . localtime(time) . "\n" );
 295         }
 296
 297         flock( $afh, LOCK_UN );
 298     }
 299
 300     if ( $self->{transaction_audit} ) {
 301         push @{$self->{transaction_audit}}, $string;
 302     }
 303
 304     return 1;
 305 }
 306
 307 sub begin_transaction {
 308     my $self = shift;
 309
 310     my $fh = $self->{fh};
 311
 312     $self->lock;
 313
 314     my $buffer = $self->read_at( $self->{transaction_offset}, 4 );
 315     my ($next, @trans) = unpack( 'C C C C C C C C C C C C C C C C', $buffer );
 316
 317     $self->{transaction_id} = ++$next;
 318
 319     die if $trans[-1] != 0;
 320
 321     for ( my $i = 0; $i <= $#trans; $i++ ) {
 322         next if $trans[$i] != 0;
 323         $trans[$i] = $next;
 324         last;
 325     }
 326
 327     $self->print_at(
 328         $self->{transaction_offset},
 329         pack( 'C C C C C C C C C C C C C C C C', $next, @trans),
 330     );
 331
 332     $self->unlock;
 333
 334     $self->{transaction_audit} = [];
 335
 336     return $self->{transaction_id};
 337 }
 338
 339 sub end_transaction {
 340     my $self = shift;
 341
 342     my $fh = $self->{fh};
 343
 344     $self->lock;
 345
 346     my $buffer = $self->read_at( $self->{transaction_offset}, 4 );
 347     my ($next, @trans) = unpack( 'C C C C C C C C C C C C C C C C', $buffer );
 348
 349     @trans = grep { $_ != $self->{transaction_id} } @trans;
 350
 351     $self->print_at(
 352         $self->{transaction_offset},
 353         pack( 'C C C C C C C C C C C C C C C C', $next, @trans),
 354     );
 355
 356     #XXX Need to free the space used by the current transaction
 357
 358     $self->unlock;
 359
 360     $self->{transaction_id} = 0;
 361     $self->{transaction_audit} = undef;
 362
 363 #    $self->{base_db_obj}->optimize;
 364 #    $self->{inode} = undef;
 365 #    $self->set_inode;
 366
 367     return 1;
 368 }
 369
 370 sub current_transactions {
 371     my $self = shift;
 372
 373     my $fh = $self->{fh};
 374
 375     $self->lock;
 376
 377     my $buffer = $self->read_at( $self->{transaction_offset}, 4 );
 378     my ($next, @trans) = unpack( 'C C C C C C C C C C C C C C C C', $buffer );
 379
 380     $self->unlock;
 381
 382     return grep { $_ && $_ != $self->{transaction_id} } @trans;
 383 }
 384
 385 sub transaction_id { return $_[0]->{transaction_id} }
 386
 387 sub commit_transaction {
 388     my $self = shift;
 389
 390     my @audit = @{$self->{transaction_audit}};
 391
 392     $self->end_transaction;
 393
 394     {
 395         my $db = $self->{base_db_obj};
 396         for ( @audit ) {
 397             eval "$_;";
 398             warn "$_: $@\n" if $@;
 399         }
 400     }
 401
 402     return 1;
 403 }
 404
 405 1;
 406 __END__
 407