1 package DBIx::Class::Storage::DBI::Replicated;
4 use Carp::Clan qw/^DBIx::Class/;
6 ## Modules required for Replication support not required for general DBIC
7 ## use, so we explicitly test for these.
9 my %replication_required = (
11 'MooseX::AttributeHelpers' => '0.21',
12 'MooseX::Types' => '0.16',
13 'namespace::clean' => '0.11',
14 'Hash::Merge' => '0.11'
19 for my $module (keys %replication_required) {
20 eval "use $module $replication_required{$module}";
21 push @didnt_load, "$module $replication_required{$module}"
25 croak("@{[ join ', ', @didnt_load ]} are missing and are required for Replication")
30 use DBIx::Class::Storage::DBI;
31 use DBIx::Class::Storage::DBI::Replicated::Pool;
32 use DBIx::Class::Storage::DBI::Replicated::Balancer;
33 use DBIx::Class::Storage::DBI::Replicated::Types qw/BalancerClassNamePart DBICSchema DBICStorageDBI/;
34 use MooseX::Types::Moose qw/ClassName HashRef Object/;
35 use Scalar::Util 'reftype';
36 use Carp::Clan qw/^DBIx::Class/;
37 use Hash::Merge 'merge';
39 use namespace::clean -except => 'meta';
43 DBIx::Class::Storage::DBI::Replicated - BETA Replicated database support
47 The Following example shows how to change an existing $schema to a replicated
48 storage type, add some replicated (readonly) databases, and perform reporting
51 You should set the 'storage_type attribute to a replicated type. You should
52 also define your arguments, such as which balancer you want and any arguments
53 that the Pool object should get.
55 $schema->storage_type( ['::DBI::Replicated', {balancer=>'::Random'}] );
57 Next, you need to add in the Replicants. Basically this is an array of
58 arrayrefs, where each arrayref is database connect information. Think of these
59 arguments as what you'd pass to the 'normal' $schema->connect method.
61 $schema->storage->connect_replicants(
62 [$dsn1, $user, $pass, \%opts],
63 [$dsn2, $user, $pass, \%opts],
64 [$dsn3, $user, $pass, \%opts],
67 Now, just use the $schema as you normally would. Automatically all reads will
68 be delegated to the replicants, while writes to the master.
70 $schema->resultset('Source')->search({name=>'etc'});
72 You can force a given query to use a particular storage using the search
73 attribute 'force_pool'. For example:
75 my $RS = $schema->resultset('Source')->search(undef, {force_pool=>'master'});
77 Now $RS will force everything (both reads and writes) to use whatever was setup
78 as the master storage. 'master' is hardcoded to always point to the Master,
79 but you can also use any Replicant name. Please see:
80 L<DBIx::Class::Storage::DBI::Replicated::Pool> and the replicants attribute for more.
82 Also see transactions and L</execute_reliably> for alternative ways to
83 force read traffic to the master. In general, you should wrap your statements
84 in a transaction when you are reading and writing to the same tables at the
85 same time, since your replicants will often lag a bit behind the master.
87 See L<DBIx::Class::Storage::DBI::Replicated::Instructions> for more help and
92 Warning: This class is marked BETA. This has been running a production
93 website using MySQL native replication as its backend and we have some decent
94 test coverage but the code hasn't yet been stressed by a variety of databases.
95 Individual DB's may have quirks we are not aware of. Please use this in first
96 development and pass along your experiences/bug fixes.
98 This class implements replicated data store for DBI. Currently you can define
99 one master and numerous slave database connections. All write-type queries
100 (INSERT, UPDATE, DELETE and even LAST_INSERT_ID) are routed to master
101 database, all read-type queries (SELECTs) go to the slave database.
103 Basically, any method request that L<DBIx::Class::Storage::DBI> would normally
104 handle gets delegated to one of the two attributes: L</read_handler> or to
105 L</write_handler>. Additionally, some methods need to be distributed
106 to all existing storages. This way our storage class is a drop in replacement
107 for L<DBIx::Class::Storage::DBI>.
109 Read traffic is spread across the replicants (slaves) occuring to a user
110 selected algorithm. The default algorithm is random weighted.
114 The consistancy betweeen master and replicants is database specific. The Pool
115 gives you a method to validate its replicants, removing and replacing them
116 when they fail/pass predefined criteria. Please make careful use of the ways
117 to force a query to run against Master when needed.
121 Replicated Storage has additional requirements not currently part of L<DBIx::Class>
124 MooseX::AttributeHelpers => '0.20',
125 MooseX::Types => '0.16',
126 namespace::clean => '0.11',
127 Hash::Merge => '0.11'
129 You will need to install these modules manually via CPAN or make them part of the
130 Makefile for your distribution.
134 This class defines the following attributes.
138 The underlying L<DBIx::Class::Schema> object this storage is attaching
151 Contains the classname which will instantiate the L</pool> object. Defaults
152 to: L<DBIx::Class::Storage::DBI::Replicated::Pool>.
159 default=>'DBIx::Class::Storage::DBI::Replicated::Pool',
161 'create_pool' => 'new',
167 Contains a hashref of initialized information to pass to the Balancer object.
168 See L<DBIx::Class::Storage::DBI::Replicated::Pool> for available arguments.
182 The replication pool requires a balance class to provider the methods for
183 choose how to spread the query load across each replicant in the pool.
187 has 'balancer_type' => (
189 isa=>BalancerClassNamePart,
192 default=> 'DBIx::Class::Storage::DBI::Replicated::Balancer::First',
194 'create_balancer' => 'new',
200 Contains a hashref of initialized information to pass to the Balancer object.
201 See L<DBIx::Class::Storage::DBI::Replicated::Balancer> for available arguments.
205 has 'balancer_args' => (
215 Is a <DBIx::Class::Storage::DBI::Replicated::Pool> or derived class. This is a
216 container class for one or more replicated databases.
222 isa=>'DBIx::Class::Storage::DBI::Replicated::Pool',
233 Is a <DBIx::Class::Storage::DBI::Replicated::Balancer> or derived class. This
234 is a class that takes a pool (<DBIx::Class::Storage::DBI::Replicated::Pool>)
240 isa=>'DBIx::Class::Storage::DBI::Replicated::Balancer',
242 handles=>[qw/auto_validate_every/],
247 The master defines the canonical state for a pool of connected databases. All
248 the replicants are expected to match this databases state. Thus, in a classic
249 Master / Slaves distributed system, all the slaves are expected to replicate
250 the Master's state as quick as possible. This is the only database in the
251 pool of databases that is allowed to handle write traffic.
261 =head1 ATTRIBUTES IMPLEMENTING THE DBIx::Storage::DBI INTERFACE
263 The following methods are delegated all the methods required for the
264 L<DBIx::Class::Storage::DBI> interface.
268 Defines an object that implements the read side of L<BIx::Class::Storage::DBI>.
272 has 'read_handler' => (
285 Defines an object that implements the write side of L<BIx::Class::Storage::DBI>.
289 has 'write_handler' => (
301 deployment_statements
304 build_datetime_parser
318 with_deferred_fk_checks
321 with_deferred_fk_checks
329 _order_select_columns
336 has _master_connect_info_opts =>
337 (is => 'rw', isa => HashRef, default => sub { {} });
339 =head2 around: connect_info
341 Preserve master's C<connect_info> options (for merging with replicants.)
342 Also set any Replicated related options from connect_info, such as
343 C<pool_type>, C<pool_args>, C<balancer_type> and C<balancer_args>.
347 around connect_info => sub {
348 my ($next, $self, $info, @extra) = @_;
350 my $wantarray = wantarray;
353 for my $arg (@$info) {
354 next unless (reftype($arg)||'') eq 'HASH';
355 %opts = %{ merge($arg, \%opts) };
359 if (@opts{qw/pool_type pool_args/}) {
360 $self->pool_type(delete $opts{pool_type})
364 merge((delete $opts{pool_args} || {}), $self->pool_args)
367 $self->pool($self->_build_pool)
371 if (@opts{qw/balancer_type balancer_args/}) {
372 $self->balancer_type(delete $opts{balancer_type})
373 if $opts{balancer_type};
375 $self->balancer_args(
376 merge((delete $opts{balancer_args} || {}), $self->balancer_args)
379 $self->balancer($self->_build_balancer)
383 $self->_master_connect_info_opts(\%opts);
387 @res = $self->$next($info, @extra);
389 $res = $self->$next($info, @extra);
392 # Make sure master is blessed into the correct class and apply role to it.
393 my $master = $self->master;
394 $master->_determine_driver;
395 Moose::Meta::Class->initialize(ref $master);
396 DBIx::Class::Storage::DBI::Replicated::WithDSN->meta->apply($master);
398 $wantarray ? @res : $res;
403 This class defines the following methods.
407 L<DBIx::Class::Schema> when instantiating its storage passed itself as the
408 first argument. So we need to massage the arguments a bit so that all the
409 bits get put into the correct places.
414 my ($class, $schema, $storage_type_args, @args) = @_;
425 Lazy builder for the L</master> attribute.
431 my $master = DBIx::Class::Storage::DBI->new($self->schema);
437 Lazy builder for the L</pool> attribute.
443 $self->create_pool(%{$self->pool_args});
446 =head2 _build_balancer
448 Lazy builder for the L</balancer> attribute. This takes a Pool object so that
449 the balancer knows which pool it's balancing.
453 sub _build_balancer {
455 $self->create_balancer(
457 master=>$self->master,
458 %{$self->balancer_args},
462 =head2 _build_write_handler
464 Lazy builder for the L</write_handler> attribute. The default is to set this to
469 sub _build_write_handler {
470 return shift->master;
473 =head2 _build_read_handler
475 Lazy builder for the L</read_handler> attribute. The default is to set this to
480 sub _build_read_handler {
481 return shift->balancer;
484 =head2 around: connect_replicants
486 All calls to connect_replicants needs to have an existing $schema tacked onto
487 top of the args, since L<DBIx::Storage::DBI> needs it, and any C<connect_info>
488 options merged with the master, with replicant opts having higher priority.
492 around connect_replicants => sub {
493 my ($next, $self, @args) = @_;
496 $r = [ $r ] unless reftype $r eq 'ARRAY';
498 croak "coderef replicant connect_info not supported"
499 if ref $r->[0] && reftype $r->[0] eq 'CODE';
501 # any connect_info options?
503 $i++ while $i < @$r && (reftype($r->[$i])||'') ne 'HASH';
506 $r->[$i] = {} unless $r->[$i];
508 # merge if two hashes
509 my @hashes = @$r[$i .. $#{$r}];
511 croak "invalid connect_info options"
512 if (grep { reftype($_) eq 'HASH' } @hashes) != @hashes;
514 croak "too many hashrefs in connect_info"
517 my %opts = %{ merge(reverse @hashes) };
520 splice @$r, $i+1, ($#{$r} - $i), ();
522 # make sure master/replicants opts don't clash
523 my %master_opts = %{ $self->_master_connect_info_opts };
524 if (exists $opts{dbh_maker}) {
525 delete @master_opts{qw/dsn user password/};
527 delete $master_opts{dbh_maker};
530 %opts = %{ merge(\%opts, \%master_opts) };
536 $self->$next($self->schema, @args);
541 Returns an array of of all the connected storage backends. The first element
542 in the returned array is the master, and the remainings are each of the
549 return grep {defined $_ && blessed $_} (
551 values %{ $self->replicants },
555 =head2 execute_reliably ($coderef, ?@args)
557 Given a coderef, saves the current state of the L</read_handler>, forces it to
558 use reliable storage (ie sets it to the master), executes a coderef and then
559 restores the original state.
565 $schema->resultset('User')->create({name=>$name});
566 my $user_rs = $schema->resultset('User')->find({name=>$name});
570 my $user_rs = $schema->storage->execute_reliably($reliably, 'John');
572 Use this when you must be certain of your database state, such as when you just
573 inserted something and need to get a resultset including it, etc.
577 sub execute_reliably {
578 my ($self, $coderef, @args) = @_;
580 unless( ref $coderef eq 'CODE') {
581 $self->throw_exception('Second argument must be a coderef');
584 ##Get copy of master storage
585 my $master = $self->master;
587 ##Get whatever the current read hander is
588 my $current = $self->read_handler;
590 ##Set the read handler to master
591 $self->read_handler($master);
593 ## do whatever the caller needs
595 my $want_array = wantarray;
599 @result = $coderef->(@args);
600 } elsif(defined $want_array) {
601 ($result[0]) = ($coderef->(@args));
607 ##Reset to the original state
608 $self->read_handler($current);
610 ##Exception testing has to come last, otherwise you might leave the
611 ##read_handler set to master.
614 $self->throw_exception("coderef returned an error: $@");
616 return $want_array ? @result : $result[0];
620 =head2 set_reliable_storage
622 Sets the current $schema to be 'reliable', that is all queries, both read and
623 write are sent to the master
627 sub set_reliable_storage {
629 my $schema = $self->schema;
630 my $write_handler = $self->schema->storage->write_handler;
632 $schema->storage->read_handler($write_handler);
635 =head2 set_balanced_storage
637 Sets the current $schema to be use the </balancer> for all reads, while all
638 writea are sent to the master only
642 sub set_balanced_storage {
644 my $schema = $self->schema;
645 my $balanced_handler = $self->schema->storage->balancer;
647 $schema->storage->read_handler($balanced_handler);
652 Check that the master and at least one of the replicants is connected.
659 $self->master->connected &&
660 $self->pool->connected_replicants;
663 =head2 ensure_connected
665 Make sure all the storages are connected.
669 sub ensure_connected {
671 foreach my $source ($self->all_storages) {
672 $source->ensure_connected(@_);
678 Set the limit_dialect for all existing storages
684 foreach my $source ($self->all_storages) {
685 $source->limit_dialect(@_);
687 return $self->master->quote_char;
692 Set the quote_char for all existing storages
698 foreach my $source ($self->all_storages) {
699 $source->quote_char(@_);
701 return $self->master->quote_char;
706 Set the name_sep for all existing storages
712 foreach my $source ($self->all_storages) {
713 $source->name_sep(@_);
715 return $self->master->name_sep;
720 Set the schema object for all existing storages
726 foreach my $source ($self->all_storages) {
727 $source->set_schema(@_);
733 set a debug flag across all storages
740 foreach my $source ($self->all_storages) {
744 return $self->master->debug;
749 set a debug object across all storages
756 foreach my $source ($self->all_storages) {
757 $source->debugobj(@_);
760 return $self->master->debugobj;
765 set a debugfh object across all storages
772 foreach my $source ($self->all_storages) {
773 $source->debugfh(@_);
776 return $self->master->debugfh;
781 set a debug callback across all storages
788 foreach my $source ($self->all_storages) {
789 $source->debugcb(@_);
792 return $self->master->debugcb;
797 disconnect everything
803 foreach my $source ($self->all_storages) {
804 $source->disconnect(@_);
810 set cursor class on all storages, or return master's
815 my ($self, $cursor_class) = @_;
818 $_->cursor_class($cursor_class) for $self->all_storages;
820 $self->master->cursor_class;
825 Due to the fact that replicants can lag behind a master, you must take care to
826 make sure you use one of the methods to force read queries to a master should
827 you need realtime data integrity. For example, if you insert a row, and then
828 immediately re-read it from the database (say, by doing $row->discard_changes)
829 or you insert a row and then immediately build a query that expects that row
830 to be an item, you should force the master to handle reads. Otherwise, due to
831 the lag, there is no certainty your data will be in the expected state.
833 For data integrity, all transactions automatically use the master storage for
834 all read and write queries. Using a transaction is the preferred and recommended
835 method to force the master to handle all read queries.
837 Otherwise, you can force a single query to use the master with the 'force_pool'
840 my $row = $resultset->search(undef, {force_pool=>'master'})->find($pk);
842 This attribute will safely be ignore by non replicated storages, so you can use
843 the same code for both types of systems.
845 Lastly, you can use the L</execute_reliably> method, which works very much like
848 For debugging, you can turn replication on/off with the methods L</set_reliable_storage>
849 and L</set_balanced_storage>, however this operates at a global level and is not
850 suitable if you have a shared Schema object being used by multiple processes,
851 such as on a web application server. You can get around this limitation by
852 using the Schema clone method.
854 my $new_schema = $schema->clone;
855 $new_schema->set_reliable_storage;
857 ## $new_schema will use only the Master storage for all reads/writes while
858 ## the $schema object will use replicated storage.
862 John Napiorkowski <john.napiorkowski@takkle.com>
864 Based on code originated by:
866 Norbert Csongrádi <bert@cpan.org>
867 Peter Siklósi <einon@einon.hu>
871 You may distribute this code under the same terms as Perl itself.
875 __PACKAGE__->meta->make_immutable;