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), ();
523 %opts = %{ merge(\%opts, $self->_master_connect_info_opts) };
529 $self->$next($self->schema, @args);
534 Returns an array of of all the connected storage backends. The first element
535 in the returned array is the master, and the remainings are each of the
542 return grep {defined $_ && blessed $_} (
544 values %{ $self->replicants },
548 =head2 execute_reliably ($coderef, ?@args)
550 Given a coderef, saves the current state of the L</read_handler>, forces it to
551 use reliable storage (ie sets it to the master), executes a coderef and then
552 restores the original state.
558 $schema->resultset('User')->create({name=>$name});
559 my $user_rs = $schema->resultset('User')->find({name=>$name});
563 my $user_rs = $schema->storage->execute_reliably($reliably, 'John');
565 Use this when you must be certain of your database state, such as when you just
566 inserted something and need to get a resultset including it, etc.
570 sub execute_reliably {
571 my ($self, $coderef, @args) = @_;
573 unless( ref $coderef eq 'CODE') {
574 $self->throw_exception('Second argument must be a coderef');
577 ##Get copy of master storage
578 my $master = $self->master;
580 ##Get whatever the current read hander is
581 my $current = $self->read_handler;
583 ##Set the read handler to master
584 $self->read_handler($master);
586 ## do whatever the caller needs
588 my $want_array = wantarray;
592 @result = $coderef->(@args);
593 } elsif(defined $want_array) {
594 ($result[0]) = ($coderef->(@args));
600 ##Reset to the original state
601 $self->read_handler($current);
603 ##Exception testing has to come last, otherwise you might leave the
604 ##read_handler set to master.
607 $self->throw_exception("coderef returned an error: $@");
609 return $want_array ? @result : $result[0];
613 =head2 set_reliable_storage
615 Sets the current $schema to be 'reliable', that is all queries, both read and
616 write are sent to the master
620 sub set_reliable_storage {
622 my $schema = $self->schema;
623 my $write_handler = $self->schema->storage->write_handler;
625 $schema->storage->read_handler($write_handler);
628 =head2 set_balanced_storage
630 Sets the current $schema to be use the </balancer> for all reads, while all
631 writea are sent to the master only
635 sub set_balanced_storage {
637 my $schema = $self->schema;
638 my $balanced_handler = $self->schema->storage->balancer;
640 $schema->storage->read_handler($balanced_handler);
645 Check that the master and at least one of the replicants is connected.
652 $self->master->connected &&
653 $self->pool->connected_replicants;
656 =head2 ensure_connected
658 Make sure all the storages are connected.
662 sub ensure_connected {
664 foreach my $source ($self->all_storages) {
665 $source->ensure_connected(@_);
671 Set the limit_dialect for all existing storages
677 foreach my $source ($self->all_storages) {
678 $source->limit_dialect(@_);
680 return $self->master->quote_char;
685 Set the quote_char for all existing storages
691 foreach my $source ($self->all_storages) {
692 $source->quote_char(@_);
694 return $self->master->quote_char;
699 Set the name_sep for all existing storages
705 foreach my $source ($self->all_storages) {
706 $source->name_sep(@_);
708 return $self->master->name_sep;
713 Set the schema object for all existing storages
719 foreach my $source ($self->all_storages) {
720 $source->set_schema(@_);
726 set a debug flag across all storages
733 foreach my $source ($self->all_storages) {
737 return $self->master->debug;
742 set a debug object across all storages
749 foreach my $source ($self->all_storages) {
750 $source->debugobj(@_);
753 return $self->master->debugobj;
758 set a debugfh object across all storages
765 foreach my $source ($self->all_storages) {
766 $source->debugfh(@_);
769 return $self->master->debugfh;
774 set a debug callback across all storages
781 foreach my $source ($self->all_storages) {
782 $source->debugcb(@_);
785 return $self->master->debugcb;
790 disconnect everything
796 foreach my $source ($self->all_storages) {
797 $source->disconnect(@_);
803 set cursor class on all storages, or return master's
808 my ($self, $cursor_class) = @_;
811 $_->cursor_class($cursor_class) for $self->all_storages;
813 $self->master->cursor_class;
818 Due to the fact that replicants can lag behind a master, you must take care to
819 make sure you use one of the methods to force read queries to a master should
820 you need realtime data integrity. For example, if you insert a row, and then
821 immediately re-read it from the database (say, by doing $row->discard_changes)
822 or you insert a row and then immediately build a query that expects that row
823 to be an item, you should force the master to handle reads. Otherwise, due to
824 the lag, there is no certainty your data will be in the expected state.
826 For data integrity, all transactions automatically use the master storage for
827 all read and write queries. Using a transaction is the preferred and recommended
828 method to force the master to handle all read queries.
830 Otherwise, you can force a single query to use the master with the 'force_pool'
833 my $row = $resultset->search(undef, {force_pool=>'master'})->find($pk);
835 This attribute will safely be ignore by non replicated storages, so you can use
836 the same code for both types of systems.
838 Lastly, you can use the L</execute_reliably> method, which works very much like
841 For debugging, you can turn replication on/off with the methods L</set_reliable_storage>
842 and L</set_balanced_storage>, however this operates at a global level and is not
843 suitable if you have a shared Schema object being used by multiple processes,
844 such as on a web application server. You can get around this limitation by
845 using the Schema clone method.
847 my $new_schema = $schema->clone;
848 $new_schema->set_reliable_storage;
850 ## $new_schema will use only the Master storage for all reads/writes while
851 ## the $schema object will use replicated storage.
855 John Napiorkowski <john.napiorkowski@takkle.com>
857 Based on code originated by:
859 Norbert Csongrádi <bert@cpan.org>
860 Peter Siklósi <einon@einon.hu>
864 You may distribute this code under the same terms as Perl itself.
868 __PACKAGE__->meta->make_immutable;