[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_01);

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) = @_;

    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
86867f3a	8	our $VERSION = q(0.99_01);
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) = @_;
	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
460b1067	174	sub DESTROY {
	175	my $self = shift;
	176	return unless $self;
	177
	178	$self->close;
	179
	180	return;
	181	}
	182
019404df	183	sub request_space {
	184	my $self = shift;
	185	my ($size) = @_;
	186
7dcefff3	187	#XXX Do I need to reset $self->{end} here? I need a testcase
019404df	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
15ba72cc	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) = @_;
42717e46	220
	221	#XXX This may not always be the correct thing to do
	222	$obj = $self->{base_db_obj} unless defined $obj;
	223
15ba72cc	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
aa83bc1e	283	sub audit {
aa83bc1e	284	my $self = shift;
25c7c8d6	285	my ($string) = @_;
aa83bc1e	286
aa83bc1e	287	if ( my $afh = $self->{audit_fh} ) {
aa83bc1e	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
633df1fd	300	if ( $self->{transaction_audit} ) {
633df1fd	301	push @{$self->{transaction_audit}}, $string;
25c7c8d6	302	}
25c7c8d6	303
aa83bc1e	304	return 1;
	305	}
	306
28394a1a	307	sub begin_transaction {
	308	my $self = shift;
	309
15ba72cc	310	my $fh = $self->{fh};
15ba72cc	311
20b7f047	312	$self->lock;
20b7f047	313
633df1fd	314	my $buffer = $self->read_at( $self->{transaction_offset}, 4 );
15d754dd	315	my ($next, @trans) = unpack( 'C C C C C C C C C C C C C C C C', $buffer );
633df1fd	316
633df1fd	317	$self->{transaction_id} = ++$next;
20b7f047	318
633df1fd	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;
20b7f047	324	last;
	325	}
	326
633df1fd	327	$self->print_at(
633df1fd	328	$self->{transaction_offset},
15d754dd	329	pack( 'C C C C C C C C C C C C C C C C', $next, @trans),
633df1fd	330	);
15ba72cc	331
20b7f047	332	$self->unlock;
20b7f047	333
633df1fd	334	$self->{transaction_audit} = [];
25c7c8d6	335
20b7f047	336	return $self->{transaction_id};
28394a1a	337	}
	338
	339	sub end_transaction {
	340	my $self = shift;
	341
20b7f047	342	my $fh = $self->{fh};
	343
	344	$self->lock;
	345
633df1fd	346	my $buffer = $self->read_at( $self->{transaction_offset}, 4 );
15d754dd	347	my ($next, @trans) = unpack( 'C C C C C C C C C C C C C C C C', $buffer );
633df1fd	348
633df1fd	349	@trans = grep { $_ != $self->{transaction_id} } @trans;
20b7f047	350
633df1fd	351	$self->print_at(
633df1fd	352	$self->{transaction_offset},
15d754dd	353	pack( 'C C C C C C C C C C C C C C C C', $next, @trans),
633df1fd	354	);
20b7f047	355
633df1fd	356	#XXX Need to free the space used by the current transaction
20b7f047	357
20b7f047	358	$self->unlock;
15ba72cc	359
28394a1a	360	$self->{transaction_id} = 0;
633df1fd	361	$self->{transaction_audit} = undef;
	362
	363	# $self->{base_db_obj}->optimize;
	364	# $self->{inode} = undef;
	365	# $self->set_inode;
25c7c8d6	366
25c7c8d6	367	return 1;
28394a1a	368	}
28394a1a	369
20b7f047	370	sub current_transactions {
28394a1a	371	my $self = shift;
28394a1a	372
20b7f047	373	my $fh = $self->{fh};
	374
	375	$self->lock;
	376
633df1fd	377	my $buffer = $self->read_at( $self->{transaction_offset}, 4 );
15d754dd	378	my ($next, @trans) = unpack( 'C C C C C C C C C C C C C C C C', $buffer );
20b7f047	379
	380	$self->unlock;
	381
633df1fd	382	return grep { $_ && $_ != $self->{transaction_id} } @trans;
28394a1a	383	}
28394a1a	384
20b7f047	385	sub transaction_id { return $_[0]->{transaction_id} }
20b7f047	386
25c7c8d6	387	sub commit_transaction {
	388	my $self = shift;
	389
633df1fd	390	my @audit = @{$self->{transaction_audit}};
25c7c8d6	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	}
28394a1a	404
460b1067	405	1;
	406	__END__
	407