1 package File
::Rsync
::Mirror
::Recentfile
;
10 File::Rsync::Mirror::Recentfile - mirroring via rsync made efficient
23 $HAVE->{$package} = eval qq{ require $package; };
26 use File
::Basename
qw(dirname fileparse);
27 use File
::Copy
qw(cp);
28 use File
::Path
qw(mkpath);
30 use List
::Util
qw(first min);
31 use Scalar
::Util
qw(reftype);
36 use version
; our $VERSION = qv
('0.0.1');
39 use constant MAX_INT
=> ~0>>1; # anything better?
40 use constant DEFAULT_PROTOCOL
=> 1;
45 # maybe subclass if this mapping is bad?
50 B<!!!! PRE-ALPHA ALERT !!!!>
52 Nothing in here is believed to be stable, nothing yet intended for
53 public consumption. The plan is to provide a script in one of the next
54 releases that acts as a frontend for all the backend functionality.
55 Option and method names will very likely change.
57 For the rationale see the section BACKGROUND.
59 This is published only for developers of the (yet to be named)
62 Writer (of a single file):
64 use File::Rsync::Mirror::Recentfile;
65 my $fr = File::Rsync::Mirror::Recentfile->new
68 filenameroot => "RECENT",
69 comment => "These 'RECENT' files are part of a test of a new CPAN mirroring concept. Please ignore them for now.",
70 localroot => "/home/ftp/pub/PAUSE/authors/",
71 aggregator => [qw(1d 1W 1M 1Q 1Y Z)],
73 $rf->update("/home/ftp/pub/PAUSE/authors/id/A/AN/ANDK/CPAN-1.92_63.tar.gz","new");
77 my $rf = File::Rsync::Mirror::Recentfile->new
79 filenameroot => "RECENT",
80 ignore_link_stat_errors => 1,
82 localroot => "/home/ftp/pub/PAUSE/authors",
84 remote_host => "pause.perl.org",
85 remote_module => "authors",
88 'rsync-path' => '/usr/bin/rsync',
91 'omit-dir-times' => 1,
98 Aggregator (usually the writer):
100 my $rf = File::Rsync::Mirror::Recentfile->new_from_file ( $file );
109 =head2 my $obj = CLASS->new(%hash)
111 Constructor. On every argument pair the key is a method name and the
112 value is an argument to that method name.
114 If a recentfile for this resource already exists, metadata that are
115 not defined by the constructor will be fetched from there as soon as
116 it is being read by recent_events().
121 my($class, @args) = @_;
122 my $self = bless {}, $class;
124 my($method,$arg) = splice @args, 0, 2;
125 $self->$method($arg);
127 unless (defined $self->protocol) {
128 $self->protocol(DEFAULT_PROTOCOL
);
130 unless (defined $self->filenameroot) {
131 $self->filenameroot("RECENT");
133 unless (defined $self->serializer_suffix) {
134 $self->serializer_suffix(".yaml");
139 =head2 my $obj = CLASS->new_from_file($file)
141 Constructor. $file is a I<recentfile>.
146 my($class, $file) = @_;
147 my $self = bless {}, $class;
148 $self->_rfile($file);
150 my $serialized = do { open my $fh, $file or die "Could not open '$file': $!";
154 # XXX: we can skip this step when the metadata are sufficient, but
155 # we cannot parse the file without some magic stuff about
158 my($name,$path) = fileparse
$file;
159 my $symlink = readlink $file;
160 if ($symlink =~ m
|/|) {
161 die "FIXME: filenames containing '/' not supported, got $symlink";
163 $file = File
::Spec
->catfile ( $path, $symlink );
165 my($name,$path,$suffix) = fileparse
$file, keys %serializers;
166 $self->serializer_suffix($suffix);
167 $self->localroot($path);
168 die "Could not determine file format from suffix" unless $suffix;
170 if ($suffix eq ".yaml") {
172 $deserialized = YAML
::Syck
::LoadFile
($file);
173 } elsif ($HAVE->{"Data::Serializer"}) {
174 my $serializer = Data
::Serializer
->new
175 ( serializer
=> $serializers{$suffix} );
176 $deserialized = $serializer->raw_deserialize($serialized);
178 die "Data::Serializer not installed, cannot proceed with suffix '$suffix'";
180 while (my($k,$v) = each %{$deserialized->{meta
}}) {
181 next if $k ne lc $k; # "Producers"
184 unless (defined $self->protocol) {
185 $self->protocol(DEFAULT_PROTOCOL
);
199 "_current_tempfile_fh",
211 split /\n/, <<'=cut'; push @accessors, grep {s/^=item\s+//} @pod_lines; }
217 A list of interval specs that tell the aggregator which I<recentfile>s
222 The name of a method to canonize the path before rsyncing. Only
223 supported value is C<naive_path_normalize>. Defaults to that.
227 A comment about this tree and setup.
231 A reference to a File::Rsync::Mirror::Recentfile::Done object that
232 keeps track of rsync activities. Only used/needed when we are a
237 The (prefix of the) filename we use for this I<recentfile>. Defaults to
242 Timestamp remembering when we mirrored this recentfile the last time.
243 Only relevant for slaves.
245 =item ignore_link_stat_errors
247 If set to true, rsync errors are ignored that complain about link stat
248 errors. These seem to happen only when there are files missing at the
249 origin. In race conditions this can always happen, so it is
250 recommended to set this value to true.
254 If set to true, this object will fetch a new recentfile from remote
255 when the timespan between the last mirror (see have_mirrored) and now
256 is too large (currently hardcoded arbitrary 420 seconds).
260 After how many seconds shall we die if we cannot lock a I<recentfile>?
261 Defaults to 600 seconds.
265 When mirror_loop is called, this accessor can specify how much time
266 every loop shall at least take. If the work of a loop is done before
267 that time has gone, sleeps for the rest of the time. Defaults to
268 arbitrary 42 seconds.
270 =item max_files_per_connection
272 Maximum number of files that are transferred on a single rsync call.
273 Setting it higher means higher performance at the price of holding
274 connections longer and potentially disturbing other users in the pool.
275 Defaults to the arbitrary value 42.
277 =item max_rsync_errors
279 When rsync operations encounter that many errors without any resetting
280 success in between, then we die. Defaults to -1 which means we run
281 forever ignoring all rsync errors.
285 Hashref denoting when this recentfile has been merged into some other
290 Hashref remembering when we read the recent_events from this file the
291 last time and what the timespan was.
295 When the RECENT file format changes, we increment the protocol. We try
296 to support older protocols in later releases.
300 The host we are mirroring from. Leave empty for the local filesystem.
304 Rsync servers have so called modules to separate directory trees from
305 each other. Put here the name of the module under which we are
306 mirroring. Leave empty for local filesystem.
310 Things like compress, links, times or checksums. Passed in to the
311 File::Rsync object used to run the mirror.
313 =item serializer_suffix
315 Mostly untested accessor. The only well tested format for
316 I<recentfile>s at the moment is YAML. It is used with YAML::Syck via
317 Data::Serializer. But in principle other formats are supported as
318 well. See section SERIALIZERS below.
320 =item sleep_per_connection
322 Sleep that many seconds (floating point OK) after every chunk of rsyncing
323 has finished. Defaults to arbitrary 0.42.
327 Time to live. Number of seconds after which this recentfile must be
328 fetched again from the origin server. Only relevant for slaves.
329 Defaults to arbitrary 24.2 seconds.
333 Boolean to turn on a bit verbosity.
339 use accessors
@accessors;
343 =head2 (void) $obj->aggregate
345 Takes all intervals that are collected in the accessor called
346 aggregator. Sorts them by actual length of the interval.
347 Removes those that are shorter than our own interval. Then merges this
348 object into the next larger object. The merging continues upwards
349 as long as the next I<recentfile> is old enough to warrant a merge.
351 If a merge is warranted is decided according to the interval of the
352 previous interval so that larger files are not so often updated as
355 Here is an example to illustrate the behaviour. Given aggregators
361 1h updates 1d on every call to aggregate()
362 1d updates 1W earliest after 1h
363 1W updates 1M earliest after 1d
364 1M updates 1Q earliest after 1W
365 1Q updates 1Y earliest after 1M
366 1Y updates Z earliest after 1Q
368 Note that all but the smallest recentfile get updated at an arbitrary
369 rate and as such are quite useless on their own.
375 my @aggs = sort { $a->{secs
} <=> $b->{secs
} }
376 grep { $_->{secs
} >= $self->interval_secs }
377 map { { interval
=> $_, secs
=> $self->interval_secs($_)} }
378 $self->interval, @
{$self->aggregator || []};
379 $aggs[0]{object
} = $self;
380 AGGREGATOR
: for my $i (0..$#aggs-1) {
381 my $this = $aggs[$i]{object
};
382 my $next = Storable
::dclone
$this;
383 $next->interval($aggs[$i+1]{interval
});
388 my $next_rfile = $next->rfile;
389 if (-e
$next_rfile) {
390 my $prev = $aggs[$i-1]{object
};
392 my $next_age = 86400 * -M
$next_rfile;
393 if ($next_age > $prev->interval_secs) {
402 $aggs[$i+1]{object
} = $next;
409 sub _debug_aggregate
{
411 my @aggs = sort { $a->{secs
} <=> $b->{secs
} }
412 map { { interval
=> $_, secs
=> $self->interval_secs($_)} }
413 $self->interval, @
{$self->aggregator || []};
415 for my $i (0..$#aggs) {
416 my $this = Storable
::dclone
$self;
417 $this->interval($aggs[$i]{interval
});
418 my $rfile = $this->rfile;
419 my @stat = stat $rfile;
420 push @
$report, [$rfile, map {$stat[$_]||"undef"} 7,9];
425 # (void) $self->_assert_symlink()
426 sub _assert_symlink
{
428 my $recentrecentfile = File
::Spec
->catfile
437 if ($Config{d_symlink
} eq "define") {
438 my $howto_create_symlink; # 0=no need; 1=straight symlink; 2=rename symlink
439 if (-l
$recentrecentfile) {
440 my $found_symlink = readlink $recentrecentfile;
441 if ($found_symlink eq $self->rfilename) {
444 $howto_create_symlink = 2;
447 $howto_create_symlink = 1;
449 if (1 == $howto_create_symlink) {
450 symlink $self->rfilename, $recentrecentfile or die "Could not create symlink '$recentrecentfile': $!"
452 unlink "$recentrecentfile.$$"; # may fail
453 symlink $self->rfilename, "$recentrecentfile.$$" or die "Could not create symlink '$recentrecentfile.$$': $!";
454 rename "$recentrecentfile.$$", $recentrecentfile or die "Could not rename '$recentrecentfile.$$' to $recentrecentfile: $!";
457 warn "Warning: symlinks not supported on this system, doing a copy instead\n";
458 unlink "$recentrecentfile.$$"; # may fail
459 cp
$self->rfilename, "$recentrecentfile.$$" or die "Could not copy to '$recentrecentfile.$$': $!";
460 rename "$recentrecentfile.$$", $recentrecentfile or die "Could not rename '$recentrecentfile.$$' to $recentrecentfile: $!";
464 =head2 $success = $obj->full_mirror
466 (TBD) Mirrors the whole remote site, starting with the smallest I<recentfile>,
467 switching to larger ones ...
473 die "FIXME: Not yet implemented";
476 =head2 $tempfilename = $obj->get_remote_recentfile_as_tempfile ($rfilename)
478 =head2 $tempfilename = $obj->get_remote_recentfile_as_tempfile ()
480 Stores the remote I<recentfile> locally as a tempfile. $rfilename must
481 be a plain filename without path separators. The second form fetches
482 the file with the default name. The caller is responsible to remove
485 Note: if you're intending to act as an rsync server for other slaves,
486 then you must prefer this method to mirror (and read) recentfiles over
487 get_remotefile(). Otherwise downstream mirrors would expect you to
488 have files that you do not have yet.
492 sub get_remote_recentfile_as_tempfile
{
493 my($self, $rfilename) = @_;
494 mkpath
$self->localroot;
497 $self->_use_tempfile (1);
498 } elsif ( $self->_use_tempfile() ) {
499 return $self->_current_tempfile if ! $self->ttl_reached;
500 $fh = $self->_current_tempfile_fh;
501 $rfilename = $self->rfilename;
503 $rfilename = $self->rfilename;
505 die "Alert: illegal filename[$rfilename] contains a slash" if $rfilename =~ m
|/|;
508 $dst = $self->_current_tempfile;
510 $fh = File
::Temp
->new
511 (TEMPLATE
=> sprintf(".%s-XXXX",
514 DIR
=> $self->localroot,
515 SUFFIX
=> $self->serializer_suffix,
516 UNLINK
=> $self->_use_tempfile,
518 if ($self->_use_tempfile) {
519 $self->_current_tempfile_fh ($fh); # delay self destruction
521 $dst = $fh->filename;
522 $self->_current_tempfile ($dst);
523 my $rfile = eval { $self->rfile; }; # may fail (RECENT.recent has no rfile)
524 if (defined $rfile && -e
$rfile) {
525 # saving on bandwidth. Might need to be configurable
526 # $self->bandwidth_is_cheap?
527 cp
$rfile, $dst or die "Could not copy '$rfile' to '$dst': $!"
534 if ($self->verbose) {
535 my $doing = -e
$dst ?
"Syncing" : "Getting";
538 "%s (1/1) temporary %s ... ",
543 while (!$self->rsync->exec(
547 $self->register_rsync_error ($self->rsync->err);
549 $self->have_mirrored (Time
::HiRes
::time);
550 $self->un_register_rsync_error ();
551 if ($self->verbose) {
552 print STDERR
"DONE\n";
555 chmod $mode, $dst or die "Could not chmod $mode '$dst': $!";
559 =head2 $localpath = $obj->get_remotefile ( $relative_path )
561 Rsyncs one single remote file to local filesystem.
563 Note: no locking is done on this file. Any number of processes may
566 Note II: do not use for recentfiles. If you are a cascading
567 slave/server combination, it would confuse other slaves. They would
568 expect the contents of these recentfiles to be available. Use
569 get_remote_recentfile_as_tempfile() instead.
574 my($self, $path) = @_;
575 my $dst = File
::Spec
->catfile($self->localroot, $path);
577 if ($self->verbose) {
578 my $doing = -e
$dst ?
"Syncing" : "Getting";
586 while (!$self->rsync->exec(
592 $self->register_rsync_error ($self->rsync->err);
594 $self->un_register_rsync_error ();
595 if ($self->verbose) {
596 print STDERR
"DONE\n";
601 =head2 $obj->interval ( $interval_spec )
603 Get/set accessor. $interval_spec is a string and described below in
604 the section INTERVAL SPEC.
609 my ($self, $interval) = @_;
611 $self->_interval($interval);
612 $self->_rfile(undef);
614 $interval = $self->_interval;
615 unless (defined $interval) {
616 # do not ask the $self too much, it recurses!
618 Carp
::confess
("Alert: interval undefined for '".$self."'. Cannot continue.");
623 =head2 $secs = $obj->interval_secs ( $interval_spec )
625 $interval_spec is described below in the section INTERVAL SPEC. If
626 empty defaults to the inherent interval for this object.
631 my ($self, $interval) = @_;
632 $interval ||= $self->interval;
633 unless (defined $interval) {
634 die "interval_secs() called without argument on an object without a declared one";
636 my ($n,$t) = $interval =~ /^(\d*)([smhdWMQYZ]$)/ or
637 die "Could not determine seconds from interval[$interval]";
638 if ($interval eq "Z") {
640 } elsif (exists $seconds{$t} and $n =~ /^\d+$/) {
641 return $seconds{$t}*$n;
643 die "Invalid interval specification: n[$n]t[$t]";
647 =head2 $obj->localroot ( $localroot )
649 Get/set accessor. The local root of the tree.
654 my ($self, $localroot) = @_;
656 $self->_localroot($localroot);
657 $self->_rfile(undef);
659 $localroot = $self->_localroot;
662 =head2 $ret = $obj->local_path($path_found_in_recentfile)
664 Combines the path to our local mirror and the path of an object found
665 in this I<recentfile>. In other words: the target of a mirror operation.
667 Implementation note: We split on slashes and then use
668 File::Spec::catfile to adjust to the local operating system.
673 my($self,$path) = @_;
674 unless (defined $path) {
675 # seems like a degenerated case
676 return $self->localroot;
678 my @p = split m
|/|, $path;
679 File
::Spec
->catfile($self->localroot,@p);
682 =head2 (void) $obj->lock
684 Locking is implemented with an C<mkdir> on a locking directory
685 (C<.lock> appended to $rfile).
691 # not using flock because it locks on filehandles instead of
692 # old school ressources.
693 my $locked = $self->_is_locked and return;
694 my $rfile = $self->rfile;
695 # XXX need a way to allow breaking the lock
697 my $locktimeout = $self->locktimeout || 600;
698 while (not mkdir "$rfile.lock") {
699 Time
::HiRes
::sleep 0.01;
700 if (time - $start > $locktimeout) {
701 die "Could not acquire lockdirectory '$rfile.lock': $!";
704 $self->_is_locked (1);
707 =head2 $ret = $obj->merge ($other)
709 Bulk update of this object with another one. It's intended (but not
710 enforced) to only merge smaller and younger $other objects into the
711 current one. If this file is a C<Z> file, then we do not merge in
712 objects of type C<delete>. But if we encounter an object of type
713 delete we delete the corresponding C<new> object.
718 my($self,$other) = @_;
720 my $other_recent = $other->recent_events || [];
722 my $my_recent = $self->recent_events || [];
724 # calculate the target time span
725 my $epoch = $other_recent->[0] ?
$other_recent->[0]{epoch
} : $my_recent->[0] ?
$my_recent->[0]{epoch
} : undef;
726 my $oldest_allowed = 0;
728 if (my $merged = $self->merged) {
729 my $secs = $self->interval_secs();
730 $oldest_allowed = min
($epoch - $secs, $merged->{epoch
});
732 # throw away outsiders
733 while (@
$my_recent && $my_recent->[-1]{epoch
} < $oldest_allowed) {
740 for my $ev (@
$other_recent) {
741 my $epoch = $ev->{epoch
} || 0;
742 next if $epoch < $oldest_allowed;
743 my $path = $ev->{path
};
744 next if $have{$path}++;
745 if ( $self->interval eq "Z"
746 and $ev->{type
} eq "delete") {
749 push @
$recent, { epoch
=> $ev->{epoch
}, path
=> $path, type
=> $ev->{type
} };
752 push @
$recent, grep { !$have{$_->{path
}}++ } @
$my_recent;
753 $self->write_recent($recent);
756 time => Time
::HiRes
::time, # not used anywhere
757 epoch
=> $epoch, # used in oldest_allowed
758 into_interval
=> $self->interval, # not used anywhere
760 $other->write_recent($other_recent);
764 =head2 $hashref = $obj->meta_data
766 Returns the hashref of metadata that the server has to add to the
773 my $ret = $self->{meta
};
789 # XXX need to reset the Producer if I am a writer, keep it when I
791 $ret->{Producers
} ||= {
792 __PACKAGE__
, "$VERSION", # stringified it looks better
794 'time', Time
::HiRes
::time,
799 =head2 $success = $obj->mirror ( %options )
801 Mirrors the files in this I<recentfile> as reported by
802 C<recent_events>. Options named C<after>, C<before>, C<max>, and
803 C<skip-deletes> are passed through to the L<recent_events> call. The
804 boolean option C<piecemeal>, if true, causes C<mirror> to only rsync
805 C<max_files_per_connection> and keep track of the rsynced files so
806 that future calls will rsync different files until all files are
812 my($self, %options) = @_;
813 my $trecentfile = $self->get_remote_recentfile_as_tempfile();
814 $self->_use_tempfile (1);
815 my %passthrough = map { ($_ => $options{$_}) } qw(before after max skip-deletes);
816 my ($recent_events) = $self->recent_events(%passthrough);
817 my(@error, @collector, @icollector);
819 my $last_item = $#$recent_events;
820 my $done = $self->done;
822 require File
::Rsync
::Mirror
::Recentfile
::Done
;
823 $done = File
::Rsync
::Mirror
::Recentfile
::Done
->new();
824 $self->done ( $done );
826 ITEM
: for my $i ($first_item..$last_item) {
827 my $recent_event = $recent_events->[$i];
828 next if $done->covered ( $recent_event->{epoch
} );
829 my $dst = $self->local_path($recent_event->{path
});
830 if ($recent_event->{type
} eq "new"){
831 if ($self->verbose) {
832 my $doing = -e
$dst ?
"Syncing" : "Getting";
835 "%s (%d/%d) %s ... ",
839 $recent_event->{path
},
842 my $max_files_per_connection = $self->max_files_per_connection || 42;
844 if ($self->verbose) {
847 push @collector, $recent_event->{path
};
848 push @icollector, $i;
849 if (@collector == $max_files_per_connection) {
850 $success = eval { $self->mirror_path(\
@collector) };
852 $done->register($recent_events, \
@icollector);
854 my $sleep = $self->sleep_per_connection;
855 $sleep = 0.42 unless defined $sleep;
856 Time
::HiRes
::sleep $sleep;
860 if (!$success || $@
) {
861 warn "Warning: Error while mirroring: $@";
865 if ($self->verbose) {
866 print STDERR
"DONE\n";
868 } elsif ($recent_event->{type
} eq "delete") {
869 if ($options{'skip-deletes'}) {
871 if (-l
$dst or not -d _
) {
872 unless (unlink $dst) {
874 Carp
::cluck
( "Warning: Error while unlinking '$dst': $!" );
877 unless (rmdir $dst) {
879 Carp
::cluck
( "Warning: Error on rmdir '$dst': $!" );
883 $done->register($recent_events, [$i]);
885 warn "Warning: invalid upload type '$recent_event->{type}'";
889 my $success = eval { $self->mirror_path(\
@collector) };
891 $done->register($recent_events, \
@icollector);
893 if (!$success || $@
) {
894 warn "Warning: Unknown error while mirroring: $@";
898 if ($self->verbose) {
899 print STDERR
"DONE\n";
902 my $rfile = $self->rfile;
903 unless (rename $trecentfile, $rfile) {
905 Carp
::confess
("Could not rename '$trecentfile' to '$rfile': $!");
907 $self->_use_tempfile (0);
908 if (my $ctfh = $self->_current_tempfile_fh) {
909 $ctfh->unlink_on_destroy (0);
910 $self->_current_tempfile_fh (undef);
915 =head2 (void) $obj->mirror_loop
917 Run mirror in an endless loop. See the accessor C<loopinterval>. XXX
918 What happens/should happen if we miss the interval during a single loop?
924 my $iteration_start = time;
927 $SIG{INT
} = sub { $Signal++ };
928 my $loopinterval = $self->loopinterval || 42;
929 my $after = -999999999;
931 $self->mirror($after);
932 last LOOP
if $Signal;
933 my $re = $self->recent_events;
934 $after = $re->[0]{epoch
};
935 if ($self->verbose) {
939 if (time - $iteration_start < $loopinterval) {
940 sleep $iteration_start + $loopinterval - time;
942 if ($self->verbose) {
949 =head2 $success = $obj->mirror_path ( $arrref | $path )
951 If the argument is a scalar it is treated as a path. The remote path
952 is mirrored into the local copy. $path is the path found in the
953 I<recentfile>, i.e. it is relative to the root directory of the
956 If the argument is an array reference then all elements are treated as
957 a path below the current tree and all are rsynced with a single
958 command (and a single connection).
963 my($self,$path) = @_;
964 # XXX simplify the two branches such that $path is treated as
965 # [$path] maybe even demand the argument as an arrayref to
966 # simplify docs and code. (rsync-over-recentfile-2.pl uses the
968 if (ref $path and ref $path eq "ARRAY") {
969 my $dst = $self->localroot;
971 my($fh) = File
::Temp
->new(TEMPLATE
=> sprintf(".%s-XXXX",
972 lc $self->filenameroot,
981 $fh->unlink_on_destroy(1);
982 while (!$self->rsync->exec
988 'files-from' => $fh->filename,
990 my($err) = $self->rsync->err;
991 if ($self->ignore_link_stat_errors && $err =~ m{^ rsync: \s link_stat }x ) {
992 if ($self->verbose) {
993 warn "Info: ignoring link_stat error '$err'";
997 $self->register_rsync_error ($err);
999 $self->un_register_rsync_error ();
1001 my $dst = $self->local_path($path);
1002 mkpath dirname
$dst;
1003 while (!$self->rsync->exec
1011 my($err) = $self->rsync->err;
1012 if ($self->ignore_link_stat_errors && $err =~ m{^ rsync: \s link_stat }x ) {
1013 if ($self->verbose) {
1014 warn "Info: ignoring link_stat error '$err'";
1018 $self->register_rsync_error ($err);
1020 $self->un_register_rsync_error ();
1025 sub _my_current_rfile
{
1028 if ($self->_use_tempfile) {
1029 $rfile = $self->_current_tempfile;
1031 $rfile = $self->rfile;
1036 =head2 $path = $obj->naive_path_normalize ($path)
1038 Takes an absolute unix style path as argument and canonicalizes it to
1039 a shorter path if possible, removing things like double slashes or
1040 C</./> and removes references to C<../> directories to get a shorter
1041 unambiguos path. This is used to make the code easier that determines
1042 if a file passed to C<upgrade()> is indeed below our C<localroot>.
1046 sub naive_path_normalize
{
1047 my($self,$path) = @_;
1049 1 while $path =~ s
|/[^/]+/\.\./|/|;
1054 =head2 $ret = $obj->read_recent_1 ( $data )
1056 Delegate of C<recent_events()> on protocol 1
1061 my($self, $data) = @_;
1062 return $data->{recent
};
1065 =head2 $array_ref = $obj->recent_events ( %options )
1067 Note: the code relies on the resource being written atomically. We
1068 cannot lock because we may have no write access. If the caller has
1069 write access (eg. aggregate() or update()), it has to care for any
1072 If $options{after} is specified, only file events after this timestamp
1075 If $options{before} is specified, only file events before this
1076 timestamp are returned.
1078 IF $options{'skip-deletes'} is specified, no files-to-be-deleted will
1081 If $options{max} is specified only this many events are returned.
1083 If $options{info} is specified, it must be a hashref. This hashref
1084 will be filled with metadata about the unfiltered recent_events of
1085 this object, in key C<first> there is the first item, in key C<last>
1091 my ($self, %options) = @_;
1092 my $info = $options{info
};
1094 and (!$self->have_mirrored || Time
::HiRes
::time-$self->have_mirrored>420)) {
1095 $self->get_remote_recentfile_as_tempfile;
1097 my $rfile_or_tempfile = $self->_my_current_rfile or return [];
1098 -e
$rfile_or_tempfile or return [];
1099 my $suffix = $self->serializer_suffix;
1101 if ($suffix eq ".yaml") {
1103 YAML
::Syck
::LoadFile
($rfile_or_tempfile);
1104 } elsif ($HAVE->{"Data::Serializer"}) {
1105 my $serializer = Data
::Serializer
->new
1106 ( serializer
=> $serializers{$suffix} );
1109 open my $fh, $rfile_or_tempfile or die "Could not open: $!";
1113 $serializer->raw_deserialize($serialized);
1115 die "Data::Serializer not installed, cannot proceed with suffix '$suffix'";
1119 if ($err or !$data) {
1123 if (reftype
$data eq 'ARRAY') { # protocol 0
1126 my $meth = sprintf "read_recent_%d", $data->{meta
}{protocol
};
1127 # we may be reading meta for the first time
1128 while (my($k,$v) = each %{$data->{meta
}}) {
1129 next if $k ne lc $k; # "Producers"
1130 next if defined $self->$k;
1133 $re = $self->$meth ($data);
1134 my @stat = stat $rfile_or_tempfile or die "Cannot stat '$rfile_or_tempfile': $!";
1135 my $minmax = { mtime
=> $stat[9] };
1137 $minmax->{min
} = $re->[-1]{epoch
};
1138 $minmax->{max
} = $re->[0]{epoch
};
1140 $self->minmax ( $minmax );
1142 return $re unless defined $options{after
}; # XXX same for before and max
1143 my $last_item = $#$re;
1145 $info->{first
} = $re->[0];
1146 $info->{last} = $re->[-1];
1148 if (defined $options{after
}) {
1149 if ($re->[0]{epoch
} > $options{after
}) {
1152 {$re->[$_]{epoch
} <= $options{after
}}
1162 if (defined $options{before
}) {
1163 if ($re->[0]{epoch
} > $options{before
}) {
1166 {$re->[$_]{epoch
} < $options{before
}}
1175 my @rre = splice @
$re, $first_item, 1+$last_item-$first_item;
1176 if ($options{'skip-deletes'}) {
1177 @rre = grep { $_->{type
} ne "delete" } @rre;
1179 if ($options{max
} && @rre > $options{max
}) {
1180 @rre = splice @rre, 0, $options{max
};
1185 =head2 $ret = $obj->rfilename
1187 Just the basename of our I<recentfile>, composed from C<filenameroot>,
1188 a dash, C<interval>, and C<serializer_suffix>. E.g. C<RECENT-6h.yaml>
1194 my $file = sprintf("%s-%s%s",
1195 $self->filenameroot,
1197 $self->serializer_suffix,
1202 =head2 $str = $self->remote_dir
1204 The directory we are mirroring from.
1209 my($self, $set) = @_;
1211 $self->_remote_dir ($set);
1213 my $x = $self->_remote_dir;
1214 $self->is_slave (1);
1218 =head2 $str = $obj->remoteroot
1220 =head2 (void) $obj->remoteroot ( $set )
1222 Get/Set the composed prefix needed when rsyncing from a remote module.
1223 If remote_host, remote_module, and remote_dir are set, it is composed
1229 my($self, $set) = @_;
1231 $self->_remoteroot($set);
1233 my $remoteroot = $self->_remoteroot;
1234 unless (defined $remoteroot) {
1235 $remoteroot = sprintf
1238 defined $self->remote_host ?
($self->remote_host."::") : "",
1239 defined $self->remote_module ?
($self->remote_module."/") : "",
1240 defined $self->remote_dir ?
$self->remote_dir : "",
1242 $self->_remoteroot($remoteroot);
1247 =head2 my $rfile = $obj->rfile
1249 Returns the full path of the I<recentfile>
1255 my $rfile = $self->_rfile;
1256 return $rfile if defined $rfile;
1257 $rfile = File
::Spec
->catfile
1261 $self->_rfile ($rfile);
1265 =head2 $rsync_obj = $obj->rsync
1267 The File::Rsync object that this object uses for communicating with an
1274 my $rsync = $self->_rsync;
1275 unless (defined $rsync) {
1276 my $rsync_options = $self->rsync_options || {};
1277 if ($HAVE->{"File::Rsync"}) {
1278 $rsync = File
::Rsync
->new($rsync_options);
1279 $self->_rsync($rsync);
1281 die "File::Rsync required for rsync operations. Cannot continue";
1287 =head2 (void) $obj->register_rsync_error($err)
1289 =head2 (void) $obj->un_register_rsync_error()
1291 Register_rsync_error is called whenever the File::Rsync object fails
1292 on an exec (say, connection doesn't succeed). It issues a warning and
1293 sleeps for an increasing amount of time. Un_register_rsync_error
1294 resets the error count. See also accessor C<max_rsync_errors>.
1299 my $no_success_count = 0;
1300 my $no_success_time = 0;
1301 sub register_rsync_error
{
1302 my($self, $err) = @_;
1304 $no_success_time = time;
1305 $no_success_count++;
1306 my $max_rsync_errors = $self->max_rsync_errors;
1307 $max_rsync_errors = -1 unless defined $max_rsync_errors;
1308 if ($max_rsync_errors>=0 && $no_success_count >= $max_rsync_errors) {
1311 "Alert: Error while rsyncing: '%s', error count: %d, exiting now,",
1316 my $sleep = 12 * $no_success_count;
1317 $sleep = 120 if $sleep > 120;
1320 "Warning: %s, Error while rsyncing: '%s', sleeping %d",
1321 scalar(localtime($no_success_time)),
1327 sub un_register_rsync_error
{
1329 $no_success_time = 0;
1330 $no_success_count = 0;
1334 =head2 $clone = $obj->_sparse_clone
1336 Clones just as much from itself that it does not hurt. Experimental method.
1342 my $new = bless {}, ref $self;
1358 $o = Storable
::dclone
$o if ref $o;
1364 =head2 $boolean = OBJ->ttl_reached ()
1370 my $have_mirrored = $self->have_mirrored;
1371 my $now = Time
::HiRes
::time;
1372 my $ttl = $self->ttl;
1373 $ttl = 24.2 unless defined $ttl;
1374 if ($now > $have_mirrored + $ttl) {
1380 =head2 (void) $obj->unlock()
1382 Unlocking is implemented with an C<rmdir> on a locking directory
1383 (C<.lock> appended to $rfile).
1389 return unless $self->_is_locked;
1390 my $rfile = $self->rfile;
1391 rmdir "$rfile.lock";
1392 $self->_is_locked (0);
1395 =head2 $ret = $obj->update ($path, $type)
1397 Enter one file into the local I<recentfile>. $path is the (usually
1398 absolute) path. If the path is outside the I<our> tree, then it is
1401 $type is one of C<new> or C<delete>.
1406 my($self,$path,$type) = @_;
1407 die "update called without path argument" unless defined $path;
1408 die "update called without type argument" unless defined $type;
1409 die "update called with illegal type argument: $type" unless $type =~ /(new|delete)/;
1410 my $canonmeth = $self->canonize;
1411 unless ($canonmeth) {
1412 $canonmeth = "naive_path_normalize";
1414 $path = $self->$canonmeth($path);
1415 my $lrd = $self->localroot;
1416 if ($path =~ s
|^\Q
$lrd\E
||) {
1418 my $interval = $self->interval;
1419 my $secs = $self->interval_secs();
1420 my $epoch = Time
::HiRes
::time;
1422 my $recent = $self->recent_events;
1424 my $oldest_allowed = 0;
1425 if (my $merged = $self->merged) {
1426 my $secs = $self->interval_secs();
1427 $oldest_allowed = min
($epoch - $secs, $merged->{epoch
});
1429 TRUNCATE
: while (@
$recent) {
1430 if ($recent->[-1]{epoch
} < $oldest_allowed) {
1436 # remove older duplicates of this $path, irrespective of $type:
1437 $recent = [ grep { $_->{path
} ne $path } @
$recent ];
1439 unshift @
$recent, { epoch
=> $epoch, path
=> $path, type
=> $type };
1440 $self->write_recent($recent);
1441 $self->_assert_symlink;
1448 True if this object has mirrored the complete interval covered by the
1457 return 0 if $self->ttl_reached;
1459 # look if recentfile has unchanged timestamp
1460 my $minmax = $self->minmax;
1461 if (exists $minmax->{mtime
}) {
1462 my $rfile = $self->_my_current_rfile;
1463 my @stat = stat $rfile;
1464 my $mtime = $stat[9];
1465 if ($mtime > $minmax->{mtime
}) {
1468 return $self->done->covered(@
$minmax{qw(min max)});
1474 =head2 $obj->write_recent ($recent_files_arrayref)
1476 Writes a I<recentfile> based on the current reflection of the current
1477 state of the tree limited by the current interval.
1482 my ($self,$recent) = @_;
1483 die "write_recent called without argument" unless defined $recent;
1484 my $meth = sprintf "write_%d", $self->protocol;
1485 $self->$meth($recent);
1488 =head2 $obj->write_0 ($recent_files_arrayref)
1490 Delegate of C<write_recent()> on protocol 0
1495 my ($self,$recent) = @_;
1496 my $rfile = $self->rfile;
1497 YAML
::Syck
::DumpFile
("$rfile.new",$recent);
1498 rename "$rfile.new", $rfile or die "Could not rename to '$rfile': $!";
1501 =head2 $obj->write_1 ($recent_files_arrayref)
1503 Delegate of C<write_recent()> on protocol 1
1508 my ($self,$recent) = @_;
1509 my $rfile = $self->rfile;
1510 my $suffix = $self->serializer_suffix;
1512 meta
=> $self->meta_data,
1516 if ($suffix eq ".yaml") {
1517 $serialized = YAML
::Syck
::Dump
($data);
1518 } elsif ($HAVE->{"Data::Serializer"}) {
1519 my $serializer = Data
::Serializer
->new
1520 ( serializer
=> $serializers{$suffix} );
1521 $serialized = $serializer->raw_serialize($data);
1523 die "Data::Serializer not installed, cannot proceed with suffix '$suffix'";
1525 open my $fh, ">", "$rfile.new" or die "Could not open >'$rfile.new': $!";
1526 print $fh $serialized;
1527 close $fh or die "Could not close '$rfile.new': $!";
1528 rename "$rfile.new", $rfile or die "Could not rename to '$rfile': $!";
1533 split /\n/, <<'=cut'; %serializers = map { eval } grep {s/^=item\s+C<<(.+)>>$/$1/} @pod_lines; }
1535 =head1 THE ARCHITECTURE OF A COLLECTION OF RECENTFILES
1537 The idea is that we want to have a short file that records really
1538 recent changes. So that a fresh mirror can be kept fresh as long as
1539 the connectivity is given. Then we want longer files that record the
1540 history before. So when the mirror falls behind the update period
1541 reflected in the shortest file, it can switch to the next one. And if
1542 this is not long enough we want another one, again a bit longer. And
1543 we want one that completes the history back to the oldest file. For
1544 practical reasons the timespans of these files must overlap a bit and
1545 to keep the bandwidth necessities low they must not be
1546 updated too frequently. That's the basic idea. The following
1547 example represents a tree that has a few updates every day:
1549 RECENT.recent -> RECENT-1h.yaml
1558 The first file is the principal file, in so far it is the one that is
1559 written first after a filesystem change. Usually a symlink links to it
1560 with a filename that has the same filenameroot and the suffix
1561 C<.recent>. On systems that do not support symlinks there is a plain
1562 copy maintained instead.
1564 The last file, the Z file, contains the complementary files that are
1565 in none of the other files. It does never contain C<deletes>. Besides
1566 this it serves the role of a recovery mechanism or spill over pond.
1567 When things go wrong, it's a valuable controlling instance to hold the
1568 differences between the collection of limited interval files and the
1571 =head2 A SINGLE RECENTFILE
1573 A I<recentfile> consists of a hash that has two keys: C<meta> and
1574 C<recent>. The C<meta> part has metadata and the C<recent> part has a
1575 list of fileobjects.
1577 =head2 THE META PART
1579 Here we find things that are pretty much self explaining: all
1580 lowercase attributes are accessors and as such explained somewhere
1581 above in this manpage. The uppercase attribute C<Producers> contains
1582 version information about involved software components. Nothing to
1583 worry about as I believe.
1585 =head2 THE RECENT PART
1587 This is the interesting part. Every entry refers to some filesystem
1588 change (with path, epoch, type). The epoch value is the point in time
1589 when some change was I<registered>. Do not be tempted to believe that
1590 the entry has a direct relation to something like modification time or
1591 change time on the filesystem level. The timestamp (I<epoch> element)
1592 is a floating point number and does practically never correspond
1593 exactly to the data recorded in the filesystem but rather to the time
1594 when some process succeeded to report to the I<recentfile> mechanism
1595 that something has changed. This is why many parts of the code refer
1596 to I<events>, because we merely try to record the I<event> of the
1597 discovery of a change, not the time of the change itself.
1599 All these entries can be devided into two types (denoted by the
1600 C<type> attribute): C<new>s and C<delete>s. Changes and creations are
1601 C<new>s. Deletes are C<delete>s.
1603 Another distinction is for objects with an epoch timestamp and others
1604 without. All files that were already existing on the filesystem before
1605 the I<recentfile> mechanism was installed, get recorded with a
1608 Besides an C<epoch> and a C<type> attribute we find a third one:
1609 C<path>. This path is relative to the directory we find the
1612 The order of the entries in the I<recentfile> is by decreasing epoch
1613 attribute. These are either 0 or a unique floating point number. They
1614 are zero for events that were happening either before the time that
1615 the I<recentfile> mechanism was set up or were left undiscovered for a
1616 while and never handed over to update(). They are floating point
1617 numbers for all events being regularly handed to update(). And when
1618 the server has ntp running correctly, then the timestamps are
1619 actually decreasing and unique.
1621 =head1 CORRUPTION AND RECOVERY
1623 If the origin host breaks the promise to deliver consistent and
1624 complete I<recentfiles> then the way back to sanity shall be achieved
1625 through either the C<zloop> (still TBD) or traditional rsyncing
1626 between the hosts. For example, if the origin server forgets to deploy
1627 ntp and the clock on it jumps backwards some day, then this would
1628 probably go unnoticed for a while and many software components that
1629 rely on the time never running backwards will make wrong decisions.
1630 After some time this accident would probably still be found in one of
1631 the I<recentfiles> but would become meaningless as soon as a mirror
1632 has run through the sanitizing procedures. Same goes for origin hosts
1633 that forget to include or deliberately omit some files.
1637 The following suffixes are supported and trigger the use of these
1642 =item C<< ".yaml" => "YAML::Syck" >>
1644 =item C<< ".json" => "JSON" >>
1646 =item C<< ".sto" => "Storable" >>
1648 =item C<< ".dd" => "Data::Dumper" >>
1656 split /\n/, <<'=cut'; %seconds = map { eval } grep {s/^=item\s+C<<(.+)>>$/$1/} @pod_lines; }
1658 =head1 INTERVAL SPEC
1660 An interval spec is a primitive way to express time spans. Normally it
1661 is composed from an integer and a letter.
1663 As a special case, a string that consists only of the single letter
1664 C<Z>, stands for unlimited time.
1666 The following letters express the specified number of seconds:
1672 =item C<< m => 60 >>
1674 =item C<< h => 60*60 >>
1676 =item C<< d => 60*60*24 >>
1678 =item C<< W => 60*60*24*7 >>
1680 =item C<< M => 60*60*24*30 >>
1682 =item C<< Q => 60*60*24*90 >>
1684 =item C<< Y => 60*60*24*365.25 >>
1692 This is about speeding up rsync operation on large trees to many
1693 places. Uses a small metadata cocktail and pull technology.
1695 =head2 NON-COMPETITORS
1697 File::Mirror JWU/File-Mirror/File-Mirror-0.10.tar.gz only local trees
1698 Mirror::YAML ADAMK/Mirror-YAML-0.03.tar.gz some sort of inner circle
1699 Net::DownloadMirror KNORR/Net-DownloadMirror-0.04.tar.gz FTP sites and stuff
1700 Net::MirrorDir KNORR/Net-MirrorDir-0.05.tar.gz dito
1701 Net::UploadMirror KNORR/Net-UploadMirror-0.06.tar.gz dito
1702 Pushmi::Mirror CLKAO/Pushmi-v1.0.0.tar.gz something SVK
1704 rsnapshot www.rsnapshot.org focus on backup
1705 csync www.csync.org more like unison
1709 The problem to solve which clusters and ftp mirrors and otherwise
1710 replicated datasets like CPAN share: how to transfer only a minimum
1711 amount of data to determine the diff between two hosts.
1713 Normally it takes a long time to determine the diff itself before it
1714 can be transferred. Known solutions at the time of this writing are
1715 csync2, and rsync 3 batch mode.
1717 For many years the best solution was csync2 which solves the
1718 problem by maintining a sqlite database on both ends and talking a
1719 highly sophisticated protocol to quickly determine which files to send
1720 and which to delete at any given point in time. Csync2 is often
1721 inconvenient because the act of syncing demands quite an intimate
1722 relationship between the sender and the receiver and suffers when the
1723 number of syncing sites is large or connections are unreliable.
1725 Rsync 3 batch mode works around these problems by providing rsync-able
1726 batch files which allow receiving nodes to replay the history of the
1727 other nodes. This reduces the need to have an incestuous relation but
1728 it has the disadvantage that these batch files replicate the contents
1729 of the involved files. This seems inappropriate when the nodes already
1730 have a means of communicating over rsync.
1732 rersyncrecent solves this problem with a couple of (usually 2-10)
1733 index files which cover different overlapping time intervals. The
1734 master writes these files and the clients can construct the full tree
1735 from the information contained in them. The most recent index file
1736 usually covers the last seconds or minutes or hours of the tree and
1737 depending on the needs, slaves can rsync every few seconds and then
1738 bring their trees in full sync.
1740 The rersyncrecent mode was developed for CPAN but I hope it is a
1741 convenient and economic general purpose solution. I'm looking forward
1742 to see a CPAN backbone that is only a few seconds behind PAUSE. And
1743 then ... the first FUSE based CPAN filesystem anyone?
1751 Please report any bugs or feature requests through the web interface
1753 L<http://rt.cpan.org/NoAuth/ReportBug.html?Queue=File-Rsync-Mirror-Recentfile>.
1754 I will be notified, and then you'll automatically be notified of
1755 progress on your bug as I make changes.
1759 You can find documentation for this module with the perldoc command.
1761 perldoc File::Rsync::Mirror::Recentfile
1763 You can also look for information at:
1767 =item * RT: CPAN's request tracker
1769 L<http://rt.cpan.org/NoAuth/Bugs.html?Dist=File-Rsync-Mirror-Recentfile>
1771 =item * AnnoCPAN: Annotated CPAN documentation
1773 L<http://annocpan.org/dist/File-Rsync-Mirror-Recentfile>
1775 =item * CPAN Ratings
1777 L<http://cpanratings.perl.org/d/File-Rsync-Mirror-Recentfile>
1781 L<http://search.cpan.org/dist/File-Rsync-Mirror-Recentfile>
1786 =head1 ACKNOWLEDGEMENTS
1788 Thanks to RJBS for module-starter.
1790 =head1 COPYRIGHT & LICENSE
1792 Copyright 2008 Andreas König, all rights reserved.
1794 This program is free software; you can redistribute it and/or modify it
1795 under the same terms as Perl itself.
1800 1; # End of File::Rsync::Mirror::Recentfile
1804 # cperl-indent-level: 4