view release on metacpan or  search on metacpan

lib/DBM/Deep/Engine/File.pm  view on Meta::CPAN

    return 1;
}

sub begin_work {
    my $self = shift;
    my ($obj) = @_;

    unless ($self->supports('transactions')) {
        DBM::Deep->_throw_error( "Cannot begin_work unless transactions are supported" );
    }

    if ( $self->trans_id ) {
        DBM::Deep->_throw_error( "Cannot begin_work within an active transaction" );
    }

    my @slots = $self->read_txn_slots;
    my $found;
    for my $i ( 0 .. $self->num_txns-2 ) {
        next if $slots[$i];

        $slots[$i] = 1;
        $self->set_trans_id( $i + 1 );
        $found = 1;
        last;
    }
    unless ( $found ) {
        DBM::Deep->_throw_error( "Cannot allocate transaction ID" );
    }
    $self->write_txn_slots( @slots );

    if ( !$self->trans_id ) {
        DBM::Deep->_throw_error( "Cannot begin_work - no available transactions" );
    }

    return;
}

sub rollback {
    my $self = shift;
    my ($obj) = @_;

    unless ($self->supports('transactions')) {
        DBM::Deep->_throw_error( "Cannot rollback unless transactions are supported" );
    }

    if ( !$self->trans_id ) {
        DBM::Deep->_throw_error( "Cannot rollback without an active transaction" );
    }

    # Each entry is the file location for a bucket that has a modification for
    # this transaction. The entries need to be expunged.
    foreach my $entry (@{ $self->get_entries } ) {
        # Remove the entry here
        my $read_loc = $entry
          + $self->hash_size
          + $self->byte_size
          + $self->byte_size
          + ($self->trans_id - 1) * ( $self->byte_size + $STALE_SIZE );

        my $data_loc = $self->storage->read_at( $read_loc, $self->byte_size );
        $data_loc = unpack( $StP{$self->byte_size}, $data_loc );
        $self->storage->print_at( $read_loc, pack( $StP{$self->byte_size}, 0 ) );

        if ( $data_loc > 1 ) {
            $self->load_sector( $data_loc )->free;
        }
    }

    $self->clear_entries;

    my @slots = $self->read_txn_slots;
    $slots[$self->trans_id-1] = 0;
    $self->write_txn_slots( @slots );
    $self->inc_txn_staleness_counter( $self->trans_id );
    $self->set_trans_id( 0 );

    return 1;
}

sub commit {
    my $self = shift;
    my ($obj) = @_;

    unless ($self->supports('transactions')) {
        DBM::Deep->_throw_error( "Cannot commit unless transactions are supported" );
    }

    if ( !$self->trans_id ) {
        DBM::Deep->_throw_error( "Cannot commit without an active transaction" );
    }

    foreach my $entry (@{ $self->get_entries } ) {
        # Overwrite the entry in head with the entry in trans_id
        my $base = $entry
          + $self->hash_size
          + $self->byte_size;

        my $head_loc = $self->storage->read_at( $base, $self->byte_size );
        $head_loc = unpack( $StP{$self->byte_size}, $head_loc );

        my $spot = $base + $self->byte_size + ($self->trans_id - 1) * ( $self->byte_size + $STALE_SIZE );
        my $trans_loc = $self->storage->read_at(
            $spot, $self->byte_size,
        );

        $self->storage->print_at( $base, $trans_loc );
        $self->storage->print_at(
            $spot,
            pack( $StP{$self->byte_size} . ' ' . $StP{$STALE_SIZE}, (0) x 2 ),
        );

        if ( $head_loc > 1 ) {
            $self->load_sector( $head_loc )->free;
        }
    }

    $self->clear_entries;

    my @slots = $self->read_txn_slots;
    $slots[$self->trans_id-1] = 0;
    $self->write_txn_slots( @slots );
    $self->inc_txn_staleness_counter( $self->trans_id );
    $self->set_trans_id( 0 );

    return 1;
}

=head1 INTERNAL METHODS

The following methods are internal-use-only to DBM::Deep::Engine::File.

=cut

=head2 read_txn_slots()

This takes no arguments.

This will return an array with a 1 or 0 in each slot. Each spot represents one
available transaction. If the slot is 1, that transaction is taken. If it is 0,
the transaction is available.

=cut

sub read_txn_slots {
    my $self = shift;
    my $bl = $self->txn_bitfield_len;
    my $num_bits = $bl * 8;
    return split '', unpack( 'b'.$num_bits,
        $self->storage->read_at(
            $self->trans_loc, $bl,
        )
    );
}

=head2 write_txn_slots( @slots )

This takes an array of 1's and 0's. This array represents the transaction slots
returned by L</read_txn_slots()>. In other words, the following is true:

  @x = read_txn_slots( write_txn_slots( @x ) );

(With the obviously missing object referents added back in.)

=cut

sub write_txn_slots {
    my $self = shift;
    my $num_bits = $self->txn_bitfield_len * 8;
    $self->storage->print_at( $self->trans_loc,
        pack( 'b'.$num_bits, join('', @_) ),
    );
}

=head2 get_running_txn_ids()

This takes no arguments.

This will return an array of taken transaction IDs. This wraps L</read_txn_slots()>.

=cut

sub get_running_txn_ids {
    my $self = shift;
    my @transactions = $self->read_txn_slots;
    my @trans_ids = map { $_+1 } grep { $transactions[$_] } 0 .. $#transactions;
}

=head2 get_txn_staleness_counter( $trans_id )

This will return the staleness counter for the given transaction ID. Please see
L<DBM::Deep::Engine/STALENESS> for more information.

=cut

sub get_txn_staleness_counter {
    my $self = shift;
    my ($trans_id) = @_;

    # Hardcode staleness of 0 for the HEAD
    return 0 unless $trans_id;

    return unpack( $StP{$STALE_SIZE},
        $self->storage->read_at(
            $self->trans_loc + $self->txn_bitfield_len + $STALE_SIZE * ($trans_id - 1),
            $STALE_SIZE,
        )
    );
}

=head2 inc_txn_staleness_counter( $trans_id )

This will increment the staleness counter for the given transaction ID. Please see
L<DBM::Deep::Engine/STALENESS> for more information.

=cut

sub inc_txn_staleness_counter {
    my $self = shift;
    my ($trans_id) = @_;

    # Hardcode staleness of 0 for the HEAD
    return 0 unless $trans_id;

    $self->storage->print_at(
        $self->trans_loc + $self->txn_bitfield_len + $STALE_SIZE * ($trans_id - 1),
        pack( $StP{$STALE_SIZE}, $self->get_txn_staleness_counter( $trans_id ) + 1 ),
    );
}

=head2 get_entries()

This takes no arguments.

This returns a list of all the sectors that have been modified by this transaction.

=cut

sub get_entries {
    my $self = shift;
    return [ keys %{ $self->{entries}{$self->trans_id} ||= {} } ];
}

=head2 add_entry( $trans_id, $location )

This takes a transaction ID and a file location and marks the sector at that
location as having been modified by the transaction identified by $trans_id.

This returns nothing.

B<NOTE>: Unlike all the other _entries() methods, there are several cases where
C<< $trans_id != $self->trans_id >> for this method.

=cut

sub add_entry {
    my $self = shift;
    my ($trans_id, $loc) = @_;

    $self->{entries}{$trans_id} ||= {};
    $self->{entries}{$trans_id}{$loc} = undef;
}