[dbsrgits/DBM-Deep.git] / lib / DBM / Deep / File.pm

package DBM::Deep::File;

use 5.6.0;

use strict;
use warnings;

our $VERSION = q(0.99_03);

use Fcntl qw( :DEFAULT :flock :seek );

sub new {
    my $class = shift;
    my ($args) = @_;

    my $self = bless {
        audit_fh           => undef,
        audit_file         => undef,
        autobless          => 1,
        autoflush          => undef,
        end                => 0,
        fh                 => undef,
        file               => undef,
        file_offset        => 0,
        locking            => undef,
        locked             => 0,
        filter_store_key   => undef,
        filter_store_value => undef,
        filter_fetch_key   => undef,
        filter_fetch_value => undef,

        # These are values that are not expected to be passed in through
        # $args. They are here for documentation purposes.
        transaction_id     => 0,
        transaction_offset => 0,
        transaction_audit  => undef,
        base_db_obj        => undef,
    }, $class;

    # Grab the parameters we want to use
    foreach my $param ( keys %$self ) {
        next unless exists $args->{$param};
        $self->{$param} = $args->{$param};
    }

    if ( $self->{fh} && !$self->{file_offset} ) {
        $self->{file_offset} = tell( $self->{fh} );
    }

    $self->open unless $self->{fh};

    if ( $self->{audit_file} && !$self->{audit_fh} ) {
        my $flags = O_WRONLY | O_APPEND | O_CREAT;

        my $fh;
        sysopen( $fh, $self->{audit_file}, $flags )
            or die "Cannot open audit file '$self->{audit_file}' for read/write: $!";

        # Set the audit_fh to autoflush
        my $old = select $fh;
        $|=1;
        select $old;

        $self->{audit_fh} = $fh;
    }


    return $self;
}

sub set_db {
    my $self = shift;

    unless ( $self->{base_db_obj} ) {
        $self->{base_db_obj} = shift;
        Scalar::Util::weaken( $self->{base_db_obj} );
    }

    return;
}

sub open {
    my $self = shift;

    # Adding O_BINARY should remove the need for the binmode below. However,
    # I'm not going to remove it because I don't have the Win32 chops to be
    # absolutely certain everything will be ok.
    my $flags = O_RDWR | O_CREAT | O_BINARY;

    my $fh;
    sysopen( $fh, $self->{file}, $flags )
        or die "DBM::Deep: Cannot sysopen file '$self->{file}': $!\n";
    $self->{fh} = $fh;

    # Even though we use O_BINARY, better be safe than sorry.
    binmode $fh;

    if ($self->{autoflush}) {
        my $old = select $fh;
        $|=1;
        select $old;
    }

    return 1;
}

sub close {
    my $self = shift;

    if ( $self->{fh} ) {
        close $self->{fh};
        $self->{fh} = undef;
    }

    return 1;
}

sub set_inode {
    my $self = shift;

    unless ( $self->{inode} ) {
        my @stats = stat($self->{fh});
        $self->{inode} = $stats[1];
        $self->{end} = $stats[7];
    }

    return 1;
}

sub print_at {
    my $self = shift;
    my $loc  = shift;

    local ($/,$\);

    my $fh = $self->{fh};
    if ( defined $loc ) {
        seek( $fh, $loc + $self->{file_offset}, SEEK_SET );
    }

    print( $fh @_ );

    return 1;
}

sub read_at {
    my $self = shift;
    my ($loc, $size) = @_;
    print join(":",map{$_||''}caller) . " - read_at(@{[$loc || 'undef']}, $size)\n" if $::DEBUG;

    local ($/,$\);

    my $fh = $self->{fh};
    if ( defined $loc ) {
        seek( $fh, $loc + $self->{file_offset}, SEEK_SET );
    }

    my $buffer;
    read( $fh, $buffer, $size);

    return $buffer;
}

sub increment_pointer {
    my $self = shift;
    my ($size) = @_;

    if ( defined $size ) {
        seek( $self->{fh}, $size, SEEK_CUR );
    }

    return 1;
}

sub DESTROY {
    my $self = shift;
    return unless $self;

    $self->close;

    return;
}

sub request_space {
    my $self = shift;
    my ($size) = @_;

    #XXX Do I need to reset $self->{end} here? I need a testcase
    my $loc = $self->{end};
    $self->{end} += $size;

    return $loc;
}

#sub release_space {
#    my $self = shift;
#    my ($size, $loc) = @_;
#
#    local($/,$\);
#
#    my $next_loc = 0;
#
#    my $fh = $self->{fh};
#    seek( $fh, $loc + $self->{file_offset}, SEEK_SET );
#    print( $fh SIG_FREE
#        . pack($self->{long_pack}, $size )
#        . pack($self->{long_pack}, $next_loc )
#    );
#
#    return;
#}

##
# If db locking is set, flock() the db file.  If called multiple
# times before unlock(), then the same number of unlocks() must
# be called before the lock is released.
##
sub lock {
    my $self = shift;
    my ($obj, $type) = @_;

    #XXX This may not always be the correct thing to do
    $obj = $self->{base_db_obj} unless defined $obj;

    $type = LOCK_EX unless defined $type;

    if (!defined($self->{fh})) { return; }

    if ($self->{locking}) {
        if (!$self->{locked}) {
            flock($self->{fh}, $type);

            # refresh end counter in case file has changed size
            my @stats = stat($self->{fh});
            $self->{end} = $stats[7];

            # double-check file inode, in case another process
            # has optimize()d our file while we were waiting.
            if ($stats[1] != $self->{inode}) {
                $self->close;
                $self->open;

                #XXX This needs work
                $obj->{engine}->setup_fh( $obj );

                flock($self->{fh}, $type); # re-lock

                # This may not be necessary after re-opening
                $self->{end} = (stat($self->{fh}))[7]; # re-end
            }
        }
        $self->{locked}++;

        return 1;
    }

    return;
}

##
# If db locking is set, unlock the db file.  See note in lock()
# regarding calling lock() multiple times.
##
sub unlock {
    my $self = shift;

    if (!defined($self->{fh})) { return; }

    if ($self->{locking} && $self->{locked} > 0) {
        $self->{locked}--;
        if (!$self->{locked}) { flock($self->{fh}, LOCK_UN); }

        return 1;
    }

    return;
}

sub set_transaction_offset {
    my $self = shift;
    $self->{transaction_offset} = shift;
}

sub audit {
    my $self = shift;
    my ($string) = @_;

    if ( my $afh = $self->{audit_fh} ) {
        flock( $afh, LOCK_EX );

        if ( $string =~ /^#/ ) {
            print( $afh "$string " . localtime(time) . "\n" );
        }
        else {
            print( $afh "$string # " . localtime(time) . "\n" );
        }

        flock( $afh, LOCK_UN );
    }

    if ( $self->{transaction_audit} ) {
        push @{$self->{transaction_audit}}, $string;
    }

    return 1;
}

sub begin_transaction {
    my $self = shift;

    my $fh = $self->{fh};

    $self->lock;

    my $buffer = $self->read_at( $self->{transaction_offset}, 4 );
    my ($next, @trans) = unpack( 'C C C C C C C C C C C C C C C C', $buffer );

    $self->{transaction_id} = ++$next;

    die if $trans[-1] != 0;

    for ( my $i = 0; $i <= $#trans; $i++ ) {
        next if $trans[$i] != 0;
        $trans[$i] = $next;
        last;
    }

    $self->print_at(
        $self->{transaction_offset},
        pack( 'C C C C C C C C C C C C C C C C', $next, @trans),
    );

    $self->unlock;

    $self->{transaction_audit} = [];

    return $self->{transaction_id};
}

sub end_transaction {
    my $self = shift;

    my $fh = $self->{fh};

    $self->lock;

    my $buffer = $self->read_at( $self->{transaction_offset}, 4 );
    my ($next, @trans) = unpack( 'C C C C C C C C C C C C C C C C', $buffer );

    @trans = grep { $_ != $self->{transaction_id} } @trans;

    $self->print_at(
        $self->{transaction_offset},
        pack( 'C C C C C C C C C C C C C C C C', $next, @trans),
    );

    #XXX Need to free the space used by the current transaction

    $self->unlock;

    $self->{transaction_id} = 0;
    $self->{transaction_audit} = undef;

#    $self->{base_db_obj}->optimize;
#    $self->{inode} = undef;
#    $self->set_inode;

    return 1;
}

sub current_transactions {
    my $self = shift;

    my $fh = $self->{fh};

    $self->lock;

    my $buffer = $self->read_at( $self->{transaction_offset}, 4 );
    my ($next, @trans) = unpack( 'C C C C C C C C C C C C C C C C', $buffer );

    $self->unlock;

    return grep { $_ && $_ != $self->{transaction_id} } @trans;
}

sub transaction_id { return $_[0]->{transaction_id} }

sub commit_transaction {
    my $self = shift;

    my @audit = @{$self->{transaction_audit}};

    $self->end_transaction;

    {
        my $db = $self->{base_db_obj};
        for ( @audit ) {
            eval "$_;";
            warn "$_: $@\n" if $@;
        }
    }

    return 1;
}

1;
__END__
Commit	Line	Data
460b1067	1	package DBM::Deep::File;
	2
	3	use 5.6.0;
	4
	5	use strict;
	6	use warnings;
	7
c3aafc14	8	our $VERSION = q(0.99_03);
460b1067	9
86867f3a	10	use Fcntl qw( :DEFAULT :flock :seek );
460b1067	11
	12	sub new {
	13	my $class = shift;
	14	my ($args) = @_;
	15
	16	my $self = bless {
359a01ac	17	audit_fh => undef,
	18	audit_file => undef,
	19	autobless => 1,
460b1067	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,
28394a1a	31
359a01ac	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,
633df1fd	36	transaction_audit => undef,
359a01ac	37	base_db_obj => undef,
460b1067	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
359a01ac	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
460b1067	68	return $self;
	69	}
	70
359a01ac	71	sub set_db {
25c7c8d6	72	my $self = shift;
633df1fd	73
25c7c8d6	74	unless ( $self->{base_db_obj} ) {
	75	$self->{base_db_obj} = shift;
	76	Scalar::Util::weaken( $self->{base_db_obj} );
359a01ac	77	}
25c7c8d6	78
25c7c8d6	79	return;
359a01ac	80	}
359a01ac	81
460b1067	82	sub open {
	83	my $self = shift;
	84
633df1fd	85	# Adding O_BINARY should remove the need for the binmode below. However,
460b1067	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
7dcefff3	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
019404df	130	sub print_at {
	131	my $self = shift;
	132	my $loc = shift;
	133
	134	local ($/,$\);
	135
	136	my $fh = $self->{fh};
7dcefff3	137	if ( defined $loc ) {
	138	seek( $fh, $loc + $self->{file_offset}, SEEK_SET );
	139	}
	140
019404df	141	print( $fh @_ );
	142
	143	return 1;
	144	}
	145
7dcefff3	146	sub read_at {
	147	my $self = shift;
	148	my ($loc, $size) = @_;
9bc79bb6	149	print join(":",map{$_\|\|''}caller) . " - read_at(@{[$loc \|\| 'undef']}, $size)\n" if $::DEBUG;
7dcefff3	150
	151	local ($/,$\);
	152
	153	my $fh = $self->{fh};
	154	if ( defined $loc ) {
	155	seek( $fh, $loc + $self->{file_offset}, SEEK_SET );
	156	}
	157
	158	my $buffer;
	159	read( $fh, $buffer, $size);
	160
	161	return $buffer;
	162	}
	163
	164	sub increment_pointer {
	165	my $self = shift;
	166	my ($size) = @_;
	167
	168	if ( defined $size ) {
	169	seek( $self->{fh}, $size, SEEK_CUR );
	170	}
	171
	172	return 1;
	173	}
	174
460b1067	175	sub DESTROY {
	176	my $self = shift;
	177	return unless $self;
	178
	179	$self->close;
	180
	181	return;
	182	}
	183
019404df	184	sub request_space {
	185	my $self = shift;
	186	my ($size) = @_;
	187
7dcefff3	188	#XXX Do I need to reset $self->{end} here? I need a testcase
019404df	189	my $loc = $self->{end};
	190	$self->{end} += $size;
	191
	192	return $loc;
	193	}
	194
	195	#sub release_space {
	196	# my $self = shift;
	197	# my ($size, $loc) = @_;
	198	#
	199	# local($/,$\);
	200	#
	201	# my $next_loc = 0;
	202	#
	203	# my $fh = $self->{fh};
	204	# seek( $fh, $loc + $self->{file_offset}, SEEK_SET );
	205	# print( $fh SIG_FREE
	206	# . pack($self->{long_pack}, $size )
	207	# . pack($self->{long_pack}, $next_loc )
	208	# );
	209	#
	210	# return;
	211	#}
	212
15ba72cc	213	##
	214	# If db locking is set, flock() the db file. If called multiple
	215	# times before unlock(), then the same number of unlocks() must
	216	# be called before the lock is released.
	217	##
	218	sub lock {
	219	my $self = shift;
	220	my ($obj, $type) = @_;
42717e46	221
	222	#XXX This may not always be the correct thing to do
	223	$obj = $self->{base_db_obj} unless defined $obj;
	224
15ba72cc	225	$type = LOCK_EX unless defined $type;
	226
	227	if (!defined($self->{fh})) { return; }
	228
	229	if ($self->{locking}) {
	230	if (!$self->{locked}) {
	231	flock($self->{fh}, $type);
	232
	233	# refresh end counter in case file has changed size
	234	my @stats = stat($self->{fh});
	235	$self->{end} = $stats[7];
	236
	237	# double-check file inode, in case another process
	238	# has optimize()d our file while we were waiting.
	239	if ($stats[1] != $self->{inode}) {
	240	$self->close;
	241	$self->open;
	242
	243	#XXX This needs work
	244	$obj->{engine}->setup_fh( $obj );
	245
	246	flock($self->{fh}, $type); # re-lock
	247
	248	# This may not be necessary after re-opening
	249	$self->{end} = (stat($self->{fh}))[7]; # re-end
	250	}
	251	}
	252	$self->{locked}++;
	253
	254	return 1;
	255	}
	256
	257	return;
	258	}
	259
	260	##
	261	# If db locking is set, unlock the db file. See note in lock()
	262	# regarding calling lock() multiple times.
	263	##
	264	sub unlock {
	265	my $self = shift;
	266
	267	if (!defined($self->{fh})) { return; }
	268
	269	if ($self->{locking} && $self->{locked} > 0) {
	270	$self->{locked}--;
	271	if (!$self->{locked}) { flock($self->{fh}, LOCK_UN); }
	272
	273	return 1;
	274	}
	275
	276	return;
	277	}
	278
	279	sub set_transaction_offset {
	280	my $self = shift;
	281	$self->{transaction_offset} = shift;
	282	}
	283
aa83bc1e	284	sub audit {
aa83bc1e	285	my $self = shift;
25c7c8d6	286	my ($string) = @_;
aa83bc1e	287
aa83bc1e	288	if ( my $afh = $self->{audit_fh} ) {
aa83bc1e	289	flock( $afh, LOCK_EX );
	290
	291	if ( $string =~ /^#/ ) {
	292	print( $afh "$string " . localtime(time) . "\n" );
	293	}
	294	else {
	295	print( $afh "$string # " . localtime(time) . "\n" );
	296	}
	297
	298	flock( $afh, LOCK_UN );
	299	}
	300
633df1fd	301	if ( $self->{transaction_audit} ) {
633df1fd	302	push @{$self->{transaction_audit}}, $string;
25c7c8d6	303	}
25c7c8d6	304
aa83bc1e	305	return 1;
	306	}
	307
28394a1a	308	sub begin_transaction {
	309	my $self = shift;
	310
15ba72cc	311	my $fh = $self->{fh};
15ba72cc	312
20b7f047	313	$self->lock;
20b7f047	314
633df1fd	315	my $buffer = $self->read_at( $self->{transaction_offset}, 4 );
eff6a245	316	my ($next, @trans) = unpack( 'C C C C C C C C C C C C C C C C', $buffer );
633df1fd	317
633df1fd	318	$self->{transaction_id} = ++$next;
20b7f047	319
633df1fd	320	die if $trans[-1] != 0;
	321
	322	for ( my $i = 0; $i <= $#trans; $i++ ) {
	323	next if $trans[$i] != 0;
	324	$trans[$i] = $next;
20b7f047	325	last;
	326	}
	327
633df1fd	328	$self->print_at(
633df1fd	329	$self->{transaction_offset},
eff6a245	330	pack( 'C C C C C C C C C C C C C C C C', $next, @trans),
633df1fd	331	);
15ba72cc	332
20b7f047	333	$self->unlock;
20b7f047	334
633df1fd	335	$self->{transaction_audit} = [];
25c7c8d6	336
20b7f047	337	return $self->{transaction_id};
28394a1a	338	}
	339
	340	sub end_transaction {
	341	my $self = shift;
	342
20b7f047	343	my $fh = $self->{fh};
	344
	345	$self->lock;
	346
633df1fd	347	my $buffer = $self->read_at( $self->{transaction_offset}, 4 );
eff6a245	348	my ($next, @trans) = unpack( 'C C C C C C C C C C C C C C C C', $buffer );
633df1fd	349
633df1fd	350	@trans = grep { $_ != $self->{transaction_id} } @trans;
20b7f047	351
633df1fd	352	$self->print_at(
633df1fd	353	$self->{transaction_offset},
eff6a245	354	pack( 'C C C C C C C C C C C C C C C C', $next, @trans),
633df1fd	355	);
20b7f047	356
633df1fd	357	#XXX Need to free the space used by the current transaction
20b7f047	358
20b7f047	359	$self->unlock;
15ba72cc	360
28394a1a	361	$self->{transaction_id} = 0;
633df1fd	362	$self->{transaction_audit} = undef;
	363
	364	# $self->{base_db_obj}->optimize;
	365	# $self->{inode} = undef;
	366	# $self->set_inode;
25c7c8d6	367
25c7c8d6	368	return 1;
28394a1a	369	}
28394a1a	370
20b7f047	371	sub current_transactions {
28394a1a	372	my $self = shift;
28394a1a	373
20b7f047	374	my $fh = $self->{fh};
	375
	376	$self->lock;
	377
633df1fd	378	my $buffer = $self->read_at( $self->{transaction_offset}, 4 );
eff6a245	379	my ($next, @trans) = unpack( 'C C C C C C C C C C C C C C C C', $buffer );
20b7f047	380
	381	$self->unlock;
	382
633df1fd	383	return grep { $_ && $_ != $self->{transaction_id} } @trans;
28394a1a	384	}
28394a1a	385
20b7f047	386	sub transaction_id { return $_[0]->{transaction_id} }
20b7f047	387
25c7c8d6	388	sub commit_transaction {
	389	my $self = shift;
	390
633df1fd	391	my @audit = @{$self->{transaction_audit}};
25c7c8d6	392
	393	$self->end_transaction;
	394
	395	{
	396	my $db = $self->{base_db_obj};
	397	for ( @audit ) {
	398	eval "$_;";
	399	warn "$_: $@\n" if $@;
	400	}
	401	}
	402
	403	return 1;
	404	}
28394a1a	405
460b1067	406	1;
	407	__END__
	408