File Coverage

blib/lib/DataStore/CAS/FS/Importer.pm

Criterion	Covered	Total	%
statement	176	268	65.6
branch	66	160	41.2
condition	12	66	18.1
subroutine	32	67	47.7
pod	17	19	89.4
total	303	580	52.2

package DataStore::CAS::FS::Importer;
use 5.008;
use Moo;
use Carp;
use Try::Tiny;
use File::Spec::Functions 'catfile', 'catdir', 'splitpath', 'catpath';
use Fcntl;
use DataStore::CAS::FS::InvalidUTF8;
use DataStore::CAS::FS::DirCodec;

our $VERSION= '0.011000';

# ABSTRACT: Copy files from filesystem into DataStore::CAS::FS.


our %_flag_defaults;
BEGIN {
	%_flag_defaults= (
		die_on_dir_error  => 1,
		die_on_file_error => 1,
		die_on_hint_error => 0,
		die_on_metadata_error => 0,
		collect_metadata_ts => 1,
		collect_access_ts => 0,
		collect_unix_perm => 1,
		collect_unix_misc => 0,
		collect_acl       => 0,
		collect_ext_attr  => 0,
		follow_symlink    => 0,
		cross_mountpoints => 0,
		reuse_digests     => 1,
		utf8_filenames    => 1,
	);
	for (keys %_flag_defaults) {
		eval "sub $_ { \$_[0]{flags}{$_}= \$_[1] if \@_ > 1; \$_[0]{flags}{$_} }; 1" or die $@
	}
}
sub _flag_defaults {
	\%_flag_defaults;
}

has dir_format     => ( is => 'rw', default => sub { 'universal' } );
has filter         => ( is => 'rw' );
has flags          => ( is => 'rw', default => sub { { } } );
has unix_uid_cache => ( is => 'rw', default => sub { {} } );
has unix_gid_cache => ( is => 'rw', default => sub { {} } );
has _hint_check_fn => ( is => 'rwp' );

sub _handle_hint_error {
	croak $_[1] if $_[0]->die_on_hint_error;
	warn "$_[1]\n";
}

sub _handle_file_error {
	croak $_[1] if $_[0]->die_on_file_error;
	warn "$_[1]\n";
}

sub _handle_dir_error {
	croak $_[1] if $_[0]->die_on_dir_error;
	warn "$_[1]\n";
}

sub _handle_metadata_error {
	croak $_[1] if $_[0]->die_on_metadata_error;
	warn "$_[1]\n";
}


sub BUILD {
	my ($self, $args)= @_;
	my $flags= $self->flags;
	my $defaults= $self->_flag_defaults;
	for (keys %$defaults) {
		$flags->{$_}= delete $args->{$_}
			if exists $args->{$_};
		$flags->{$_}= $_flag_defaults{$_}
			unless defined $flags->{$_};
	}
	defined $defaults->{$_} || croak "Unknown flag: '$_'"
		for keys %$flags;
	$self->can($_) || croak "Unknown attribute: '$_'"
		for keys %$args;
}

# locally-scoped to the device number which we should stay on
our $_DEVICE_CONSTRAINT;


sub import_tree {
	my ($self, $src, $dest)= @_;

	my $stat= $self->_stat($src)
		or croak "Source does not exist";

	local $_DEVICE_CONSTRAINT= $stat->dev
		unless defined $_DEVICE_CONSTRAINT or $self->cross_mountpoints;

	$self->_build__hint_check_fn;

	my $ent_name= $self->_entname_from_path($src);
	my $ent= $self->_import_directory_entry($dest->filesystem->store, $src, $ent_name, $stat, $dest);
	$dest->filesystem->set_path($dest->path_names, $ent);
	1;
}


sub import_directory {
	my ($self, $cas, $path, $hint)= @_;

	my $stat= $self->_stat($path)
		or croak "Source does not exist";

	local $_DEVICE_CONSTRAINT= $stat->dev
		unless defined $_DEVICE_CONSTRAINT or $self->cross_mountpoints;

	$self->_build__hint_check_fn;

	$self->_import_directory($cas, $path, $hint);
}

sub _import_directory {
	my ($self, $cas, $path, $hint)= @_;
	my $names= $self->_readdir($path)
		or return undef;
	my @entries;
	my $filter= $self->filter;
	for my $ent_name (@$names) {
		my $ent_path= catfile($path, $ent_name);
		my $stat= $self->_stat($ent_path);

		if ($self->utf8_filenames) {
			$ent_name= DataStore::CAS::FS::InvalidUTF8->decode_utf8($ent_name);
		} else {
			utf8::upgrade($ent_name);
		}

		my $keep= $filter? $filter->($ent_name, $ent_path, $stat) : 1;
		next unless $keep;

		# Check for crossing mount point.
		if (defined $_DEVICE_CONSTRAINT && $stat->dev ne $_DEVICE_CONSTRAINT) {
			# TODO: log skipped mount points
			# Metadata comes from mounted filesystem, so ignore it
			push @entries, { type => 'dir', name => $ent_name };
		}
		# If keep is < 0, store the metadata but not the file/dir
		elsif ($keep < 0) {
			push @entries, $self->collect_dirent_metadata($ent_path, $ent_name, $stat);
		}
		# Else recursively store the whole thing
		else {
			push @entries, $self->_import_directory_entry($cas, $ent_path, $ent_name, $stat, $hint);
		}
	}
	return DataStore::CAS::FS::DirCodec->put($cas, $self->dir_format, \@entries, {} );
}


sub import_directory_entry {
	my ($self, $cas, $path, $ent_name, $stat, $hint)= @_;

	$stat||= $self->_stat($path)
		or croak "Source does not exist";

	$self->_build__hint_check_fn;

	local $_DEVICE_CONSTRAINT= $stat->dev
		unless defined $_DEVICE_CONSTRAINT or $self->cross_mountpoints;

	$ent_name= $self->_entname_from_path($path)
		unless defined $ent_name;
	return DataStore::CAS::FS::DirEnt->new(
		$self->_import_directory_entry($cas, $path, $ent_name, $stat, $hint)
	);
}

sub _import_directory_entry {
	my ($self, $cas, $ent_path, $ent_name, $stat, $hint)= @_;
	my $attrs= $self->collect_dirent_metadata($ent_path, $ent_name, $stat)
		or croak "Path does not exist: '$ent_path'";
	if ($attrs->{type} eq 'file') {
		if ($hint && $self->_can_reuse_hash($attrs, $hint)) {
			$attrs->{ref}= $hint->ref;
		} else {
			my $err;
			$attrs->{ref}= try { $cas->put_file($ent_path); } catch { $err= $_; undef; };
			$self->_handle_file_error("Error while importing file '$ent_path': $err")
				if defined $err;
		}
	}
	elsif ($attrs->{type} eq 'dir') {
		if (defined $_DEVICE_CONSTRAINT && $stat->dev ne $_DEVICE_CONSTRAINT) {
			# TODO: log skipped mount points
		} else {
			local $_DEVICE_CONSTRAINT= $stat->dev
				unless defined $_DEVICE_CONSTRAINT || $self->cross_mountpoints;

			my $subdir_hint;
			if (defined $hint) {
				my $err;
				try {
					$subdir_hint= $hint->path_if_exists($attrs->{name});
					$subdir_hint->resolve
						if $subdir_hint;
				} catch {
					$err= $_;
				};
				$self->_handle_hint_error("Error while loading virtual path '".$hint->resolved_canonical_path.'/'.$attrs->{name}."': $err")
					if defined $err;
			}
			$attrs->{ref}= $self->_import_directory($cas, $ent_path, $subdir_hint);
		}
	}
	return $attrs;
}


our %_ModeToType;
# Making this a function allows other code to call it in a BEGIN block if needed
sub _build_ModeToType {
	local $@;
	eval { $_ModeToType{Fcntl::S_IFREG()}= 'file'     };
	eval { $_ModeToType{Fcntl::S_IFDIR()}= 'dir'      };
	eval { $_ModeToType{Fcntl::S_IFLNK()}= 'symlink'  };
	eval { $_ModeToType{Fcntl::S_IFBLK()}= 'blockdev' };
	eval { $_ModeToType{Fcntl::S_IFCHR()}= 'chardev'  };
	eval { $_ModeToType{Fcntl::S_IFIFO()}= 'pipe'     };
	eval { $_ModeToType{Fcntl::S_IFWHT()}= 'whiteout' };
	eval { $_ModeToType{Fcntl::S_IFSOCK()}= 'socket'  };
}

_build_ModeToType();

sub collect_dirent_metadata {
	my ($self, $path, $ent_name, $stat)= @_;

	$stat ||= $self->_stat($path)
		or return undef;

	$ent_name= $self->_entname_from_path($path)
		unless defined $ent_name;

	my %attrs= (
		type => ($_ModeToType{$stat->[2] & Fcntl::S_IFMT()}),
		name => $ent_name,
		size => $stat->[7],
		modify_ts => $stat->[9],
	);
	if (!defined $attrs{type}) {
		$self->_handle_dir_error("Type of dirent is unknown: ".($stat->[2] & Fcntl::S_IFMT()));
		$attrs{type}= 'file';
	}
	if ($self->{flags}{collect_unix_perm}) {
		$attrs{unix_mode}= ($stat->[2] & ~Fcntl::S_IFMT());
		my $uid= $attrs{unix_uid}= $stat->[4];
		if (my $cache= $self->unix_uid_cache) {
			if (!exists $cache->{$uid}) {
				my $name= getpwuid($uid);
				if (!defined $name) {
					$self->_handle_metadata_error("No username for UID $uid");
				} elsif ($self->utf8_filenames) {
					$name= DataStore::CAS::FS::InvalidUTF8->decode_utf8($name);
				} else {
					utf8::upgrade($name);
				}
				$cache->{$uid}= $name;
			}
			$attrs{unix_user}= $cache->{$uid}
				if defined $cache->{$uid};
		}
		my $gid= $attrs{unix_gid}= $stat->[5];
		if (my $cache= $self->unix_gid_cache) {
			if (!exists $cache->{$gid}) {
				my $name= getgrgid($gid);
				if (!defined $name) {
					$self->_handle_metadata_error("No groupname for GID $gid");
				} elsif ($self->utf8_filenames) {
					$name= DataStore::CAS::FS::InvalidUTF8->decode_utf8($name);
				} else {
					utf8::upgrade($name);
				}
				$cache->{$gid}= $name;
			}
			$attrs{unix_group}= $cache->{$gid}
				if defined $cache->{$gid};
		}
	}
	if ($self->{flags}{collect_metadata_ts}) {
		$attrs{metadata_ts}= $stat->[10];
	}
	if ($self->{flags}{collect_access_ts}) {
		$attrs{access_ts}= $stat->[8];
	}
	if ($self->{flags}{collect_unix_misc}) {
		$attrs{unix_dev}= $stat->[0];
		$attrs{unix_inode}= $stat->[1];
		$attrs{unix_nlink}= $stat->[3];
		$attrs{unix_blocksize}= $stat->[11];
		$attrs{unix_blockcount}= $stat->[12];
	}
	if ($self->{flags}{collect_acl}) {
		# TODO
	}
	if ($self->{flags}{collect_ext_attr}) {
		# TODO
	}
	if ($attrs{type} eq 'dir') {
		delete $attrs{size};
	}
	elsif ($attrs{type} eq 'symlink') {
		$attrs{ref}= readlink $path;
	}
	elsif ($attrs{type} eq 'blockdev' or $attrs{type} eq 'chardev') {
		$attrs{ref}= $self->_split_dev_node($stat->[6]);
	}
	\%attrs;
}

sub _build__hint_check_fn {
	my $self= shift;
	my $reuse= $self->reuse_digests;
	return $self->{_hint_check_fn}= $reuse > 1?
		($reuse > 2? \&_hint_check_ctime : \&_hint_check_mtime)
		: ($reuse > 0? \&_hint_check_size : \&_hint_check_none);
}

sub _hint_check_none {
	return undef;
}
sub _hint_check_size {
	my ($self, $attrs, $hint)= @_;
	return undef unless defined $hint && defined $hint->ref;
	my ($size, $h_size)= ($attrs->{size}, $hint->size);
	return defined $size && defined $h_size && $size eq $h_size;
}
sub _hint_check_mtime {
	my ($self, $attrs, $hint)= @_;
	return undef unless defined $hint && defined $hint->ref;
	my ($size, $h_size)= ($attrs->{size}, $hint->size);
	return undef unless defined $size && defined $h_size && $size eq $h_size;
	my ($modify_ts, $h_modify_ts)= ($attrs->{modify_ts}, $hint->modify_ts);
	return defined $modify_ts && defined $h_modify_ts && $modify_ts eq $h_modify_ts;
}
sub _hint_check_ctime {
	my ($self, $attrs, $hint)= @_;
	return undef unless defined $hint && defined $hint->ref;
	my ($size, $h_size)= ($attrs->{size}, $hint->size);
	return undef unless defined $size && defined $h_size && $size eq $h_size;
	my ($modify_ts, $h_modify_ts)= ($attrs->{metadata_ts}, $hint->metadata_ts);
	return defined $modify_ts && defined $h_modify_ts && $modify_ts eq $h_modify_ts;
}

sub _entname_from_path {
	my ($self, $path)= @_;
	my (undef, undef, $ent_name)= splitpath($path);
	if ($self->utf8_filenames) {
		$ent_name= DataStore::CAS::FS::InvalidUTF8->decode_utf8($ent_name);
	} else {
		utf8::upgrade($ent_name);
	}
	$ent_name;
}

sub _split_dev_node {
	($_[1] >> 8).','.($_[1] & 0xFF);
}

sub _stat {
	my $fn= \&_stat_unix;
	no warnings 'redefine';
	*_stat= $fn;
	goto $fn;
}

sub _stat_unix {
	my ($self, $path)= @_;
	my @stat= $self->follow_symlink? stat($path) : lstat($path);
	unless (@stat) {
		$self->_handle_dir_error("Can't stat '$path': $!");
		return undef;
	}
	bless \@stat, 'DataStore::CAS::FS::Importer::FastStat';
}

sub _readdir {
	my $fn= \&_readdir_unix;
	no warnings 'redefine';
	*_readdir= $fn;
	goto $fn;
}

sub _readdir_unix {
	my ($self, $path)= @_;
	my $dh;
	if (!opendir($dh, $path)) {
		$self->_handle_dir_error("opendir($path): $!");
		return undef;
	}

	my @names= grep { $_ ne '.' && $_ ne '..' } readdir($dh);

	if (!closedir $dh) {
		$self->_handle_dir_error("closedir($path): $!");
		return undef;
	}

	\@names;
}

package DataStore::CAS::FS::Importer::FastStat;
use strict;
use warnings;


sub dev     { $_[0][0] }
sub ino     { $_[0][1] }
sub mode    { $_[0][2] }
sub nlink   { $_[0][3] }
sub uid     { $_[0][4] }
sub gid     { $_[0][5] }
sub rdev    { $_[0][6] }
sub size    { $_[0][7] }
sub atime   { $_[0][8] }
sub mtime   { $_[0][9] }
sub ctime   { $_[0][10] }
sub blksize { $_[0][11] }
sub blocks  { $_[0][12] }

1;

__END__