MySQL transactional semantics test script

gistfile1.pl

use DBI;
use strict;
use warnings;

use Time::HiRes qw(time sleep);
use Test::More 0.88;
use Test::Exception;
use List::Util qw(shuffle);

my $start = time(); # like $^T but fractional

my $duration = 2;
my $accounts = 5; # how many rows in the accounts table


#our $isolation_level = "SERIALIZABLE";     # uncomment to make it pass

#our $isolation_level = "REPEATABLE READ";  # the default level, also fails
                                            # (though reads may be repeatable,
                                            # the updates still succeed even if
                                            # they overwrite data other threads
                                            # have read

#our $isolation_level = "READ COMMITTED";   # fails, but that's expected

#our $isolation_level = "READ UNCOMMITTED"; # yechh

{
    # setup the tables and test data
    my $dbh = dbh();

    $dbh->begin_work;

    $dbh->do("DROP TABLE IF EXISTS account");
    $dbh->do("CREATE TABLE account ( id INTEGER PRIMARY KEY NOT NULL, balance INTEGER NOT NULL ) TYPE=INNODB");

    my $sth = $dbh->prepare("INSERT INTO account VALUES ( ?, 0 )");
    $sth->execute($_) for 1 .. $accounts;

    $dbh->commit;

    $dbh->disconnect;
}

if ( fork ) {
    my $dbh = dbh();

    # technically this isn't necessary, any level except for READ UNCOMMITTED
    # is supposed to give consistent reads
    $dbh->do("SET SESSION TRANSACTION ISOLATION LEVEL SERIALIZABLE");

    # we only read so we can reenable auto commit
    $dbh->{AutoCommit} = 1;
    
    # checks 
    until ( time > $start + $duration + 0.5 ) { 
        is_balanced($dbh);
        sleep 0.25;
    }


    # wait for all worker processes
    while ( wait != -1 ) { }

    # check the final outcome
    is_balanced($dbh);

    $dbh->disconnect;
    done_testing();
} else {
    fork for 1 .. 3; # if commented there is only one worker, and no race conditions so test passes

    srand $$;

    until ( time > $start + $duration ) {
        my $dbh = dbh();

        eval {
            # this eval block simulates a single transaction.
            # under the SERIALIZABLE level it could fail, and RaiseError would
            # die, which is why it's an eval block.
            $dbh->begin_work;

            sleep(rand 0.2) if rand > 0.5;

            my $balance = get_balance($dbh);

            sleep(rand 0.2) if rand > 0.5;

            my ( $from, $to ) = shuffle(1 .. $accounts);
            my $amount = int rand 100;
            transfer($dbh, $balance, $from, $to, $amount);

            sleep(rand 0.2) if rand > 0.5;

            $dbh->commit;
        };

        $dbh->disconnect;
    }

    exit;
}


sub dbh {
    my $dbh = DBI->connect("dbi:mysql:test", undef, undef, { RaiseError => 1, PrintError => 0, });

    if ( defined(our $isolation_level) ) {
        $dbh->do("SET SESSION TRANSACTION ISOLATION LEVEL $isolation_level");
    }

    return $dbh;
}

sub is_balanced {
    my $dbh = shift;
    is_deeply( $dbh->selectall_arrayref("SELECT sum(balance) FROM account"), [ [ 0 ] ], "accounts balanced" );
}

sub get_balance {
    my $dbh = shift;
    $dbh->selectall_hashref("SELECT * FROM account", "id"); # adding FOR UPDATE makes this safe on any isolation level
}

sub transfer {
    my ($dbh, $balances, $from_id, $to_id, $amount) = @_;

    my ( $from, $to ) = @{ $balances }{$from_id, $to_id};

    $from->{balance} -= $amount;
    $to->{balance}   += $amount;

    my $sth = $dbh->prepare("UPDATE account SET balance = ? WHERE id = ?");

    $sth->execute(map { 0+$_ } @{ $_ }{qw(balance id)}) for $from, $to;
};