File Coverage

blib/lib/DBIx/TxnPool.pm

Criterion	Covered	Total	%
statement	24	136	17.6
branch	0	42	0.0
condition	0	13	0.0
subroutine	8	38	21.0
pod	8	13	61.5
total	40	242	16.5

package DBIx::TxnPool;

use strict;
use warnings;
use Exporter 5.57 qw( import );

use Try::Tiny;
use Signal::Mask;
use Carp qw( confess croak );

our $VERSION = 0.10;
our $BlockSignals = [ qw( TERM INT ) ];
our @EXPORT = qw( txn_item txn_post_item txn_commit txn_sort );

# It's better to look for the "try restarting transaction" string
# because sometime may be happens other error: Lock wait timeout exceeded
use constant DEADLOCK_REGEXP    => qr/try restarting transaction/o;

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

    croak( __PACKAGE__ . ": the dbh should be defined" )
      unless $args{dbh};

    $args{size}                   ||= 100;
    $args{block_signals}          ||= $BlockSignals;
    $args{max_repeated_deadlocks} ||= 5;
    $args{_amnt_nested_signals}   = 0;
    $args{_saved_signal_masks}    = {};
    $args{pool}                   = [];
    $args{amount_deadlocks}       = 0;

    bless \%args, ref $class || $class;
}

sub DESTROY {
    $_[0]->finish;
}

sub txn_item (&@) {
    __PACKAGE__->new( %{ __make_chain( 'item_callback', @_ ) } );
}

sub txn_post_item (&@) {
    __make_chain( 'post_item_callback', @_ );
}

sub txn_commit (&@) {
    __make_chain( 'commit_callback', @_ );
}

sub txn_sort (&@) {
    my $ret = __make_chain( 'sort_callback', @_ );
    $ret->{sort_callback_package} = caller;
    $ret;
}

sub __make_chain {
    my $cb_name = shift;
    my $cb_func = shift;
    my $ret;

    ( $ret = ref $_[0] eq 'HASH' ? $_[0] : { @_ } )->{ $cb_name } = $cb_func;
    $ret;
}

sub dbh { $_[0]->{dbh} }

sub add {
    my ( $self, $data ) = @_;

    $self->{repeated_deadlocks} = 0;

    die "assert: _amnt_nested_signals is not zero!"
      if $self->{_amnt_nested_signals};

    try {
        push @{ $self->{pool} }, $data;

        if ( ! $self->{sort_callback} ) {
            $self->start_txn;
            $self->_safe_signals( sub {
                local $_ = $data;
                $self->{item_callback}->( $self, $data );
            } );
        }
    }
    catch {
        $self->_check_deadlock( $_ );
    };

    $self->finish
      if ( @{ $self->{pool} } >= $self->{size} );

    die "assert: _amnt_nested_signals is not zero!"
      if $self->{_amnt_nested_signals};
}

sub _check_deadlock {
    my ( $self, $error ) = @_;

    $self->rollback_txn;

    if ( $error =~ DEADLOCK_REGEXP ) {
        $self->{amount_deadlocks}++;
        if ( $self->{repeated_deadlocks} >= $self->{max_repeated_deadlocks} ) {
            confess( "limit ($self->{repeated_deadlocks}) of deadlock resolvings" )
        }
        else {
            $self->play_pool
        }
    } else {
        # Fatal error - may be bad SQL statement - finish
        $self->{pool} = []; # If DESTROY calls finish() there will not problems
        confess( "error in item callback ($error)" );
    }
}

sub play_pool {
    my $self = shift;

    $self->start_txn;

    $self->_safe_signals( sub {
        select( undef, undef, undef, 0.5 * ++$self->{repeated_deadlocks} );
    } );

    try {
        foreach my $data ( @{ $self->{pool} } ) {
            $self->_safe_signals( sub {
                local $_ = $data;
                $self->{item_callback}->( $self, $data );
            } );
        }
    }
    catch {
        $self->_check_deadlock( $_ );
    };
}

sub finish {
    my $self = shift;

    die "assert: _amnt_nested_signals is not zero!"
      if $self->{_amnt_nested_signals};

    if ( $self->{sort_callback} && @{ $self->{pool} } ) {
        no strict 'refs';
        local *a = *{"$self->{sort_callback_package}\::a"};
        local *b = *{"$self->{sort_callback_package}\::b"};

        $self->{pool} = [ sort { $self->{sort_callback}->() } ( @{ $self->{pool} } ) ];

        $self->play_pool;
    }

    try {
        $self->{repeated_deadlocks} = 0;
        $self->commit_txn;
    }
    catch {
        $self->_check_deadlock( $_ );
    };

    if ( exists $self->{post_item_callback} ) {
        foreach my $data ( @{ $self->{pool} } ) {
            local $_ = $data;
            $self->{post_item_callback}->( $self, $data );
        }
    }

    $self->{pool} = [];

    die "assert: _amnt_nested_signals is not zero!"
      if $self->{_amnt_nested_signals};
}

sub start_txn {
    my $self = shift;

    if ( ! $self->{in_txn} ) {
        $self->_safe_signals( sub {
            $self->{dbh}->begin_work or die $self->{dbh}->errstr;
            $self->{in_txn} = 1;
        } );
    }
}

sub rollback_txn {
    my $self = shift;

    if ( $self->{in_txn} ) {
        $self->_safe_signals( sub {
            $self->{dbh}->rollback or die $self->{dbh}->errstr;
            $self->{in_txn} = undef;
        } );
    }
}

sub commit_txn {
    my $self = shift;

    if ( $self->{in_txn} ) {
        $self->_safe_signals( sub {
            $self->{dbh}->commit or die $self->{dbh}->errstr;
            $self->{in_txn} = undef;
        } );
        $self->{commit_callback}->( $self )
          if exists $self->{commit_callback};
    }
}

sub amount_deadlocks { $_[0]->{amount_deadlocks} }

sub _safe_signals {
    my ( $self, $code ) = @_;

    if ( ! $self->{_amnt_nested_signals}++ ) {
        for ( @{ $self->{block_signals} } ) {
            $self->{_saved_signal_masks}{ $_ } = $Signal::Mask{ $_ };
            $Signal::Mask{ $_ } = 1;
        }
    }
    try {
        $code->();
    }
    catch {
        die $_;
    }
    finally {
        if ( ! --$self->{_amnt_nested_signals} ) {
            for ( @{ $self->{block_signals} } ) {
                $Signal::Mask{ $_ } = delete $self->{_saved_signal_masks}{ $_ };
            }
        }
    };
}

1;

__END__