1 package File
::Rsync
::Mirror
::Recentfile
;
10 File::Rsync::Mirror::Recentfile - mirroring via rsync made efficient
19 $HAVE->{$package} = eval qq{ require $package; };
22 use File
::Basename
qw(basename dirname fileparse);
23 use File
::Copy
qw(cp);
24 use File
::Path
qw(mkpath);
25 use File
::Rsync
::Mirror
::Recentfile
::FakeBigFloat
qw(:all);
27 use List
::Util
qw(first max min);
28 use Scalar
::Util
qw(reftype);
33 use version
; our $VERSION = qv
('0.0.8');
35 use constant MAX_INT
=> ~0>>1; # anything better?
36 use constant DEFAULT_PROTOCOL
=> 1;
41 # maybe subclass if this mapping is bad?
46 Writer (of a single file):
48 use File::Rsync::Mirror::Recentfile;
49 my $fr = File::Rsync::Mirror::Recentfile->new
52 filenameroot => "RECENT",
53 comment => "These 'RECENT' files are part of a test of a new CPAN mirroring concept. Please ignore them for now.",
54 localroot => "/home/ftp/pub/PAUSE/authors/",
55 aggregator => [qw(1d 1W 1M 1Q 1Y Z)],
57 $rf->update("/home/ftp/pub/PAUSE/authors/id/A/AN/ANDK/CPAN-1.92_63.tar.gz","new");
61 my $rf = File::Rsync::Mirror::Recentfile->new
63 filenameroot => "RECENT",
65 localroot => "/home/ftp/pub/PAUSE/authors",
67 remote_host => "pause.perl.org",
68 remote_module => "authors",
71 'rsync-path' => '/usr/bin/rsync',
74 'omit-dir-times' => 1,
81 Aggregator (usually the writer):
83 my $rf = File::Rsync::Mirror::Recentfile->new_from_file ( $file );
88 Lower level than F:R:M:Recent, handles one recentfile. Whereas a tree
89 is always composed of several recentfiles, controlled by the
90 F:R:M:Recent object. The Recentfile object has to do the bookkeeping
91 for a single timeslice.
97 =head1 CONSTRUCTORS / DESTRUCTOR
99 =head2 my $obj = CLASS->new(%hash)
101 Constructor. On every argument pair the key is a method name and the
102 value is an argument to that method name.
104 If a recentfile for this resource already exists, metadata that are
105 not defined by the constructor will be fetched from there as soon as
106 it is being read by recent_events().
111 my($class, @args) = @_;
112 my $self = bless {}, $class;
114 my($method,$arg) = splice @args, 0, 2;
115 $self->$method($arg);
117 unless (defined $self->protocol) {
118 $self->protocol(DEFAULT_PROTOCOL
);
120 unless (defined $self->filenameroot) {
121 $self->filenameroot("RECENT");
123 unless (defined $self->serializer_suffix) {
124 $self->serializer_suffix(".yaml");
129 =head2 my $obj = CLASS->new_from_file($file)
131 Constructor. $file is a I<recentfile>.
136 my($class, $file) = @_;
137 my $self = bless {}, $class;
138 $self->_rfile($file);
140 my $serialized = do { open my $fh, $file or die "Could not open '$file': $!";
144 # XXX: we can skip this step when the metadata are sufficient, but
145 # we cannot parse the file without some magic stuff about
148 my($name,$path) = fileparse
$file;
149 my $symlink = readlink $file;
150 if ($symlink =~ m
|/|) {
151 die "FIXME: filenames containing '/' not supported, got $symlink";
153 $file = File
::Spec
->catfile ( $path, $symlink );
155 my($name,$path,$suffix) = fileparse
$file, keys %serializers;
156 $self->serializer_suffix($suffix);
157 $self->localroot($path);
158 die "Could not determine file format from suffix" unless $suffix;
160 if ($suffix eq ".yaml") {
162 $deserialized = YAML
::Syck
::LoadFile
($file);
163 } elsif ($HAVE->{"Data::Serializer"}) {
164 my $serializer = Data
::Serializer
->new
165 ( serializer
=> $serializers{$suffix} );
166 $deserialized = $serializer->raw_deserialize($serialized);
168 die "Data::Serializer not installed, cannot proceed with suffix '$suffix'";
170 while (my($k,$v) = each %{$deserialized->{meta
}}) {
171 next if $k ne lc $k; # "Producers"
174 unless (defined $self->protocol) {
175 $self->protocol(DEFAULT_PROTOCOL
);
188 unless ($self->_current_tempfile_fh) {
189 if (my $tempfile = $self->_current_tempfile) {
191 # unlink $tempfile; # may fail in global destruction
206 "_current_tempfile_fh",
207 "_delayed_operations",
214 "_remember_last_uptodate_call",
219 "__verified_tempdir",
221 "_uptodateness_ever_reached",
226 split /\n/, <<'=cut'; push @accessors, grep {s/^=item\s+//} @pod_lines; }
232 A list of interval specs that tell the aggregator which I<recentfile>s
237 The name of a method to canonize the path before rsyncing. Only
238 supported value is C<naive_path_normalize>. Defaults to that.
242 A comment about this tree and setup.
246 A timestamp. The dirtymark is updated whenever an out of band change
247 on the origin server is performed that violates the protocol. Say,
248 they add or remove files in the middle somewhere. Slaves must react
249 with a devaluation of their C<done> structure which then leads to a
250 full re-sync of all files. Implementation note: dirtymark may increase
255 The (prefix of the) filename we use for this I<recentfile>. Defaults to
256 C<RECENT>. The string must not contain a directory separator.
260 Timestamp remembering when we mirrored this recentfile the last time.
261 Only relevant for slaves.
263 =item ignore_link_stat_errors
265 If set to true, rsync errors are ignored that complain about link stat
266 errors. These seem to happen only when there are files missing at the
267 origin. In race conditions this can always happen, so it defaults to
272 If set to true, this object will fetch a new recentfile from remote
273 when the timespan between the last mirror (see have_mirrored) and now
274 is too large (see C<ttl>).
276 =item keep_delete_objects_forever
278 The default for delete events is that they are passed through the
279 collection of recentfile objects until they reach the Z file. There
280 they get dropped so that the associated file object ceases to exist at
281 all. By setting C<keep_delete_objects_forever> the delete objects are
282 kept forever. This makes the Z file larger but has the advantage that
283 slaves that have interrupted mirroring for a long time still can clean
288 After how many seconds shall we die if we cannot lock a I<recentfile>?
289 Defaults to 600 seconds.
293 When mirror_loop is called, this accessor can specify how much time
294 every loop shall at least take. If the work of a loop is done before
295 that time has gone, sleeps for the rest of the time. Defaults to
296 arbitrary 42 seconds.
298 =item max_files_per_connection
300 Maximum number of files that are transferred on a single rsync call.
301 Setting it higher means higher performance at the price of holding
302 connections longer and potentially disturbing other users in the pool.
303 Defaults to the arbitrary value 42.
305 =item max_rsync_errors
307 When rsync operations encounter that many errors without any resetting
308 success in between, then we die. Defaults to unlimited. A value of
309 -1 means we run forever ignoring all rsync errors.
313 Hashref remembering when we read the recent_events from this file the
314 last time and what the timespan was.
318 When the RECENT file format changes, we increment the protocol. We try
319 to support older protocols in later releases.
323 The host we are mirroring from. Leave empty for the local filesystem.
327 Rsync servers have so called modules to separate directory trees from
328 each other. Put here the name of the module under which we are
329 mirroring. Leave empty for local filesystem.
333 Things like compress, links, times or checksums. Passed in to the
334 File::Rsync object used to run the mirror.
336 =item serializer_suffix
338 Mostly untested accessor. The only well tested format for
339 I<recentfile>s at the moment is YAML. It is used with YAML::Syck via
340 Data::Serializer. But in principle other formats are supported as
341 well. See section SERIALIZERS below.
343 =item sleep_per_connection
345 Sleep that many seconds (floating point OK) after every chunk of rsyncing
346 has finished. Defaults to arbitrary 0.42.
350 Directory to write temporary files to. Must allow rename operations
351 into the tree which usually means it must live on the same partition
352 as the target directory. Defaults to C<< $self->localroot >>.
356 Time to live. Number of seconds after which this recentfile must be
357 fetched again from the origin server. Only relevant for slaves.
358 Defaults to arbitrary 24.2 seconds.
362 Boolean to turn on a bit verbosity.
366 Path to the logfile to write verbose progress information to. This is
367 a primitive stop gap solution to get simple verbose logging working.
368 Switching to Log4perl or similar is probably the way to go.
374 use accessors
@accessors;
378 =head2 (void) $obj->aggregate( %options )
380 Takes all intervals that are collected in the accessor called
381 aggregator. Sorts them by actual length of the interval.
382 Removes those that are shorter than our own interval. Then merges this
383 object into the next larger object. The merging continues upwards
384 as long as the next I<recentfile> is old enough to warrant a merge.
386 If a merge is warranted is decided according to the interval of the
387 previous interval so that larger files are not so often updated as
388 smaller ones. If $options{force} is true, all files get updated.
390 Here is an example to illustrate the behaviour. Given aggregators
396 1h updates 1d on every call to aggregate()
397 1d updates 1W earliest after 1h
398 1W updates 1M earliest after 1d
399 1M updates 1Q earliest after 1W
400 1Q updates 1Y earliest after 1M
401 1Y updates Z earliest after 1Q
403 Note that all but the smallest recentfile get updated at an arbitrary
404 rate and as such are quite useless on their own.
409 my($self, %option) = @_;
410 my @aggs = sort { $a->{secs
} <=> $b->{secs
} }
411 grep { $_->{secs
} >= $self->interval_secs }
412 map { { interval
=> $_, secs
=> $self->interval_secs($_)} }
413 $self->interval, @
{$self->aggregator || []};
415 $aggs[0]{object
} = $self;
416 AGGREGATOR
: for my $i (0..$#aggs-1) {
417 my $this = $aggs[$i]{object
};
418 my $next = $this->_sparse_clone;
419 $next->interval($aggs[$i+1]{interval
});
421 if ($option{force
} || $i == 0) {
424 my $next_rfile = $next->rfile;
425 if (-e
$next_rfile) {
426 my $prev = $aggs[$i-1]{object
};
428 my $next_age = 86400 * -M
$next_rfile;
429 if ($next_age > $prev->interval_secs) {
438 $aggs[$i+1]{object
} = $next;
445 # collect file size and mtime for all files of this aggregate
446 sub _debug_aggregate
{
448 my @aggs = sort { $a->{secs
} <=> $b->{secs
} }
449 map { { interval
=> $_, secs
=> $self->interval_secs($_)} }
450 $self->interval, @
{$self->aggregator || []};
452 for my $i (0..$#aggs) {
453 my $this = Storable
::dclone
$self;
454 $this->interval($aggs[$i]{interval
});
455 my $rfile = $this->rfile;
456 my @stat = stat $rfile;
457 push @
$report, {rfile
=> $rfile, size
=> $stat[7], mtime
=> $stat[9]};
462 # (void) $self->_assert_symlink()
463 sub _assert_symlink
{
465 my $recentrecentfile = File
::Spec
->catfile
474 if ($Config{d_symlink
} eq "define") {
475 my $howto_create_symlink; # 0=no need; 1=straight symlink; 2=rename symlink
476 if (-l
$recentrecentfile) {
477 my $found_symlink = readlink $recentrecentfile;
478 if ($found_symlink eq $self->rfilename) {
481 $howto_create_symlink = 2;
484 $howto_create_symlink = 1;
486 if (1 == $howto_create_symlink) {
487 symlink $self->rfilename, $recentrecentfile or die "Could not create symlink '$recentrecentfile': $!"
489 unlink "$recentrecentfile.$$"; # may fail
490 symlink $self->rfilename, "$recentrecentfile.$$" or die "Could not create symlink '$recentrecentfile.$$': $!";
491 rename "$recentrecentfile.$$", $recentrecentfile or die "Could not rename '$recentrecentfile.$$' to $recentrecentfile: $!";
494 warn "Warning: symlinks not supported on this system, doing a copy instead\n";
495 unlink "$recentrecentfile.$$"; # may fail
496 cp
$self->rfilename, "$recentrecentfile.$$" or die "Could not copy to '$recentrecentfile.$$': $!";
497 rename "$recentrecentfile.$$", $recentrecentfile or die "Could not rename '$recentrecentfile.$$' to $recentrecentfile: $!";
501 =head2 $hashref = $obj->delayed_operations
503 A hash of hashes containing unlink and rmdir operations which had to
504 wait until the recentfile got unhidden in order to not confuse
505 downstream mirrors (in case we have some).
509 sub delayed_operations
{
511 my $x = $self->_delayed_operations;
512 unless (defined $x) {
517 $self->_delayed_operations ($x);
522 =head2 $done = $obj->done
524 C<$done> is a reference to a L<File::Rsync::Mirror::Recentfile::Done>
525 object that keeps track of rsync activities. Only needed and used when
526 we are a mirroring slave.
532 my $done = $self->_done;
534 require File
::Rsync
::Mirror
::Recentfile
::Done
;
535 $done = File
::Rsync
::Mirror
::Recentfile
::Done
->new();
536 $done->_rfinterval ($self->interval);
537 $self->_done ( $done );
542 =head2 $tempfilename = $obj->get_remote_recentfile_as_tempfile ()
544 Stores the remote I<recentfile> locally as a tempfile. The caller is
545 responsible to remove the file after use.
547 Note: if you're intending to act as an rsync server for other slaves,
548 then you must prefer this method to fetch that file with
549 get_remotefile(). Otherwise downstream mirrors would expect you to
550 already have mirrored all the files that are in the I<recentfile>
551 before you have them mirrored.
555 sub get_remote_recentfile_as_tempfile
{
557 mkpath
$self->localroot;
560 if ( $self->_use_tempfile() ) {
561 if ($self->ttl_reached) {
562 $fh = $self->_current_tempfile_fh;
563 $trfilename = $self->rfilename;
565 return $self->_current_tempfile;
568 $trfilename = $self->rfilename;
573 $dst = $self->_current_tempfile;
575 $fh = $self->_get_remote_rat_provide_tempfile_object ($trfilename);
576 $dst = $fh->filename;
577 $self->_current_tempfile ($dst);
578 my $rfile = eval { $self->rfile; }; # may fail (RECENT.recent has no rfile)
579 if (defined $rfile && -e
$rfile) {
580 # saving on bandwidth. Might need to be configurable
581 # $self->bandwidth_is_cheap?
582 cp
$rfile, $dst or die "Could not copy '$rfile' to '$dst': $!"
589 if ($self->verbose) {
590 my $doing = -e
$dst ?
"Sync" : "Get";
591 my $display_dst = join "/", "...", basename
(dirname
($dst)), basename
($dst);
592 my $LFH = $self->_logfilehandle;
595 "%-4s %d (1/1/%s) temp %s ... ",
604 local($ENV{LANG
}) = "C";
605 while (!$self->rsync->exec(
609 $self->register_rsync_error ($self->rsync->err);
610 if (++$retried >= 3) {
611 warn "XXX giving up";
617 my $LFH = $self->_logfilehandle;
618 printf $LFH "Warning: gave up mirroring %s, will try again later", $self->interval;
620 $self->_refresh_internals ($dst);
621 $self->have_mirrored (Time
::HiRes
::time);
622 $self->un_register_rsync_error ();
625 if ($self->verbose) {
626 my $LFH = $self->_logfilehandle;
630 chmod $mode, $dst or die "Could not chmod $mode '$dst': $!";
634 sub _verified_tempdir
{
636 my $tempdir = $self->__verified_tempdir();
637 return $tempdir if defined $tempdir;
638 unless ($tempdir = $self->tempdir) {
639 $tempdir = $self->localroot;
641 unless (-d
$tempdir) {
644 $self->__verified_tempdir($tempdir);
648 sub _get_remote_rat_provide_tempfile_object
{
649 my($self, $trfilename) = @_;
650 my $_verified_tempdir = $self->_verified_tempdir;
651 my $fh = File
::Temp
->new
652 (TEMPLATE
=> sprintf(".FRMRecent-%s-XXXX",
655 DIR
=> $_verified_tempdir,
656 SUFFIX
=> $self->serializer_suffix,
657 UNLINK
=> $self->_use_tempfile,
660 my $dst = $fh->filename;
661 chmod $mode, $dst or die "Could not chmod $mode '$dst': $!";
662 if ($self->_use_tempfile) {
663 $self->_current_tempfile_fh ($fh); # delay self destruction
671 if (my $vl = $self->verboselog) {
672 open $fh, ">>", $vl or die "Could not open >> '$vl': $!";
679 =head2 $localpath = $obj->get_remotefile ( $relative_path )
681 Rsyncs one single remote file to local filesystem.
683 Note: no locking is done on this file. Any number of processes may
686 Note II: do not use for recentfiles. If you are a cascading
687 slave/server combination, it would confuse other slaves. They would
688 expect the contents of these recentfiles to be available. Use
689 get_remote_recentfile_as_tempfile() instead.
694 my($self, $path) = @_;
695 my $dst = File
::Spec
->catfile($self->localroot, $path);
697 if ($self->verbose) {
698 my $doing = -e
$dst ?
"Sync" : "Get";
699 my $LFH = $self->_logfilehandle;
702 "%-4s %d (1/1/%s) %s ... ",
709 local($ENV{LANG
}) = "C";
710 my $remoteroot = $self->remoteroot or die "Alert: missing remoteroot. Cannot continue";
711 while (!$self->rsync->exec(
717 $self->register_rsync_error ($self->rsync->err);
719 $self->un_register_rsync_error ();
720 if ($self->verbose) {
721 my $LFH = $self->_logfilehandle;
727 =head2 $obj->interval ( $interval_spec )
729 Get/set accessor. $interval_spec is a string and described below in
730 the section INTERVAL SPEC.
735 my ($self, $interval) = @_;
737 $self->_interval($interval);
738 $self->_rfile(undef);
740 $interval = $self->_interval;
741 unless (defined $interval) {
742 # do not ask the $self too much, it recurses!
744 Carp
::confess
("Alert: interval undefined for '".$self."'. Cannot continue.");
749 =head2 $secs = $obj->interval_secs ( $interval_spec )
751 $interval_spec is described below in the section INTERVAL SPEC. If
752 empty defaults to the inherent interval for this object.
757 my ($self, $interval) = @_;
758 $interval ||= $self->interval;
759 unless (defined $interval) {
760 die "interval_secs() called without argument on an object without a declared one";
762 my ($n,$t) = $interval =~ /^(\d*)([smhdWMQYZ]$)/ or
763 die "Could not determine seconds from interval[$interval]";
764 if ($interval eq "Z") {
766 } elsif (exists $seconds{$t} and $n =~ /^\d+$/) {
767 return $seconds{$t}*$n;
769 die "Invalid interval specification: n[$n]t[$t]";
773 =head2 $obj->localroot ( $localroot )
775 Get/set accessor. The local root of the tree.
780 my ($self, $localroot) = @_;
782 $self->_localroot($localroot);
783 $self->_rfile(undef);
785 $localroot = $self->_localroot;
788 =head2 $ret = $obj->local_path($path_found_in_recentfile)
790 Combines the path to our local mirror and the path of an object found
791 in this I<recentfile>. In other words: the target of a mirror operation.
793 Implementation note: We split on slashes and then use
794 File::Spec::catfile to adjust to the local operating system.
799 my($self,$path) = @_;
800 unless (defined $path) {
801 # seems like a degenerated case
802 return $self->localroot;
804 my @p = split m
|/|, $path;
805 File
::Spec
->catfile($self->localroot,@p);
808 =head2 (void) $obj->lock
810 Locking is implemented with an C<mkdir> on a locking directory
811 (C<.lock> appended to $rfile).
817 # not using flock because it locks on filehandles instead of
818 # old school ressources.
819 my $locked = $self->_is_locked and return;
820 my $rfile = $self->rfile;
821 # XXX need a way to allow breaking the lock
823 my $locktimeout = $self->locktimeout || 600;
825 GETLOCK
: while (not mkdir "$rfile.lock") {
826 if (open my $fh, "<", "$rfile.lock/process") {
827 chomp(my $process = <$fh>);
829 } elsif ($$ == $process) {
831 } elsif (kill 0, $process) {
832 warn "Warning: process $process holds a lock, waiting..." unless $have_warned{$process}++;
834 warn "Warning: breaking lock held by process $process";
839 Time
::HiRes
::sleep 0.01;
840 if (time - $start > $locktimeout) {
841 die "Could not acquire lockdirectory '$rfile.lock': $!";
844 open my $fh, ">", "$rfile.lock/process" or die "Could not open >$rfile.lock/process\: $!";
846 close $fh or die "Could not close: $!";
847 $self->_is_locked (1);
850 =head2 (void) $obj->merge ($other)
852 Bulk update of this object with another one. It's used to merge a
853 smaller and younger $other object into the current one. If this file
854 is a C<Z> file, then we normally do not merge in objects of type
855 C<delete>; this can be overridden by setting
856 keep_delete_objects_forever. But if we encounter an object of type
857 delete we delete the corresponding C<new> object if we have it.
859 If there is nothing to be merged, nothing is done.
864 my($self, $other) = @_;
865 $self->_merge_sanitycheck ( $other );
867 my $other_recent = $other->recent_events || [];
868 # $DB::single++ if $other->interval_secs eq "2" and grep {$_->{epoch} eq "999.999"} @$other_recent;
870 $self->_merge_locked ( $other, $other_recent );
876 my($self, $other, $other_recent) = @_;
877 my $my_recent = $self->recent_events || [];
879 # calculate the target time span
880 my $myepoch = $my_recent->[0] ?
$my_recent->[0]{epoch
} : undef;
881 my $epoch = $other_recent->[0] ?
$other_recent->[0]{epoch
} : $myepoch;
882 my $oldest_allowed = 0;
884 unless ($my_recent->[0]) {
889 if (($other->dirtymark||0) ne ($self->dirtymark||0)) {
892 } elsif (my $merged = $self->merged) {
893 my $secs = $self->interval_secs();
894 $oldest_allowed = min
($epoch - $secs, $merged->{epoch
}||0);
895 if (@
$other_recent and
896 _bigfloatlt
($other_recent->[-1]{epoch
}, $oldest_allowed)
898 $oldest_allowed = $other_recent->[-1]{epoch
};
901 while (@
$my_recent && _bigfloatlt
($my_recent->[-1]{epoch
}, $oldest_allowed)) {
908 my $other_recent_filtered = [];
909 for my $oev (@
$other_recent) {
910 my $oevepoch = $oev->{epoch
} || 0;
911 next if _bigfloatlt
($oevepoch, $oldest_allowed);
912 my $path = $oev->{path
};
913 next if $have_path{$path}++;
914 if ( $self->interval eq "Z"
915 and $oev->{type
} eq "delete"
916 and ! $self->keep_delete_objects_forever
920 if (!$myepoch || _bigfloatgt
($oevepoch, $myepoch)) {
923 push @
$other_recent_filtered, { epoch
=> $oev->{epoch
}, path
=> $path, type
=> $oev->{type
} };
926 if ($something_done) {
927 $self->_merge_something_done ($other_recent_filtered, $my_recent, $other_recent, $other, \
%have_path, $epoch);
931 sub _merge_something_done
{
932 my($self, $other_recent_filtered, $my_recent, $other_recent, $other, $have_path, $epoch) = @_;
934 my $epoch_conflict = 0;
936 ZIP
: while (@
$other_recent_filtered || @
$my_recent) {
939 @
$other_recent_filtered && _bigfloatge
($other_recent_filtered->[0]{epoch
},$my_recent->[0]{epoch
})) {
940 $event = shift @
$other_recent_filtered;
942 $event = shift @
$my_recent;
943 next ZIP
if $have_path->{$event->{path
}}++;
945 $epoch_conflict=1 if defined $last_epoch && $event->{epoch
} eq $last_epoch;
946 $last_epoch = $event->{epoch
};
947 push @
$recent, $event;
949 if ($epoch_conflict) {
951 for (my $i = $#$recent;$i>=0;$i--) {
952 my $epoch = $recent->[$i]{epoch
};
953 if ($have_epoch{$epoch}++) {
954 while ($have_epoch{$epoch}) {
955 $epoch = _increase_a_bit
($epoch);
957 $recent->[$i]{epoch
} = $epoch;
958 $have_epoch{$epoch}++;
962 if (!$self->dirtymark || $other->dirtymark ne $self->dirtymark) {
963 $self->dirtymark ( $other->dirtymark );
965 $self->write_recent($recent);
967 time => Time
::HiRes
::time, # not used anywhere
968 epoch
=> $recent->[0]{epoch
},
969 into_interval
=> $self->interval, # not used anywhere
971 $other->write_recent($other_recent);
974 sub _merge_sanitycheck
{
975 my($self, $other) = @_;
976 if ($self->interval_secs <= $other->interval_secs) {
979 "Alert: illegal merge operation of a bigger interval[%d] into a smaller[%d]",
980 $self->interval_secs,
981 $other->interval_secs,
988 Hashref denoting when this recentfile has been merged into some other
994 my($self, $set) = @_;
996 $self->_merged ($set);
998 my $merged = $self->_merged;
1000 if ($merged and $into = $merged->{into_interval
} and defined $self->_interval) {
1002 if ($into eq $self->interval) {
1006 "Warning: into_interval[%s] same as own interval[%s]. Danger ahead.",
1010 } elsif ($self->interval_secs($into) < $self->interval_secs) {
1014 "Warning: into_interval_secs[%s] smaller than own interval_secs[%s] on interval[%s]. Danger ahead.",
1015 $self->interval_secs($into),
1016 $self->interval_secs,
1024 =head2 $hashref = $obj->meta_data
1026 Returns the hashref of metadata that the server has to add to the
1033 my $ret = $self->{meta
};
1044 "serializer_suffix",
1051 # XXX need to reset the Producer if I am a writer, keep it when I
1053 $ret->{Producers
} ||= {
1054 __PACKAGE__
, "$VERSION", # stringified it looks better
1056 'time', Time
::HiRes
::time,
1058 $ret->{dirtymark
} ||= Time
::HiRes
::time;
1062 =head2 $success = $obj->mirror ( %options )
1064 Mirrors the files in this I<recentfile> as reported by
1065 C<recent_events>. Options named C<after>, C<before>, C<max> are passed
1066 through to the C<recent_events> call. The boolean option C<piecemeal>,
1067 if true, causes C<mirror> to only rsync C<max_files_per_connection>
1068 and keep track of the rsynced files so that future calls will rsync
1069 different files until all files are brought to sync.
1074 my($self, %options) = @_;
1075 my $trecentfile = $self->get_remote_recentfile_as_tempfile();
1076 $self->_use_tempfile (1);
1077 # skip-deletes is inadequat for passthrough within mirror. We
1078 # would never reach uptodateness when a delete were on a
1080 my %passthrough = map { ($_ => $options{$_}) } qw(before after max);
1081 my ($recent_events) = $self->recent_events(%passthrough);
1082 my(@error, @dlcollector); # download-collector: array containing paths we need
1084 my $last_item = $#$recent_events;
1085 my $done = $self->done;
1086 my $pathdb = $self->_pathdb;
1087 ITEM
: for my $i ($first_item..$last_item) {
1101 last if $i == $last_item;
1102 if ($status->{mustreturn
}){
1103 if ($self->_current_tempfile && ! $self->_current_tempfile_fh) {
1104 # looks like a bug somewhere else
1105 my $t = $self->_current_tempfile;
1106 unlink $t or die "Could not unlink '$t': $!";
1107 $self->_current_tempfile(undef);
1108 $self->_use_tempfile(0);
1114 my $success = eval { $self->_mirror_dlcollector (\
@dlcollector,$pathdb,$recent_events);};
1115 if (!$success || $@
) {
1116 warn "Warning: Unknown error while mirroring: $@";
1121 if ($self->verbose) {
1122 my $LFH = $self->_logfilehandle;
1123 print $LFH "DONE\n";
1125 # once we've gone to the end we consider ourselves free of obligations
1127 $self->_mirror_unhide_tempfile ($trecentfile);
1128 $self->_mirror_perform_delayed_ops(\
%options);
1144 my $recent_event = $recent_events->[$i];
1145 return if $done->covered ( $recent_event->{epoch
} );
1147 my $rec = $pathdb->{$recent_event->{path
}};
1148 if ($rec && $rec->{recentepoch
}) {
1150 ( $rec->{recentepoch
}, $recent_event->{epoch
} )){
1151 $done->register ($recent_events, [$i]);
1156 my $dst = $self->local_path($recent_event->{path
});
1157 if ($recent_event->{type
} eq "new"){
1158 $self->_mirror_item_new
1171 } elsif ($recent_event->{type
} eq "delete") {
1173 if ($options->{'skip-deletes'}) {
1174 $activity = "skipped";
1177 $activity = "not_found";
1178 } elsif (-l
$dst or not -d _
) {
1179 $self->delayed_operations->{unlink}{$dst}++;
1180 $activity = "deleted";
1182 $self->delayed_operations->{rmdir}{$dst}++;
1183 $activity = "deleted";
1186 $done->register ($recent_events, [$i]);
1188 $self->_mirror_register_path($pathdb,[$recent_event],$activity);
1191 warn "Warning: invalid upload type '$recent_event->{type}'";
1195 sub _mirror_item_new
{
1208 if ($self->verbose) {
1209 my $doing = -e
$dst ?
"Sync" : "Get";
1210 my $LFH = $self->_logfilehandle;
1213 "%-4s %d (%d/%d/%s) %s ... ",
1219 $recent_event->{path
},
1222 my $max_files_per_connection = $self->max_files_per_connection || 42;
1224 if ($self->verbose) {
1225 my $LFH = $self->_logfilehandle;
1228 push @
$dlcollector, { rev
=> $recent_event, i
=> $i };
1229 if (@
$dlcollector >= $max_files_per_connection) {
1230 $success = eval {$self->_mirror_dlcollector ($dlcollector,$pathdb,$recent_events);};
1231 my $sleep = $self->sleep_per_connection;
1232 $sleep = 0.42 unless defined $sleep;
1233 Time
::HiRes
::sleep $sleep;
1234 if ($options->{piecemeal
}) {
1235 $status->{mustreturn
} = 1;
1241 if (!$success || $@
) {
1242 warn "Warning: Error while mirroring: $@";
1246 if ($self->verbose) {
1247 my $LFH = $self->_logfilehandle;
1248 print $LFH "DONE\n";
1252 sub _mirror_dlcollector
{
1253 my($self,$xcoll,$pathdb,$recent_events) = @_;
1254 my $success = $self->mirror_path([map {$_->{rev
}{path
}} @
$xcoll]);
1256 $self->_mirror_register_path($pathdb,[map {$_->{rev
}} @
$xcoll],"rsync");
1258 $self->done->register($recent_events, [map {$_->{i
}} @
$xcoll]);
1263 sub _mirror_register_path
{
1264 my($self,$pathdb,$coll,$activity) = @_;
1266 for my $item (@
$coll) {
1267 $pathdb->{$item->{path
}} =
1269 recentepoch
=> $item->{epoch
},
1270 ($activity."_on") => $time,
1275 sub _mirror_unhide_tempfile
{
1276 my($self, $trecentfile) = @_;
1277 my $rfile = $self->rfile;
1278 if (rename $trecentfile, $rfile) {
1279 # warn "DEBUG: renamed '$trecentfile' to '$rfile'";
1282 Carp
::confess
("Could not rename '$trecentfile' to '$rfile': $!");
1284 $self->_use_tempfile (0);
1285 if (my $ctfh = $self->_current_tempfile_fh) {
1286 $ctfh->unlink_on_destroy (0);
1287 $self->_current_tempfile_fh (undef);
1291 sub _mirror_perform_delayed_ops
{
1292 my($self,$options) = @_;
1293 my $delayed = $self->delayed_operations;
1294 for my $dst (keys %{$delayed->{unlink}}) {
1295 unless (unlink $dst) {
1297 Carp
::cluck
( "Warning: Error while unlinking '$dst': $!" ) if $options->{verbose
};
1299 if ($self->verbose) {
1301 my $LFH = $self->_logfilehandle;
1304 "%-4s %d (%s) %s DONE\n",
1310 delete $delayed->{unlink}{$dst};
1313 for my $dst (sort {length($b) <=> length($a)} keys %{$delayed->{rmdir}}) {
1314 unless (rmdir $dst) {
1316 Carp
::cluck
( "Warning: Error on rmdir '$dst': $!" ) if $options->{verbose
};
1318 if ($self->verbose) {
1320 my $LFH = $self->_logfilehandle;
1323 "%-4s %d (%s) %s DONE\n",
1329 delete $delayed->{rmdir}{$dst};
1334 =head2 $success = $obj->mirror_path ( $arrref | $path )
1336 If the argument is a scalar it is treated as a path. The remote path
1337 is mirrored into the local copy. $path is the path found in the
1338 I<recentfile>, i.e. it is relative to the root directory of the
1341 If the argument is an array reference then all elements are treated as
1342 a path below the current tree and all are rsynced with a single
1343 command (and a single connection).
1348 my($self,$path) = @_;
1349 # XXX simplify the two branches such that $path is treated as
1350 # [$path] maybe even demand the argument as an arrayref to
1351 # simplify docs and code. (rsync-over-recentfile-2.pl uses the
1353 if (ref $path and ref $path eq "ARRAY") {
1354 my $dst = $self->localroot;
1355 mkpath dirname
$dst;
1356 my($fh) = File
::Temp
->new(TEMPLATE
=> sprintf(".%s-XXXX",
1357 lc $self->filenameroot,
1362 for my $p (@
$path) {
1366 $fh->unlink_on_destroy(1);
1369 local($ENV{LANG
}) = "C";
1370 while (!$self->rsync->exec
1376 'files-from' => $fh->filename,
1378 my(@err) = $self->rsync->err;
1379 if ($self->_my_ignore_link_stat_errors && "@err" =~ m{^ rsync: \s link_stat }x ) {
1380 if ($self->verbose) {
1381 my $LFH = $self->_logfilehandle;
1382 print $LFH "Info: ignoring link_stat error '@err'";
1386 $self->register_rsync_error (@err);
1387 if (++$retried >= 3) {
1388 my $batchsize = @
$path;
1389 warn "The number of rsync retries now reached 3 within a batch of size $batchsize. Error was '@err'. Giving up now, will retry later, ";
1396 $self->un_register_rsync_error ();
1399 my $dst = $self->local_path($path);
1400 mkpath dirname
$dst;
1401 local($ENV{LANG
}) = "C";
1402 while (!$self->rsync->exec
1410 my(@err) = $self->rsync->err;
1411 if ($self->_my_ignore_link_stat_errors && "@err" =~ m{^ rsync: \s link_stat }x ) {
1412 if ($self->verbose) {
1413 my $LFH = $self->_logfilehandle;
1414 print $LFH "Info: ignoring link_stat error '@err'";
1418 $self->register_rsync_error (@err);
1420 $self->un_register_rsync_error ();
1425 sub _my_ignore_link_stat_errors
{
1427 my $x = $self->ignore_link_stat_errors;
1428 $x = 1 unless defined $x;
1432 sub _my_current_rfile
{
1435 if ($self->_use_tempfile) {
1436 $rfile = $self->_current_tempfile;
1438 $rfile = $self->rfile;
1443 =head2 $path = $obj->naive_path_normalize ($path)
1445 Takes an absolute unix style path as argument and canonicalizes it to
1446 a shorter path if possible, removing things like double slashes or
1447 C</./> and removes references to C<../> directories to get a shorter
1448 unambiguos path. This is used to make the code easier that determines
1449 if a file passed to C<upgrade()> is indeed below our C<localroot>.
1453 sub naive_path_normalize
{
1454 my($self,$path) = @_;
1456 1 while $path =~ s
|/[^/]+/\.\./|/|;
1461 =head2 $ret = $obj->read_recent_1 ( $data )
1463 Delegate of C<recent_events()> on protocol 1
1468 my($self, $data) = @_;
1469 return $data->{recent
};
1472 =head2 $array_ref = $obj->recent_events ( %options )
1474 Note: the code relies on the resource being written atomically. We
1475 cannot lock because we may have no write access. If the caller has
1476 write access (eg. aggregate() or update()), it has to care for any
1477 necessary locking and it MUST write atomically.
1479 If C<$options{after}> is specified, only file events after this
1480 timestamp are returned.
1482 If C<$options{before}> is specified, only file events before this
1483 timestamp are returned.
1485 If C<$options{max}> is specified only a maximum of this many events is
1488 If C<$options{'skip-deletes'}> is specified, no files-to-be-deleted
1491 If C<$options{contains}> is specified the value must be a hash
1492 reference containing a query. The query may contain the keys C<epoch>,
1493 C<path>, and C<type>. Each represents a condition that must be met. If
1494 there is more than one such key, the conditions are ANDed.
1496 If C<$options{info}> is specified, it must be a hashref. This hashref
1497 will be filled with metadata about the unfiltered recent_events of
1498 this object, in key C<first> there is the first item, in key C<last>
1504 my ($self, %options) = @_;
1505 my $info = $options{info
};
1506 if ($self->is_slave) {
1507 # XXX seems dubious, might produce tempfiles without removing them?
1508 $self->get_remote_recentfile_as_tempfile;
1510 my $rfile_or_tempfile = $self->_my_current_rfile or return [];
1511 -e
$rfile_or_tempfile or return [];
1512 my $suffix = $self->serializer_suffix;
1514 $self->_try_deserialize
1521 if ($err or !$data) {
1525 if (reftype
$data eq 'ARRAY') { # protocol 0
1528 $re = $self->_recent_events_protocol_x
1534 return $re unless grep {defined $options{$_}} qw(after before contains max skip-deletes);
1535 $self->_recent_events_handle_options ($re, \
%options);
1538 # File::Rsync::Mirror::Recentfile::_recent_events_handle_options
1539 sub _recent_events_handle_options
{
1540 my($self, $re, $options) = @_;
1541 my $last_item = $#$re;
1542 my $info = $options->{info
};
1544 $info->{first
} = $re->[0];
1545 $info->{last} = $re->[-1];
1547 if (defined $options->{after
}) {
1548 if ($re->[0]{epoch
} > $options->{after
}) {
1551 {$re->[$_]{epoch
} <= $options->{after
}}
1561 if (defined $options->{before
}) {
1562 if ($re->[0]{epoch
} > $options->{before
}) {
1565 {$re->[$_]{epoch
} < $options->{before
}}
1574 if (0 != $first_item || -1 != $last_item) {
1575 @
$re = splice @
$re, $first_item, 1+$last_item-$first_item;
1577 if ($options->{'skip-deletes'}) {
1578 @
$re = grep { $_->{type
} ne "delete" } @
$re;
1580 if (my $contopt = $options->{contains
}) {
1581 my $seen_allowed = 0;
1582 for my $allow (qw(epoch path type)) {
1583 if (exists $contopt->{$allow}) {
1585 my $v = $contopt->{$allow};
1586 @
$re = grep { $_->{$allow} eq $v } @
$re;
1589 if (keys %$contopt > $seen_allowed) {
1592 (sprintf "unknown query: %s", join ", ", %$contopt);
1595 if ($options->{max
} && @
$re > $options->{max
}) {
1596 @
$re = splice @
$re, 0, $options->{max
};
1601 sub _recent_events_protocol_x
{
1606 my $meth = sprintf "read_recent_%d", $data->{meta
}{protocol
};
1607 # we may be reading meta for the first time
1608 while (my($k,$v) = each %{$data->{meta
}}) {
1609 if ($k ne lc $k){ # "Producers"
1610 $self->{ORIG
}{$k} = $v;
1613 next if defined $self->$k;
1616 my $re = $self->$meth ($data);
1618 if (my @stat = stat $rfile_or_tempfile) {
1619 $minmax = { mtime
=> $stat[9] };
1621 # defensive because ABH encountered:
1623 #### Sync 1239828608 (1/1/Z) temp .../authors/.FRMRecent-RECENT-Z.yaml-
1624 #### Ydr_.yaml ... DONE
1625 #### Cannot stat '/mirrors/CPAN/authors/.FRMRecent-RECENT-Z.yaml-
1626 #### Ydr_.yaml': No such file or directory at /usr/lib/perl5/site_perl/
1627 #### 5.8.8/File/Rsync/Mirror/Recentfile.pm line 1558.
1628 #### unlink0: /mirrors/CPAN/authors/.FRMRecent-RECENT-Z.yaml-Ydr_.yaml is
1629 #### gone already at cpan-pause.pl line 0
1631 my $LFH = $self->_logfilehandle;
1632 print $LFH "Warning (maybe harmless): Cannot stat '$rfile_or_tempfile': $!"
1635 $minmax->{min
} = $re->[-1]{epoch
};
1636 $minmax->{max
} = $re->[0]{epoch
};
1638 $self->minmax ( $minmax );
1642 sub _try_deserialize
{
1647 if ($suffix eq ".yaml") {
1649 YAML
::Syck
::LoadFile
($rfile_or_tempfile);
1650 } elsif ($HAVE->{"Data::Serializer"}) {
1651 my $serializer = Data
::Serializer
->new
1652 ( serializer
=> $serializers{$suffix} );
1655 open my $fh, $rfile_or_tempfile or die "Could not open: $!";
1659 $serializer->raw_deserialize($serialized);
1661 die "Data::Serializer not installed, cannot proceed with suffix '$suffix'";
1665 sub _refresh_internals
{
1666 my($self, $dst) = @_;
1667 my $class = ref $self;
1668 my $rfpeek = $class->new_from_file ($dst);
1673 $self->$acc ( $rfpeek->$acc );
1675 my $old_dirtymark = $self->dirtymark;
1676 my $new_dirtymark = $rfpeek->dirtymark;
1677 if ($old_dirtymark && $new_dirtymark && $new_dirtymark ne $old_dirtymark) {
1679 $self->dirtymark ( $new_dirtymark );
1680 $self->_uptodateness_ever_reached(0);
1685 =head2 $ret = $obj->rfilename
1687 Just the basename of our I<recentfile>, composed from C<filenameroot>,
1688 a dash, C<interval>, and C<serializer_suffix>. E.g. C<RECENT-6h.yaml>
1694 my $file = sprintf("%s-%s%s",
1695 $self->filenameroot,
1697 $self->serializer_suffix,
1702 =head2 $str = $self->remote_dir
1704 The directory we are mirroring from.
1709 my($self, $set) = @_;
1711 $self->_remote_dir ($set);
1713 my $x = $self->_remote_dir;
1714 $self->is_slave (1);
1718 =head2 $str = $obj->remoteroot
1720 =head2 (void) $obj->remoteroot ( $set )
1722 Get/Set the composed prefix needed when rsyncing from a remote module.
1723 If remote_host, remote_module, and remote_dir are set, it is composed
1729 my($self, $set) = @_;
1731 $self->_remoteroot($set);
1733 my $remoteroot = $self->_remoteroot;
1734 unless (defined $remoteroot) {
1735 $remoteroot = sprintf
1738 defined $self->remote_host ?
($self->remote_host."::") : "",
1739 defined $self->remote_module ?
($self->remote_module."/") : "",
1740 defined $self->remote_dir ?
$self->remote_dir : "",
1742 $self->_remoteroot($remoteroot);
1747 =head2 (void) $obj->split_rfilename ( $recentfilename )
1749 Inverse method to C<rfilename>. C<$recentfilename> is a plain filename
1752 $filenameroot-$interval$serializer_suffix
1758 This filename is split into its parts and the parts are fed to the
1763 sub split_rfilename
{
1764 my($self, $rfname) = @_;
1765 my($splitter) = qr
(^(.+)-([^-\
.]+)(\
.[^\
.]+));
1766 if (my($f,$i,$s) = $rfname =~ $splitter) {
1767 $self->filenameroot ($f);
1768 $self->interval ($i);
1769 $self->serializer_suffix ($s);
1771 die "Alert: cannot split '$rfname', doesn't match '$splitter'";
1776 =head2 my $rfile = $obj->rfile
1778 Returns the full path of the I<recentfile>
1784 my $rfile = $self->_rfile;
1785 return $rfile if defined $rfile;
1786 $rfile = File
::Spec
->catfile
1790 $self->_rfile ($rfile);
1794 =head2 $rsync_obj = $obj->rsync
1796 The File::Rsync object that this object uses for communicating with an
1803 my $rsync = $self->_rsync;
1804 unless (defined $rsync) {
1805 my $rsync_options = $self->rsync_options || {};
1806 if ($HAVE->{"File::Rsync"}) {
1807 $rsync = File
::Rsync
->new($rsync_options);
1808 $self->_rsync($rsync);
1810 die "File::Rsync required for rsync operations. Cannot continue";
1816 =head2 (void) $obj->register_rsync_error(@err)
1818 =head2 (void) $obj->un_register_rsync_error()
1820 Register_rsync_error is called whenever the File::Rsync object fails
1821 on an exec (say, connection doesn't succeed). It issues a warning and
1822 sleeps for an increasing amount of time. Un_register_rsync_error
1823 resets the error count. See also accessor C<max_rsync_errors>.
1828 my $no_success_count = 0;
1829 my $no_success_time = 0;
1830 sub register_rsync_error
{
1831 my($self, @err) = @_;
1833 $no_success_time = time;
1834 $no_success_count++;
1835 my $max_rsync_errors = $self->max_rsync_errors;
1836 $max_rsync_errors = MAX_INT
unless defined $max_rsync_errors;
1837 if ($max_rsync_errors>=0 && $no_success_count >= $max_rsync_errors) {
1843 "Alert: Error while rsyncing (%s): '%s', error count: %d, exiting now,",
1849 my $sleep = 12 * $no_success_count;
1850 $sleep = 300 if $sleep > 300;
1855 "Warning: %s, Error while rsyncing (%s): '%s', sleeping %d",
1856 scalar(localtime($no_success_time)),
1863 sub un_register_rsync_error
{
1865 $no_success_time = 0;
1866 $no_success_count = 0;
1870 =head2 $clone = $obj->_sparse_clone
1872 Clones just as much from itself that it does not hurt. Experimental
1875 Note: what fits better: sparse or shallow? Other suggestions?
1881 my $new = bless {}, ref $self;
1890 ignore_link_stat_errors
1892 max_files_per_connection
1896 sleep_per_connection
1901 $o = Storable
::dclone
$o if ref $o;
1907 =head2 $boolean = OBJ->ttl_reached ()
1913 my $have_mirrored = $self->have_mirrored || 0;
1914 my $now = Time
::HiRes
::time;
1915 my $ttl = $self->ttl;
1916 $ttl = 24.2 unless defined $ttl;
1917 if ($now > $have_mirrored + $ttl) {
1923 =head2 (void) $obj->unlock()
1925 Unlocking is implemented with an C<rmdir> on a locking directory
1926 (C<.lock> appended to $rfile).
1932 return unless $self->_is_locked;
1933 my $rfile = $self->rfile;
1934 unlink "$rfile.lock/process" or warn "Could not unlink lockfile '$rfile.lock/process': $!";
1935 rmdir "$rfile.lock" or warn "Could not rmdir lockdir '$rfile.lock': $!";;
1936 $self->_is_locked (0);
1941 Sets this recentfile in the state of not 'seeded'.
1949 =head2 $ret = $obj->update ($path, $type)
1951 =head2 $ret = $obj->update ($path, "new", $dirty_epoch)
1953 =head2 $ret = $obj->update ()
1955 Enter one file into the local I<recentfile>. $path is the (usually
1956 absolute) path. If the path is outside I<our> tree, then it is
1959 C<$type> is one of C<new> or C<delete>.
1961 Events of type C<new> may set $dirty_epoch. $dirty_epoch is normally
1962 not used and the epoch is calculated by the update() routine itself
1963 based on current time. But if there is the demand to insert a
1964 not-so-current file into the dataset, then the caller sets
1965 $dirty_epoch. This causes the epoch of the registered event to become
1966 $dirty_epoch or -- if the exact value given is already taken -- a tiny
1967 bit more. As compensation the dirtymark of the whole dataset is set to
1968 now or the current epoch, whichever is higher. Note: setting the
1969 dirty_epoch to the future is prohibited as it's very unlikely to be
1970 intended: it definitely might wreak havoc with the index files.
1972 The new file event is unshifted (or, if dirty_epoch is set, inserted
1973 at the place it belongs to, according to the rule to have a sequence
1974 of strictly decreasing timestamps) to the array of recent_events and
1975 the array is shortened to the length of the timespan allowed. This is
1976 usually the timespan specified by the interval of this recentfile but
1977 as long as this recentfile has not been merged to another one, the
1978 timespan may grow without bounds.
1980 The third form runs an update without inserting a new file. This may
1981 be desired to truncate a recentfile.
1984 sub _epoch_monotonically_increasing
{
1985 my($self,$epoch,$recent) = @_;
1986 return $epoch unless @
$recent; # the first one goes unoffended
1987 if (_bigfloatgt
("".$epoch,$recent->[0]{epoch
})) {
1990 return _increase_a_bit
($recent->[0]{epoch
});
1994 my($self,$path,$type,$dirty_epoch) = @_;
1995 if (defined $path or defined $type or defined $dirty_epoch) {
1996 die "update called without path argument" unless defined $path;
1997 die "update called without type argument" unless defined $type;
1998 die "update called with illegal type argument: $type" unless $type =~ /(new|delete)/;
2001 my $ctx = $self->_locked_batch_update([{path
=>$path,type
=>$type,epoch
=>$dirty_epoch}]);
2002 $self->write_recent($ctx->{recent
}) if $ctx->{something_done
};
2003 $self->_assert_symlink;
2007 =head2 $obj->batch_update($batch)
2009 Like update but for many files. $batch is an arrayref containing
2010 hashrefs with the structure
2022 my($self,$batch) = @_;
2024 my $ctx = $self->_locked_batch_update($batch);
2025 $self->write_recent($ctx->{recent
}) if $ctx->{something_done
};
2026 $self->_assert_symlink;
2029 sub _locked_batch_update
{
2030 my($self,$batch) = @_;
2031 my $something_done = 0;
2032 my $recent = $self->recent_events;
2033 my %paths_in_recent = map { $_->{path
} => undef } @
$recent;
2034 my $interval = $self->interval;
2035 my $canonmeth = $self->canonize;
2036 unless ($canonmeth) {
2037 $canonmeth = "naive_path_normalize";
2039 my $oldest_allowed = 0;
2040 my $setting_new_dirty_mark = 0;
2042 if ($self->verbose && @
$batch > 1) {
2043 eval {require Time
::Progress
};
2044 warn "dollarat[$@]" if $@
;
2046 $console = new Time
::Progress
;
2047 $console->attr( min
=> 1, max
=> scalar @
$batch );
2052 ITEM
: for my $item (sort {($b->{epoch
}||0) <=> ($a->{epoch
}||0)} @
$batch) {
2054 print $console->report( "\rdone %p elapsed: %L (%l sec), ETA %E (%e sec)", $i ) if $console and not $i % 50;
2055 my $ctx = $self->_update_batch_item($item,$canonmeth,$recent,$setting_new_dirty_mark,$oldest_allowed,$something_done,\
%paths_in_recent,$memo_splicepos);
2056 $something_done = $ctx->{something_done
};
2057 $oldest_allowed = $ctx->{oldest_allowed
};
2058 $setting_new_dirty_mark = $ctx->{setting_new_dirty_mark
};
2059 $recent = $ctx->{recent
};
2060 $memo_splicepos = $ctx->{memo_splicepos
};
2062 print "\n" if $console;
2063 if ($setting_new_dirty_mark) {
2064 $oldest_allowed = 0;
2066 TRUNCATE
: while (@
$recent) {
2067 # $DB::single++ unless defined $oldest_allowed;
2068 if (_bigfloatlt
($recent->[-1]{epoch
}, $oldest_allowed)) {
2070 $something_done = 1;
2075 return {something_done
=>$something_done,recent
=>$recent};
2077 sub _update_batch_item
{
2078 my($self,$item,$canonmeth,$recent,$setting_new_dirty_mark,$oldest_allowed,$something_done,$paths_in_recent,$memo_splicepos) = @_;
2079 my($path,$type,$dirty_epoch) = @
{$item}{qw(path type epoch)};
2080 if (defined $path or defined $type or defined $dirty_epoch) {
2081 $path = $self->$canonmeth($path);
2083 # you must calculate the time after having locked, of course
2084 my $now = Time
::HiRes
::time;
2087 if (defined $dirty_epoch && _bigfloatgt
($now,$dirty_epoch)) {
2088 $epoch = $dirty_epoch;
2090 $epoch = $self->_epoch_monotonically_increasing($now,$recent);
2093 my $merged = $self->merged;
2094 if ($merged->{epoch
} && !$setting_new_dirty_mark) {
2095 my $virtualnow = _bigfloatmax
($now,$epoch);
2096 # for the lower bound I think we need no big math, we calc already
2097 my $secs = $self->interval_secs();
2098 $oldest_allowed = min
($virtualnow - $secs, $merged->{epoch
}, $epoch);
2100 # as long as we are not merged at all, no limits!
2102 my $lrd = $self->localroot;
2103 if (defined $path && $path =~ s
|^\Q
$lrd\E
||) {
2106 # remove the older duplicates of this $path, irrespective of $type:
2107 if (defined $dirty_epoch) {
2108 my $ctx = $self->_update_with_dirty_epoch($path,$recent,$epoch,$paths_in_recent,$memo_splicepos);
2109 $recent = $ctx->{recent
};
2110 $splicepos = $ctx->{splicepos
};
2111 $epoch = $ctx->{epoch
};
2112 my $dirtymark = $self->dirtymark;
2114 if (_bigfloatgt
($epoch, $now)) { # just in case we had to increase it
2117 $self->dirtymark($new_dm);
2118 $setting_new_dirty_mark = 1;
2119 if (not defined $merged->{epoch
} or _bigfloatlt
($epoch,$merged->{epoch
})) {
2123 $recent = [ grep { $_->{path
} ne $path } @
$recent ];
2126 if (defined $splicepos) {
2127 splice @
$recent, $splicepos, 0, { epoch
=> $epoch, path
=> $path, type
=> $type };
2128 $paths_in_recent->{$path} = undef;
2130 $memo_splicepos = $splicepos;
2131 $something_done = 1;
2135 something_done
=> $something_done,
2136 oldest_allowed
=> $oldest_allowed,
2137 setting_new_dirty_mark
=> $setting_new_dirty_mark,
2139 memo_splicepos
=> $memo_splicepos,
2142 sub _update_with_dirty_epoch
{
2143 my($self,$path,$recent,$epoch,$paths_in_recent,$memo_splicepos) = @_;
2145 my $new_recent = [];
2146 if (exists $paths_in_recent->{$path}) {
2148 KNOWN_EVENT
: for my $i (0..$#$recent) {
2149 if ($recent->[$i]{path
} eq $path) {
2150 if ($recent->[$i]{epoch
} eq $epoch) {
2156 push @
$new_recent, $recent->[$i];
2159 @
$recent = @
$new_recent unless $cancel;
2161 if (!exists $recent->[0] or _bigfloatgt
($epoch,$recent->[0]{epoch
})) {
2163 } elsif (_bigfloatlt
($epoch,$recent->[-1]{epoch
})) {
2164 $splicepos = @
$recent;
2167 if (_bigfloatgt
($memo_splicepos<=$#$recent && $epoch, $recent->[$memo_splicepos]{epoch})) {
2170 $startingpoint = $memo_splicepos;
2172 RECENT
: for my $i ($startingpoint..$#$recent) {
2173 my $ev = $recent->[$i];
2174 if ($epoch eq $recent->[$i]{epoch
}) {
2175 $epoch = _increase_a_bit
($epoch, $i ?
$recent->[$i-1]{epoch
} : undef);
2177 if (_bigfloatgt
($epoch,$recent->[$i]{epoch
})) {
2185 splicepos
=> $splicepos,
2192 Sets this recentfile in the state of 'seeded' which means it has to
2193 re-evaluate its uptodateness.
2203 Tells if the recentfile is in the state 'seeded'.
2207 my($self, $set) = @_;
2209 $self->_seeded ($set);
2211 my $x = $self->_seeded;
2212 unless (defined $x) {
2214 $self->_seeded ($x);
2221 True if this object has mirrored the complete interval covered by the
2229 if ($self->_uptodateness_ever_reached and not $self->seeded) {
2233 # it's too easy to misconfigure ttl and related timings and then
2234 # never reach uptodateness, so disabled 2009-03-22
2235 if (0 and not defined $uptodate) {
2236 if ($self->ttl_reached){
2237 $why = "ttl_reached returned true, so we are not uptodate";
2241 unless (defined $uptodate) {
2242 # look if recentfile has unchanged timestamp
2243 my $minmax = $self->minmax;
2244 if (exists $minmax->{mtime
}) {
2245 my $rfile = $self->_my_current_rfile;
2246 my @stat = stat $rfile or die "Could not stat '$rfile': $!";
2247 my $mtime = $stat[9];
2248 if (defined $mtime && defined $minmax->{mtime
} && $mtime > $minmax->{mtime
}) {
2249 $why = "mtime[$mtime] of rfile[$rfile] > minmax/mtime[$minmax->{mtime}], so we are not uptodate";
2252 my $covered = $self->done->covered(@
$minmax{qw(max min)});
2253 $why = sprintf "minmax covered[%s], so we return that", defined $covered ?
$covered : "UNDEF";
2254 $uptodate = $covered;
2258 unless (defined $uptodate) {
2259 $why = "fallthrough, so not uptodate";
2263 $self->_uptodateness_ever_reached(1);
2267 uptodate
=> $uptodate,
2270 $self->_remember_last_uptodate_call($remember);
2274 =head2 $obj->write_recent ($recent_files_arrayref)
2276 Writes a I<recentfile> based on the current reflection of the current
2277 state of the tree limited by the current interval.
2282 @
{$_[1]} = sort { _bigfloatcmp
($b->{epoch
},$a->{epoch
}) } @
{$_[1]};
2286 my ($self,$recent) = @_;
2287 die "write_recent called without argument" unless defined $recent;
2289 SANITYCHECK
: for my $i (0..$#$recent) {
2290 if (defined($Last_epoch) and _bigfloatge
($recent->[$i]{epoch
},$Last_epoch)) {
2292 Carp
::confess
(sprintf "Warning: disorder '%s'>='%s', re-sorting %s\n",
2293 $recent->[$i]{epoch
}, $Last_epoch, $self->interval);
2295 # $self->_resort($recent);
2298 $Last_epoch = $recent->[$i]{epoch
};
2300 my $minmax = $self->minmax;
2301 if (!defined $minmax->{max
} || _bigfloatlt
($minmax->{max
},$recent->[0]{epoch
})) {
2302 $minmax->{max
} = $recent->[0]{epoch
};
2304 if (!defined $minmax->{min
} || _bigfloatlt
($minmax->{min
},$recent->[-1]{epoch
})) {
2305 $minmax->{min
} = $recent->[-1]{epoch
};
2307 $self->minmax($minmax);
2308 my $meth = sprintf "write_%d", $self->protocol;
2309 $self->$meth($recent);
2312 =head2 $obj->write_0 ($recent_files_arrayref)
2314 Delegate of C<write_recent()> on protocol 0
2319 my ($self,$recent) = @_;
2320 my $rfile = $self->rfile;
2321 YAML
::Syck
::DumpFile
("$rfile.new",$recent);
2322 rename "$rfile.new", $rfile or die "Could not rename to '$rfile': $!";
2325 =head2 $obj->write_1 ($recent_files_arrayref)
2327 Delegate of C<write_recent()> on protocol 1
2332 my ($self,$recent) = @_;
2333 my $rfile = $self->rfile;
2334 my $suffix = $self->serializer_suffix;
2336 meta
=> $self->meta_data,
2340 if ($suffix eq ".yaml") {
2341 $serialized = YAML
::Syck
::Dump
($data);
2342 } elsif ($HAVE->{"Data::Serializer"}) {
2343 my $serializer = Data
::Serializer
->new
2344 ( serializer
=> $serializers{$suffix} );
2345 $serialized = $serializer->raw_serialize($data);
2347 die "Data::Serializer not installed, cannot proceed with suffix '$suffix'";
2349 open my $fh, ">", "$rfile.new" or die "Could not open >'$rfile.new': $!";
2350 print $fh $serialized;
2351 close $fh or die "Could not close '$rfile.new': $!";
2352 rename "$rfile.new", $rfile or die "Could not rename to '$rfile': $!";
2356 my $nq = qr/[^"]+/; # non-quotes
2358 split /\n/, <<'=cut'; %serializers = map { my @x = /"($nq)"\s+=>\s+"($nq)"/; @x } grep {s/^=item\s+C<<\s+(.+)\s+>>$/$1/} @pod_lines; }
2362 The following suffixes are supported and trigger the use of these
2367 =item C<< ".yaml" => "YAML::Syck" >>
2369 =item C<< ".json" => "JSON" >>
2371 =item C<< ".sto" => "Storable" >>
2373 =item C<< ".dd" => "Data::Dumper" >>
2381 split /\n/, <<'=cut'; %seconds = map { eval } grep {s/^=item\s+C<<(.+)>>$/$1/} @pod_lines; }
2383 =head1 INTERVAL SPEC
2385 An interval spec is a primitive way to express time spans. Normally it
2386 is composed from an integer and a letter.
2388 As a special case, a string that consists only of the single letter
2389 C<Z>, stands for MAX_INT seconds.
2391 The following letters express the specified number of seconds:
2397 =item C<< m => 60 >>
2399 =item C<< h => 60*60 >>
2401 =item C<< d => 60*60*24 >>
2403 =item C<< W => 60*60*24*7 >>
2405 =item C<< M => 60*60*24*30 >>
2407 =item C<< Q => 60*60*24*90 >>
2409 =item C<< Y => 60*60*24*365.25 >>
2417 L<File::Rsync::Mirror::Recent>,
2418 L<File::Rsync::Mirror::Recentfile::Done>,
2419 L<File::Rsync::Mirror::Recentfile::FakeBigFloat>
2423 Please report any bugs or feature requests through the web interface
2425 L<http://rt.cpan.org/NoAuth/ReportBug.html?Queue=File-Rsync-Mirror-Recentfile>.
2426 I will be notified, and then you'll automatically be notified of
2427 progress on your bug as I make changes.
2431 Memory hungry: it seems all memory is allocated during the initial
2432 rsync where a list of all files is maintained in memory.
2436 You can find documentation for this module with the perldoc command.
2438 perldoc File::Rsync::Mirror::Recentfile
2440 You can also look for information at:
2444 =item * RT: CPAN's request tracker
2446 L<http://rt.cpan.org/NoAuth/Bugs.html?Dist=File-Rsync-Mirror-Recentfile>
2448 =item * AnnoCPAN: Annotated CPAN documentation
2450 L<http://annocpan.org/dist/File-Rsync-Mirror-Recentfile>
2452 =item * CPAN Ratings
2454 L<http://cpanratings.perl.org/d/File-Rsync-Mirror-Recentfile>
2458 L<http://search.cpan.org/dist/File-Rsync-Mirror-Recentfile>
2463 =head1 ACKNOWLEDGEMENTS
2465 Thanks to RJBS for module-starter.
2471 =head1 COPYRIGHT & LICENSE
2473 Copyright 2008,2009 Andreas König.
2475 This program is free software; you can redistribute it and/or modify it
2476 under the same terms as Perl itself.
2481 1; # End of File::Rsync::Mirror::Recentfile
2485 # cperl-indent-level: 4