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 use Fcntl qw( :DEFAULT :flock :seek );
   9
  10 our $VERSION = '0.01';
  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     $type = LOCK_EX unless defined $type;
 221
 222     if (!defined($self->{fh})) { return; }
 223
 224     if ($self->{locking}) {
 225         if (!$self->{locked}) {
 226             flock($self->{fh}, $type);
 227
 228             # refresh end counter in case file has changed size
 229             my @stats = stat($self->{fh});
 230             $self->{end} = $stats[7];
 231
 232             # double-check file inode, in case another process
 233             # has optimize()d our file while we were waiting.
 234             if ($stats[1] != $self->{inode}) {
 235                 $self->close;
 236                 $self->open;
 237
 238                 #XXX This needs work
 239                 $obj->{engine}->setup_fh( $obj );
 240
 241                 flock($self->{fh}, $type); # re-lock
 242
 243                 # This may not be necessary after re-opening
 244                 $self->{end} = (stat($self->{fh}))[7]; # re-end
 245             }
 246         }
 247         $self->{locked}++;
 248
 249         return 1;
 250     }
 251
 252     return;
 253 }
 254
 255 ##
 256 # If db locking is set, unlock the db file.  See note in lock()
 257 # regarding calling lock() multiple times.
 258 ##
 259 sub unlock {
 260     my $self = shift;
 261
 262     if (!defined($self->{fh})) { return; }
 263
 264     if ($self->{locking} && $self->{locked} > 0) {
 265         $self->{locked}--;
 266         if (!$self->{locked}) { flock($self->{fh}, LOCK_UN); }
 267
 268         return 1;
 269     }
 270
 271     return;
 272 }
 273
 274 sub set_transaction_offset {
 275     my $self = shift;
 276     $self->{transaction_offset} = shift;
 277 }
 278
 279 sub audit {
 280     my $self = shift;
 281     my ($string) = @_;
 282
 283     if ( my $afh = $self->{audit_fh} ) {
 284         flock( $afh, LOCK_EX );
 285
 286         if ( $string =~ /^#/ ) {
 287             print( $afh "$string " . localtime(time) . "\n" );
 288         }
 289         else {
 290             print( $afh "$string # " . localtime(time) . "\n" );
 291         }
 292
 293         flock( $afh, LOCK_UN );
 294     }
 295
 296     if ( $self->{transaction_audit} ) {
 297         push @{$self->{transaction_audit}}, $string;
 298     }
 299
 300     return 1;
 301 }
 302
 303 sub begin_transaction {
 304     my $self = shift;
 305
 306     my $fh = $self->{fh};
 307
 308     $self->lock;
 309
 310     my $buffer = $self->read_at( $self->{transaction_offset}, 4 );
 311     my ($next, @trans) = unpack( 'C C C C', $buffer );
 312
 313     $self->{transaction_id} = ++$next;
 314
 315     die if $trans[-1] != 0;
 316
 317     for ( my $i = 0; $i <= $#trans; $i++ ) {
 318         next if $trans[$i] != 0;
 319         $trans[$i] = $next;
 320         last;
 321     }
 322
 323     $self->print_at(
 324         $self->{transaction_offset},
 325         pack( 'C C C C', $next, @trans),
 326     );
 327
 328     $self->unlock;
 329
 330     $self->{transaction_audit} = [];
 331
 332     return $self->{transaction_id};
 333 }
 334
 335 sub end_transaction {
 336     my $self = shift;
 337
 338     my $fh = $self->{fh};
 339
 340     $self->lock;
 341
 342     my $buffer = $self->read_at( $self->{transaction_offset}, 4 );
 343     my ($next, @trans) = unpack( 'C C C C', $buffer );
 344
 345     @trans = grep { $_ != $self->{transaction_id} } @trans;
 346
 347     $self->print_at(
 348         $self->{transaction_offset},
 349         pack( 'C C C C', $next, @trans),
 350     );
 351
 352     #XXX Need to free the space used by the current transaction
 353
 354     $self->unlock;
 355
 356     $self->{transaction_id} = 0;
 357     $self->{transaction_audit} = undef;
 358
 359 #    $self->{base_db_obj}->optimize;
 360 #    $self->{inode} = undef;
 361 #    $self->set_inode;
 362
 363     return 1;
 364 }
 365
 366 sub current_transactions {
 367     my $self = shift;
 368
 369     my $fh = $self->{fh};
 370
 371     $self->lock;
 372
 373     my $buffer = $self->read_at( $self->{transaction_offset}, 4 );
 374     my ($next, @trans) = unpack( 'C C C C', $buffer );
 375
 376     $self->unlock;
 377
 378     return grep { $_ && $_ != $self->{transaction_id} } @trans;
 379 }
 380
 381 sub transaction_id { return $_[0]->{transaction_id} }
 382
 383 sub commit_transaction {
 384     my $self = shift;
 385
 386     my @audit = @{$self->{transaction_audit}};
 387
 388     $self->end_transaction;
 389
 390     {
 391         my $db = $self->{base_db_obj};
 392         for ( @audit ) {
 393             eval "$_;";
 394             warn "$_: $@\n" if $@;
 395         }
 396     }
 397
 398     return 1;
 399 }
 400
 401 1;
 402 __END__
 403