package DBD::Multi;
our $VERSION = '1.02';
package DBD::Multi::dr;
our $VERSION = '1.02';
package DBD::Multi::db;
our $VERSION = '1.02';
package DBD::Multi::st;
our $VERSION = '1.02';
package DBD::Multi::Handler;
our $VERSION = '1.02';
package DBD::Multi;
use strict;
use DBI;
DBI->setup_driver('DBD::Multi');
use vars qw[$err $errstr $sqlstate $drh];
$err = 0; # DBI::err
$errstr = ""; # DBI::errstr
$sqlstate = ""; # DBI::state
$drh = undef;
sub driver {
return $drh if $drh;
my($class, $attr) = @_;
$class .= '::dr';
$drh = DBI::_new_drh($class, {
Name => 'Multi',
Version => $VERSION,
Err => \$DBD::Multi::err,
Errstr => \$DBD::Multi::errstr,
State => \$DBD::Multi::sqlstate,
Attribution => 'DBD::Multi, pair Networks Inc.',
});
DBD::Multi::db->install_method('multi_do_all');
return $drh;
}
#######################################################################
package DBD::Multi::dr;
use strict;
$DBD::Multi::dr::imp_data_size = 0;
use DBD::File;
sub DESTROY { shift->STORE(Active => 0) }
sub connect {
my($drh, $dbname, $user, $auth, $attr) = @_;
my $dbh = DBI::_new_dbh(
$drh => {
Name => $dbname,
USER => $user,
CURRENT_USER => $user,
},
);
my @dsns = $attr->{dsns} && ref($attr->{dsns}) eq 'ARRAY'
? @{$attr->{dsns}}
: ();
if ( $dbname =~ /dsn=(.*)/ ) {
push @dsns, ( -1, [$1, $user, $auth] );
}
my $handler = DBD::Multi::Handler->new({
dsources => [ @dsns ],
});
$handler->failed_max($attr->{failed_max})
if exists $attr->{failed_max};
$handler->failed_expire($attr->{failed_expire})
if exists $attr->{failed_expire};
$dbh->STORE(_handler => $handler);
$dbh->STORE(handler => $handler); # temporary
$drh->{_handler} = $handler;
$dbh->STORE(Active => 1);
return $dbh;
}
sub data_sources { shift->FETCH('_handler')->all_sources }
#######################################################################
package DBD::Multi::db;
use strict;
$DBD::Multi::db::imp_data_size = 0;
sub prepare {
my ($dbh, $statement, @attribs) = @_;
# create a 'blank' sth
my ($outer, $sth) = DBI::_new_sth($dbh, { Statement => $statement });
my $handler = $dbh->FETCH('_handler');
$sth->STORE(_handler => $handler);
my $_dbh = $handler->dbh;
my $_sth;
until ( $_sth ) {
$_sth = $_dbh->prepare($statement, @attribs);
unless ( $_sth ) {
$handler->dbh_failed;
$_dbh = $handler->dbh;
}
}
$sth->STORE(NUM_OF_PARAMS => $_sth->FETCH('NUM_OF_PARAMS'));
$sth->STORE(_dbh => $_dbh);
$sth->STORE(_sth => $_sth);
return $outer;
}
sub multi_do_all {
my $dbh = shift;
my $handler = $dbh->FETCH('_handler');
return $handler->multi_do_all( @_ );
}
sub disconnect {
my ($dbh) = @_;
$dbh->STORE(Active => 0);
1;
}
sub commit {
my ($dbh) = @_;
if ( $dbh->FETCH('Active') ) {
return $dbh->FETCH('_dbh')->commit if $dbh->FETCH('_dbh');
}
return;
}
sub rollback {
my ($dbh) = @_;
if ( $dbh->FETCH('Active') ) {
return $dbh->FETCH('_dbh')->rollback if $dbh->FETCH('_dbh');
}
return;
}
sub get_info {
my($dbh, $info_type) = @_;
# return info from current connection
my $handler = $dbh->FETCH('_handler');
my $_dbh = $handler->dbh;
return $_dbh->get_info($info_type);
}
sub STORE {
my ($self, $attr, $val) = @_;
$self->{$attr} = $val;
}
sub DESTROY { shift->disconnect }
#######################################################################
package DBD::Multi::st;
use strict;
$DBD::Multi::st::imp_data_size = 0;
use vars qw[@METHODS @FIELDS];
@METHODS = qw[
bind_param
bind_param_inout
bind_param_array
execute_array
execute_for_fetch
fetch
fetchrow_arrayref
fetchrow_array
fetchrow_hashref
fetchall_arrayref
fetchall_hashref
bind_col
bind_columns
dump_results
];
@FIELDS = qw[
NUM_OF_FIELDS
CursorName
ParamValues
RowsInCache
];
sub execute {
my $sth = shift;
my $_sth = $sth->FETCH('_sth');
my $params = @_
? $sth->{f_params} = [ @_ ]
: $sth->{f_params};
$sth->finish if $sth->FETCH('Active');
$sth->{Active} = 1;
my $rc = $_sth->execute(@{$params});
for my $field ( @FIELDS ) {
my $value = $_sth->FETCH($field);
$sth->STORE($field => $value)
unless ! defined $value
|| defined $sth->FETCH($field);
}
return $rc;
}
sub FETCH {
my ($sth, $attrib) = @_;
$sth->{'_sth'}->FETCH($attrib) || $sth->{$attrib};
}
sub STORE {
my ($self, $attr, $val) = @_;
$self->{$attr} = $val;
}
sub rows { shift->FETCH('_sth')->rows }
sub finish {
my ($sth) = @_;
$sth->STORE(Active => 0);
return $sth->FETCH('_sth')->finish;
}
foreach my $method ( @METHODS ) {
no strict;
*{$method} = sub { shift->FETCH('_sth')->$method(@_) };
}
#######################################################################
package DBD::Multi::Handler;
use strict;
use base qw[Class::Accessor::Fast];
use Sys::SigAction qw(timeout_call);
use List::Util qw(shuffle);
=begin ImplementationNotes
dsources - This thing changes from an arrayref to a hashref during construction. :(
Initially, when data is passed in during construction, it's an arrayref
containing the 'dsns' param from the user's connect() call.
Later, when _configure_dsources gets called, it turns into a multi-dimension
hashref:
$dsources->{$pri}->{$dsource_id} = 1;
The first key is the priority number, the second key is the data source index
number. The value is always just a true value.
nextid - A counter. Stores the index number of the next data source to be added.
all_dsources - A hashref. Maps index number to the connect data.
current_dsource - The most recent chosen datasource index number.
used - A hashref. Keys are index numbers. Values are true when the datasource
has been previously assigned and we want to prefer other datasources of the
same priority (for round-robin load distribution).
failed - A hashref. Keys are index numbers. Values are counters indicating
how many times the data source has failed.
failed_last - A hashref. Keys are index number. Values are unix timestamp
indicating the most recent time a data source failed.
failed_max - A scalar value. Number of times a datasource may fail before we
stop trying it.
failed_expire - A scalar value. Number of seconds since we stopped trying a
datasource before we'll try it again.
timeout - A scalar value. Number of seconds we try to connect to a datasource
before giving up.
=end ImplementationNotes
=cut
__PACKAGE__->mk_accessors(qw[
dsources
nextid
all_dsources
current_dsource
used
failed
failed_last
failed_max
failed_expire
timeout
]);
sub new {
my ($class, $args) = @_;
my $self = $class->SUPER::new($args);
$self->nextid(0) unless defined $self->nextid;
$self->all_dsources({});
$self->used({});
$self->failed({});
$self->failed_last({});
$self->failed_max(3) unless defined $self->failed_max;
$self->failed_expire(60*5) unless defined $self->failed_expire;
$self->timeout( 5 ) unless defined $self->timeout;
$self->_configure_dsources;
return $self;
}
sub all_sources {
my ($self) = @_;
return values %{$self->all_dsources};
}
sub add_to_pri {
my ($self, $pri, $dsource) = @_;
my $dsource_id = $self->nextid;
my $dsources = $self->dsources;
my $all = $self->all_dsources;
$all->{$dsource_id} = $dsource;
$dsources->{$pri}->{$dsource_id} = 1;
$self->nextid($dsource_id + 1);
}
sub dbh {
my $self = shift;
my $dbh = $self->_connect_dsource;
return $dbh if $dbh;
$self->dbh_failed;
$self->dbh;
}
sub dbh_failed {
my ($self) = @_;
my $current_dsource = $self->current_dsource;
$self->failed->{$current_dsource}++;
$self->failed_last->{$current_dsource} = time;
}
sub _purge_old_failures {
my ($self) = @_;
my $now = time;
my @all = keys %{$self->all_dsources};
foreach my $dsource ( @all ) {
next unless $self->failed->{$dsource};
if ( ($now - $self->failed_last->{$dsource}) > $self->failed_expire ) {
delete $self->failed->{$dsource};
delete $self->failed_last->{$dsource};
}
}
}
sub _pick_dsource {
my ($self) = @_;
$self->_purge_old_failures;
my $dsources = $self->dsources;
my @pri = sort { $a <=> $b } keys %{$dsources};
foreach my $pri ( @pri ) {
my $dsource = $self->_pick_pri_dsource($dsources->{$pri});
if ( defined $dsource ) {
$self->current_dsource($dsource);
return;
}
}
$self->used({});
return $self->_pick_dsource
if (grep {$self->failed->{$_} >= $self->failed_max} keys(%{$self->failed})) < keys(%{$self->all_dsources});
die("All data sources failed!");
}
### _pick_pri_dsource
# Given a list of data sources, all of the same priority, choose one to use.
# Passed in a structure like this:
# _pick_pri_dsource( $self, { 3 => 1, 4 => 1, 5 => 1 } )
# Where the keys 3,4,5 are dsource id's, and the values are always true.
sub _pick_pri_dsource {
my ($self, $dsources) = @_;
my @dsources = sort { $a <=> $b } keys %{$dsources}; # Indexes of data sources to try.
my @used = grep { exists $self->used->{$_} } @dsources; # List of data sources already used.
my @failed = grep { exists($self->failed->{$_}) && $self->failed->{$_} >= $self->failed_max } @dsources; # List of data sources that won't be tried.
# We've used them all and they all failed. Escallate.
return if @used == @dsources && @failed == @dsources;
# We've used them all but some are good. Purge and reuse.
delete @{$self->used}{@dsources} if @used == @dsources;
foreach my $dsource ( shuffle @dsources ) {
next if $self->failed->{$dsource}
&& $self->failed->{$dsource} >= $self->failed_max;
next if $self->used->{$dsource};
$self->used->{$dsource} = 1;
return $dsource;
}
return;
}
sub _configure_dsources {
my ($self) = @_;
my $dsources = $self->dsources;
$self->dsources({});
while ( my $pri = shift @{$dsources} ) {
my $dsource = shift @{$dsources} or last;
$self->add_to_pri($pri => $dsource);
}
}
sub _connect_dsource {
my ($self, $dsource) = @_;
unless ( $dsource ) {
$self->_pick_dsource;
$dsource = $self->all_dsources->{$self->current_dsource};
}
# Support ready-made handles
return $dsource if UNIVERSAL::isa($dsource, 'DBI::db');
# Support code-refs which return handles
if (ref $dsource eq 'CODE') {
my $handle = $dsource->();
return $handle if UNIVERSAL::isa($handle, 'DBI::db');
return undef; # Connect by coderef failed.
}
my $dbh;
local $ENV{DBI_AUTOPROXY};
if (timeout_call( $self->timeout, sub { $dbh = DBI->connect_cached(@{$dsource}) } )) {
#warn "Timeout[", $self->current_dsource, "] at ", time, "\n";
}
return $dbh;
}
sub connect_dsource {
my ($self, $dsource) = @_;
$self->_connect_dsource($dsource);
}
sub multi_do_all {
my ($self, $code) = @_;
my @all = values %{$self->all_dsources};
foreach my $source ( @all ) {
my $dbh = $self->connect_dsource($source);
next unless $dbh;
if ( $dbh->{handler} ) {
$dbh->{handler}->multi_do_all($code, $source);
next;
}
$code->($dbh);
}
}
1;
__END__
=head1 NAME
DBD::Multi - Manage Multiple Data Sources with Failover and Load Balancing
=head1 SYNOPSIS
use DBI;
my $other_dbh = DBI->connect(...);
my $dbh = DBI->connect( 'dbi:Multi:', undef, undef, {
dsns => [ # in priority order
10 => [ 'dbi:SQLite:read_one.db', '', '' ],
10 => [ 'dbi:SQLite:read_two.db', '', '' ],
20 => [ 'dbi:SQLite:master.db', '', '' ],
30 => $other_dbh,
40 => sub { DBI->connect },
],
# optional
failed_max => 1, # short credibility
failed_expire => 60*60, # long memory
timeout => 10, # time out connection attempts after 10 seconds.
});
$dbh->prepare(...); # Works like any other DBI handle.
$dbh->multi_do_all( # Loops through every single DB handle.
sub {
my $dbh = shift;
...
} );
=head1 DESCRIPTION
This software manages multiple database connections for failovers and also
simple load balancing. It acts as a proxy between your code and your database
connections, transparently choosing a connection for each query, based on your
preferences and present availability of the DB server.
This module is intended for read-only operations (where some other application
is being used to handle replication).
This software does not prevent write operations from being executed. This is
left up to the user. See L<SUGGESTED USES> below for ideas.
The interface is nearly the same as other DBI drivers except that it allows you
to specify multiple connections for a single handle.
=head1 CONFIGURING CONNECTIONS
=head2 Configuring DSNs
Specify an attribute to the C<connect()> constructor, C<dsns>. This is a list
of DSNs to configure. The configuration is given in pairs. First comes the
priority of the DSN. Second is the DSN.
The priorities specify which connections should be used first (lowest to
highest). As long as the lowest priority connection is responding, the higher
priority connections will never be used. If multiple connections have the same
priority, then one connection will be chosen randomly for each operation. Note
that the random DB is chosen when the statement is prepared. Therefore
executing multiple queries on the same prepared statement handle will always
run on the same connection.
The second parameter can a DBI object, a code ref which returns a DBI object,
or a list of parameters to pass to the DBI C<connect()> instructor. If a set
of parameters or a code ref is given, then DBD::Multi will be able to attempt
re-connect in the event that the connection is lost. If a DBI object is used,
the DBD::Multi will give up permanently once that connection is lost.
These connections are lazy loaded, meaning they aren't made until they are
actually used.
=head2 Configuring Failures
By default, after a data source fails three times, it will not be tried again
for 5 minutes. After that period, the data source will be tried again for
future requests until it reaches its three failure limit (the cycle repeats
forever).
To change the maximum number of failures allowed before a data source is
deemed failed, set the C<failed_max> parameter. To change the amount of
time we remember a data source as being failed, set the C<failed_expire>
parameter in seconds.
=head2 Timing out connections.
By default, if you attempt to connect to an IP that isn't answering, DBI will
hang for a very long period of time. This behavior is not desirable in a
multi database setup. Instead, it is better to give up on slow connections
and move on to other databases quickly.
DBD::Multi will give up on connection attempts after 5 seconds and then try
another connection. You may set the C<timeout> parameter to change the
timeout time, or set it to 0 to disable the timeout feature completely.
=head1 EXTRA METHODS
=head2 multi_do_all
Loops through every database handle, executing an arbitrary coderef passing the
current database handle as the first parameter and the original connection
parameters as the second parameter.
If a database is unreachable, multi_do_all will skip over it.
use Data::Dumper;
my $expected_value = ...;
$dbh->multi_do_all(
sub {
my $dbh = shift;
my $source = shift;
my($value) = $dbh->selectrow_array("SELECT ...");
unless ( $value eq $expected_value ) {
die "Unexpected value, $value found. (Expected $expected_value). Data Source:\n", Dumper( $source );
}
} );
=head1 SUGGESTED USES
Here are some ideas on how to use this module effectively and safely.
It is important to remember that C<DBD::Multi> is not intended for read-write
operations. One suggestion to prevent accidental write operations is to make
sure that the user you are connecting to the databases with has privileges
sufficiently restricted to prevent updates.
Read-write operations should happen through a separate database handle that
will somehow trigger replication to all of your databases. For example, your
read-write handle might be connected to the master server that replicates
itself to all of the subordinate servers.
Read-only database calls within your application would be updated to explicitly
use the read-only (DBD::Multi) handle. It is not necessary to find every single
call that can be load balanced, since they can safely be sent through the
read/write handle as well.
=head1 TODO
There really isn't much of a TODO list for this module at this time. Feel free
to submit a bug report to L<github|https://github.com/dwright/DBD-Multi/issues>
if you think there is a feature missing.
Although there is some code intended for read/write operations, this should be
considered not supported and not actively developed at this time. The actual
read/write code remains un-documented because in the event that I ever do
decide to work on supporting read/write operations, the API is not guaranteed
to stay the same. The focus of this module is presently limited to read-only
operations.
=head1 TESTING
DBD::Multi has it's own suite of regression tests. But, suppose you want to
verify that you can slip DBD::Multi into whatever application you already have
written without breaking anything.
Thanks to a feature of DBI, you can regression test DBD::Multi using any
existing tests that already use DBI without having to update any of your code.
Simply set the environment variable DBI_AUTOPROXY to 'dbi:Multi:' and then run
your tests. DBD::Multi should act as a silent pipe between your application
and whatever database driver you were previously using. This will help you
verify that you aren't currently using some feature of the DBI that breaks
DBD::Multi (If you are, please do me a favor and submit a bug report so I can
fix it).
=head1 SEE ALSO
There are other modules that have similar, but different objectives. Depending
on your specific needs these may be more or less suitable for your task:
=over
=item L<CGI::Application::Plugin::DBH>
A plugin for the L<CGI::Application> framework which makes it easy to support
two database handles, and also supports lazy-loading.
=item L<DBD::Multiplex>
The original inspiration for this module. It doesn't support as many
connection configurations options at this module. It does try to support
write options in a single master, mutliple slave configuration. It does this
by parsing your SQL and trying to decide if you were doing a read or write
operation.
=item L<DBIx::HA>
Written after this module. Built for high availability rather than load
balancing. It purposely ignores some DBI features in favor of producing the
fastest results for the most common operations. It doesn't utilize the
standard DBI->connect() API, which means it will not work as a drop-in auto
proxy.
=back
L<DBI>, L<perl> - You should probably already know about these before using
this module.
=head1 AUTHOR
Initially written by Casey West and Dan Wright for pair Networks, Inc.
(www.pair.com)
Maintained by Dan Wright. <F<DWRIGHT@CPAN.ORG>>.
=cut