view release on metacpan or  search on metacpan

lib/App/Glacier/Command/Get.pm  view on Meta::CPAN

    return if $self->dry_run;
    my $fd = $self->_open_output($localname);
    my ($res, $tree_hash) = $self->glacier->Get_job_output($job->vault,
							   $job->id);
    if ($self->glacier->lasterr) {
	$self->abend(EX_FAILURE, "downoad failed: ",
		     $self->glacier->last_error_message);
    }
    syswrite($fd, $res);
    close($fd);
    return $tree_hash;
}

sub _download_multipart {
    my ($self, $job, $localname) = @_;
        
    my $glacier = $self->{_glacier};

    my $tree_hash;
    
    my $njobs = $self->{_options}{jobs}
                || $self->cf_transfer_param(qw(download jobs));

    my $archive_size = $job->get('ArchiveSizeInBytes');
    my $part_size;
    # Compute approximate part size
    $part_size = ($archive_size - 1) / 10000;
    if ($part_size < TWOMB) {
	$part_size = TWOMB;
    } else {
	# Make sure the chunk is Tree-Hash aligned
	# http://docs.aws.amazon.com/amazonglacier/latest/dev/checksum-calculations-range.html?shortFooter=true#checksum-calculations-upload-archive-with-ranges
	$part_size = TWOMB * 2 ** int(log($part_size / TWOMB) / log(2) + 1);
    }
    # Number of parts to download:
    my $total_parts = int(($archive_size + $part_size - 1) / $part_size);
    # Compute the number of parts per job
    my $job_parts = int(($total_parts + $njobs - 1) / $njobs);

    $self->debug(1, "$job: downloading in chunks of $part_size bytes, in $njobs jobs, with $job_parts parts per job");

    return if $self->dry_run;

    use Fcntl qw(SEEK_SET);

    my $fd = $self->_open_output($localname);
    my @part_hashes :shared = ();
    my $p = new App::Glacier::Progress($total_parts,
				       prefix => $localname)
	unless $self->{_options}{quiet};
    for (my $i = 0; $i < $njobs; $i++) {
	my ($thr) = threads->create(
	    sub {
		my ($job_idx) = @_;
		# Number of part to start from
		my $part_idx = $job_idx * $job_parts;
		# Offset in file
		my $off = $part_idx * $part_size;
		# Number of retries in case of failure
		my $retries = $self->cf_transfer_param(qw(download retries));
		Scalar::Util::weaken($p);
		for (my $j = 0; $j < $job_parts;
		     $j++, $part_idx++, $off += $part_size) {
		    last if $off >= $archive_size;
		    if ($part_size > $archive_size - $off) {
			$part_size = $archive_size - $off;
		    }
		    my $range = 'bytes=' . $off . '-' . ($off + $part_size - 1);
		    my ($res, $hash);
		    for (my $try = 0;;) {
			($res, $hash) =
			    $self->glacier->Get_job_output($job->vault,
							   $job->id, $range);
			if ($self->glacier->lasterr) {
			    if (++$try < $retries) {
				$self->debug(1, "part $part_idx: ",
					     $self->glacier->last_error_message);
				$self->debug(1, "retrying");
			    } else {
				$self->error("failed to download part $part_idx: ",
					     $self->glacier->last_error_message);
				return 0;
			    }
			} else {
			    last;
			}
		    }

		    lock @part_hashes;
		    seek($fd, $off, SEEK_SET);
		    syswrite($fd, $res);
		    $part_hashes[$part_idx] = $hash;
		    $p->update if $p;
		}
		return 1;
	    }, $i);
    }
    
    $self->debug(2, "waiting for download to finish");
    foreach my $thr (threads->list()) {
	# FIXME: error handling
	$thr->join() or croak "thread $thr failed";
    }
    $p->finish('downloaded') if $p;
    close($fd);
    return $glacier->_tree_hash_from_array_ref(\@part_hashes);
}
    
1;