stream_tar.pl

gistfile1.txt

#!/usr/bin/env perl

use v5.40;
use Mojo::Base -strict, -signatures;

use Fcntl qw(SEEK_SET);
use Getopt::Long qw(GetOptions);

use Mojo::Collection;
use Mojo::Log;
use Mojo::Promise;
use Mojo::Tar::File;
use Mojo::URL;
use Mojo::UserAgent;
use Mojo::Util qw(humanize_bytes);
use Time::HiRes qw(gettimeofday tv_interval);

my $concurrency = 5;
my $chunk_size  = 1024 * 1024;
my $base_url    = 'http://localhost:3000/';

GetOptions(
    'concurrency=i' => \$concurrency,
    'chunk-size=i'  => \$chunk_size,
    'url=s'         => \$base_url,
) or die "Invalid options\n";

my $tar_path = shift
    or die <<"USAGE";
Usage:
  $0 [--concurrency 5]
     [--chunk-size 1048576]
     [--url http://localhost:3000/upload]
     archive.tar
USAGE

die "--concurrency must be >= 1\n" unless $concurrency >= 1;

my $log = Mojo::Log->new(level => $ENV{MOJO_LOG_LEVEL});
my $ua  = Mojo::UserAgent->new;

# Scan the TAR.
# This reads ONLY the 512-byte headers.
# Member payloads are skipped with sysseek().
sub index_tar ($path) {
  open my $fh, '<:raw', $path or die "$path: $!";

  my @members;
  my $header_offset = 0;

  while (1) {
    my $n = sysread $fh, my $header, 512, 0;

    die "$path: $!" unless defined $n;

    last unless $n;

    die "Short TAR header at offset $header_offset\n"
      unless $n == 512;

    last if $header eq "\0" x 512; # Two zero blocks normally terminate a TAR.

    my $file = Mojo::Tar::File->new->from_header($header);

    die sprintf("Invalid TAR header checksum at offset %d\n", $header_offset) unless $file->checksum;

    my $size        = $file->size;
    my $data_offset = $header_offset + 512;

    if ($file->type eq '0') { # A regular file.
      push @members => {file => $file, offset => $data_offset, size => $size};
      $log->info(sprintf('[INDEX] %-45s offset=%12d size=%12d', $file->path, $data_offset, $size));
    }

    #
    # TAR member contents are padded to a 512-byte boundary.
    #
    #             size + 511
    # blocks = ----------------
    #                512
    #
    my $padded_size = int(($size + 511) / 512) * 512;
    $header_offset = $data_offset + $padded_size;

    sysseek($fh, $header_offset, SEEK_SET) == $header_offset
      or die "Cannot seek to $header_offset in $path: $!";
  }

  close $fh;

  $log->debug(sprintf '%d regular files', scalar @members);
  return @members;
}

sub upload_url ($name) {
  my $url = Mojo::URL->new($base_url);

  my $path = $url->path->to_string;
  $path =~ s{/$}{};

  $url->path("$path/$name");

  return $url;
}

# Stream exactly one byte range directly from the original TAR.
sub put_member_p ($member) {
  my $file   = $member->{file};
  my $name   = $file->path;
  my $offset = $member->{offset};
  my $size   = $member->{size};

  open my $fh, '<:raw', $tar_path or return Mojo::Promise->reject("$tar_path: $!");

  unless (defined sysseek $fh, $offset, SEEK_SET) {
    close $fh;
    return Mojo::Promise->reject("Cannot seek to $offset for $name: $!");
  }

  $log->info(sprintf('[START] %-45s offset=%12d size=%12d', $name, $offset, $size));

  my $tx = $ua->build_tx(PUT => upload_url($name));
  $tx->req->headers->content_length($size);

  my $remaining = $size;
  my $sent      = 0;

  my $content = $tx->req->content;
  my $drain = sub ($content, $offset=undef) {
    return unless $remaining;

    my $want = $remaining < $chunk_size ? $remaining : $chunk_size;
    my $n    = sysread($fh, my $buf, $want);

    die "Read failure for $name: $!" unless defined $n;
    die sprintf('Unexpected EOF reading %s: sent=%d expected=%d', $name, $sent, $size) unless $n;

    $remaining -= $n;
    $sent      += $n;

    $log->trace(sprintf('[SEND ] %-45s %12d / %12d', $name, $sent, $size));

    # On the LAST chunk, do not install another drain callback.
    $content->write($buf => $remaining ? __SUB__ : undef);
  };

  my $start = [gettimeofday];
  $ua->start_p($tx->tap(sub{shift->req->content->$drain}))->then(sub ($tx) {
    my $elapsed = tv_interval($start, [gettimeofday]);
    my $res = $tx->result;

    die sprintf('%s: HTTP PUT failed: %s %s', $name, $res->code // 'no response', $res->message // '',)
      unless $res->is_success;

    $log->info(sprintf('[DONE ] %-45s sent %s bytes in %.4fs (%s/s) HTTP %s', $name, humanize_bytes($sent), $elapsed, humanize_bytes($sent/$elapsed), $res->code));

    return {name => $name, bytes => $sent, elapsed => $elapsed};
  })->finally(sub {
    close $fh if defined fileno $fh;
  });
}

my @members = index_tar($tar_path);

my $start = [gettimeofday];
Mojo::Promise->map({concurrency => $concurrency}, sub { put_member_p($_) }, @members)->then(sub (@results) {
  my $elapsed = tv_interval($start, [gettimeofday]);
  my $results = Mojo::Collection->new(@results)->flatten;
  my $bytes = $results->reduce(sub{$a+$b->{bytes}}, 0);
  $log->info(sprintf 'All uploads complete. Sent %s bytes in %.4fs (%s/s)', humanize_bytes($bytes), $elapsed, humanize_bytes($bytes/$elapsed));
})->catch(sub ($err) {
  $log->error("Upload failed: $err");
})->wait;

__END__

$ MOJO_LOG_LEVEL=trace MOJO_INACTIVITY_TIMEOUT=3600 MOJO_MAX_MESSAGE_SIZE=7516192768 perl -Mojo -E 'a("/*loc" => sub { $_->render(json => {size => 1}) })->start' prefork
$ MOJO_LOG_LEVEL=debug MOJO_INACTIVITY_TIMEOUT=3600 MOJO_MAX_MESSAGE_SIZE=7516192768 perl mojo-tar.pl --concurrency 2 3.15G.tar 
[2026-08-25 22:31:03.88334] [8250] [info] [INDEX] 5G                                            offset=         512 size=  5368709120
[2026-08-25 22:31:03.88342] [8250] [info] [INDEX] 7G                                            offset=  5368710144 size=  7516192768
[2026-08-25 22:31:03.88346] [8250] [info] [INDEX] 3G                                            offset= 12884903424 size=  3221225472
[2026-08-25 22:31:03.88347] [8250] [debug] 3 regular files
[2026-08-25 22:31:03.88349] [8250] [info] [START] 5G                                            offset=         512 size=  5368709120
[2026-08-25 22:31:03.88455] [8250] [info] [START] 7G                                            offset=  5368710144 size=  7516192768
[2026-08-25 22:31:10.99932] [8250] [info] [DONE ] 5G                                            sent 5GiB bytes in 7.1156s (720MiB/s) HTTP 200
[2026-08-25 22:31:10.99952] [8250] [info] [START] 3G                                            offset= 12884903424 size=  3221225472
[2026-08-25 22:31:13.53103] [8250] [info] [DONE ] 7G                                            sent 7GiB bytes in 9.6464s (743MiB/s) HTTP 200
[2026-08-25 22:31:14.35784] [8250] [info] [DONE ] 3G                                            sent 3GiB bytes in 3.3582s (915MiB/s) HTTP 200
[2026-08-25 22:31:14.35803] [8250] [info] All uploads complete. Sent 15GiB bytes in 10.4745s (1.4GiB/s)

Mojo-Tar-File.pm

package Mojo::Tar::File;
use Mojo::Base -base, -signatures;

use Carp         qw(croak);
use Exporter     qw(import);
use Mojo::File   ();
use Scalar::Util qw(blessed);

use constant DEBUG => !!$ENV{MOJO_TAR_DEBUG};

our $GID = $( =~ /(\d+)/ && int($1) || 0;
our ($PACK_FORMAT, @EXPORT);

BEGIN {
  $PACK_FORMAT = q(
    a100 # pos=0   name=name      desc=file name (chars)
    a8   # pos=100 name=mode      desc=file mode (octal)
    a8   # pos=108 name=uid       desc=uid (octal)
    a8   # pos=116 name=gid       desc=gid (octal)
    a12  # pos=124 name=size      desc=size (octal)
    a12  # pos=136 name=mtime     desc=mtime (octal)
    a8   # pos=148 name=checksum  desc=checksum (octal)
    a1   # pos=156 name=type      desc=type
    a100 # pos=157 name=symlink   desc=file symlink destination (chars)
    A6   # pos=257 name=ustar     desc=ustar
    a2   # pos=263 name=ustar_ver desc=ustar version (00)
    a32  # pos=265 name=owner     desc=owner user name (chars)
    a32  # pos=297 name=group     desc=owner group name (chars)
    a8   # pos=329 name=dev_major desc=device major number
    a8   # pos=337 name=dev_minor desc=device minor number
    a155 # pos=345 name=prefix    desc=file name prefix
    a12  # pos=500 name=padding   desc=padding (\0)
  );

  # Generate constants:
  # TAR_USTAR_NAME_LEN       TAR_USTAR_NAME_POS
  # TAR_USTAR_MODE_LEN       TAR_USTAR_MODE_POS
  # TAR_USTAR_UID_LEN        TAR_USTAR_UID_POS
  # TAR_USTAR_GID_LEN        TAR_USTAR_GID_POS
  # TAR_USTAR_SIZE_LEN       TAR_USTAR_SIZE_POS
  # TAR_USTAR_MTIME_LEN      TAR_USTAR_MTIME_POS
  # TAR_USTAR_CHECKSUM_LEN   TAR_USTAR_CHECKSUM_POS
  # TAR_USTAR_TYPE_LEN       TAR_USTAR_TYPE_POS
  # TAR_USTAR_SYMLINK_LEN    TAR_USTAR_SYMLINK_POS
  # TAR_USTAR_USTAR_LEN      TAR_USTAR_USTAR_POS
  # TAR_USTAR_USTAR_VER_LEN  TAR_USTAR_USTAR_VER_POS
  # TAR_USTAR_OWNER_LEN      TAR_USTAR_OWNER_POS
  # TAR_USTAR_GROUP_LEN      TAR_USTAR_GROUP_POS
  # TAR_USTAR_DEV_MAJOR_LEN  TAR_USTAR_DEV_MAJOR_POS
  # TAR_USTAR_DEV_MINOR_LEN  TAR_USTAR_DEV_MINOR_POS
  # TAR_USTAR_PREFIX_LEN     TAR_USTAR_PREFIX_POS
  # TAR_USTAR_PADDING_LEN    TAR_USTAR_PADDING_POS
  for my $line (split /\n/, $PACK_FORMAT) {
    my ($len, $pos, $name) = $line =~ /(\d+)\W+pos=(\d+)\W+name=(\w+)/ or next;

    my $const = uc "TAR_USTAR_${name}_LEN";
    constant->import($const => $len);
    push @EXPORT, $const;

    $const = uc "TAR_USTAR_${name}_POS";
    constant->import($const => $pos);
    push @EXPORT, $const;
  }
}

has asset => sub ($self) {Mojo::File::tempfile};
has checksum =>
  sub ($self) { substr $self->to_header, TAR_USTAR_CHECKSUM_POS, TAR_USTAR_CHECKSUM_LEN };
has dev_major   => '';
has dev_minor   => '';
has gid         => sub ($self) { $self->_stat('gid')  || $GID };
has group       => sub ($self) { getgrgid($self->gid) || '' };
has is_complete => sub ($self) { $self->_stat('size') == $self->size ? 1 : 0 };
has mode        => sub ($self) { ($self->_stat('mode') || 0) & 0777 };
has mtime       => sub ($self) { $self->_stat('mtime')   || time };
has owner       => sub ($self) { getpwuid($self->uid)    || '' };
has path        => sub ($self) { $self->asset->to_string || '' };
has size        => sub ($self) { $self->_stat('size')    || 0 };
has symlink     => '';
has type        => sub ($self) { $self->_build_type };
has uid         => sub ($self) { $self->_stat('uid') || $( };

sub add_block ($self, $block) {
  return $self unless $self->type eq 0;

  $self->{bytes_added} //= 0;
  my $chunk = substr $block, 0, $self->size - $self->{bytes_added};
  $self->{bytes_added} += length $chunk;
  croak 'File size is out of range' if $self->{bytes_added} > $self->size;

  my $handle = $self->{add_block_handle} //= $self->asset->open('>');
  ($handle->syswrite($chunk) // -1) == length $chunk or croak "Can't write to asset: $!";
  $self->is_complete(1)->_cleanup if $self->{bytes_added} == $self->size;

  warn sprintf "[tar:add_block] chunk=%s/%s size=%s/%s is_complete=%s path=%s\n", length($chunk),
    length($block), $self->{bytes_added}, $self->size, $self->is_complete, $self->path
    if DEBUG;

  return $self;
}

sub from_header ($self, $header) {
  my @fields   = unpack $PACK_FORMAT, $header;
  my $checksum = $self->_checksum($header);

  my ($prefix, $path) = map { _trim_nul($fields[$_]) } 15, 0;
  $path = Mojo::File->new($prefix, $path)->to_string if length $prefix;

  $self->path($path);
  $self->mode(_from_oct($fields[1]));
  $self->uid(_from_oct($fields[2]));
  $self->gid(_from_oct($fields[3]));
  $self->size(_from_number($fields[4]));
  $self->mtime(_from_oct($fields[5]));
  $self->checksum($checksum eq $fields[6] =~ s/\0\s$//r ? $checksum : '');
  $self->type($fields[7] eq "\0"                        ? '0'       : $fields[7]);
  $self->symlink(_trim_nul($fields[8]));
  $self->owner(_trim_nul($fields[11]));
  $self->group(_trim_nul($fields[12]));
  $self->dev_major($fields[13]);
  $self->dev_minor($fields[14]);

  warn sprintf
    "[tar:from_header] path=%s mode=%s uid=%s gid=%s size=%s mtime=%s checksum=%s type=%s symlink=%s owner=%s group=%s\n",
    map { $self->$_ } qw(path mode uid gid size mtime checksum type symlink owner group)
    if DEBUG;

  return $self;
}

sub to_header ($self) {
  my ($name, $prefix) = (Mojo::File->new($self->path), '');
  ($name, $prefix) = ($name->basename, $name->dirname->to_string) if length($name) > 100;
  croak qq(path "@{[$self->path]}" is too long) if length($name) > 100 or length($prefix) > 155;

  my $header = pack $PACK_FORMAT, $name,    # 0
    sprintf('%06o ',  $self->mode),         # 1
    sprintf('%06o ',  $self->uid),          # 2
    sprintf('%06o ',  $self->gid),          # 3
    _to_number($self->size, 12),            # 4
    sprintf('%011o ', $self->mtime),        # 5
    '',                                     # 6 - checksum
    $self->type,                            # 7
    $self->symlink,                         # 8
    "ustar\0",                              # 9 - ustar
    '00',                                   # 10 - ustar version
    $self->owner,                           # 11
    $self->group,                           # 12
    sprintf('%07s', $self->dev_major),      # 13
    sprintf('%07s', $self->dev_minor),      # 14
    $prefix,                                # 15
    '';                                     # 16 - padding

  # Inject checksum
  substr $header, TAR_USTAR_CHECKSUM_POS, TAR_USTAR_CHECKSUM_LEN, $self->_checksum($header) . "\0 ";

  return $header;
}

sub _build_type ($self) {
  return '0' unless my $asset = $self->{asset};
  return '0' if -f $asset;                  # plain file
  return '1' if -l _;                       # symlink
  return '3' if -c _;                       # char dev
  return '4' if -b _;                       # block dev
  return '5' if -d _;                       # directory
  return '6' if -p _;                       # pipe
  return '8' if -s _;                       # socket
  return '2' if $asset->stat->nlink > 1;    # hard link
  return '9';                               # unknown
}

sub _checksum ($self, $header) {
  return sprintf '%06o', int unpack '%16C*', join '        ',
    substr($header, 0, TAR_USTAR_CHECKSUM_POS), substr($header, TAR_USTAR_TYPE_POS);
}

sub _cleanup ($self) {
  my $handle = delete $self->{add_block_handle};
  $handle->close if $handle;
}

sub _from_oct ($str) {
  no warnings 'portable';
  $str =~ s/^0+//;
  $str =~ s/[\s\0]+$//;
  return length($str) ? oct $str : 0;
}

sub _stat ($self, $field) {
  return undef unless my $stat = $self->{stat} //= $self->{asset} && $self->{asset}->stat || 0;
  return $stat->$field;
}

sub _trim_nul ($str) {
  my $idx = index $str, "\0";
  return $idx == -1 ? $str : substr $str, 0, $idx;
}

sub _from_number ($str) {
  # GNU/base-256 encoding.  The high bit of the first byte marks the
  # field as binary rather than ASCII octal.
  return _from_base256($str) if ord(substr($str, 0, 1)) & 0x80;
  return _from_oct($str);
}

sub _from_base256 ($str) {
  my @bytes = unpack 'C*', $str;

  # Clear the base-256 marker bit.  Mojo::Tar only needs non-negative
  # sizes here, so we intentionally reject negative values.
  croak 'Negative base-256 tar value is not supported'
    if $bytes[0] & 0x40;

  $bytes[0] &= 0x7f;

  my $value = 0;
  $value = ($value << 8) | $_ for @bytes;

  return $value;
}

sub _to_number ($value, $len) {
  # Classic tar numeric fields contain one fewer octal digit than the
  # field width because of the trailing NUL/space.
  my $max_oct = oct('7' x ($len - 1));

  return sprintf "%0*o ", $len - 1, $value
    if $value <= $max_oct;

  return _to_base256($value, $len);
}

sub _to_base256 ($value, $len) {
  croak 'Negative tar values are not supported' if $value < 0;

  my $str = '';

  for (1 .. $len) {
    $str = chr($value & 0xff) . $str;
    $value >>= 8;
  }

  croak 'Tar numeric value is too large'
    if $value;

  # Mark the field as base-256.
  substr($str, 0, 1, chr(ord(substr($str, 0, 1)) | 0x80));

  return $str;
}

sub DESTROY ($self) { $self->_cleanup }

1;

=encoding utf8

=head1 NAME

Mojo::Tar::File - A Mojo::Tar file

=head1 SYNOPSIS

  my $file = Mojo::Tar->new(path => 'some/file.txt');

  # This can be dangerous! Make sure path() does not contain ".."
  # or other dangerous path parts.
  $file->asset->move_to($file->path);

=head1 DESCRIPTION

L<Mojo::Asset::File> represents a tar file.

=head1 ATTRIBUTES

=head2 asset

  $file = $file->asset(Mojo::File->new);
  $asset = $file->asset;

Returns a L<Mojo::File> object. Defaults to L<Mojo::File/tempfile>.

This attribute is currently EXPERIMENTAL, but unlikely to change.

=head2 checksum

  $str = $file->checksum;

Holds the checksum read by L</from_header> or contains empty string if
the checksum does not match. This attribute can also be built from all the
attributes if L</from_header> was not called.

=head2 dev_major

This attribute is not supported yet. Pull request welcome!

=head2 dev_minor

This attribute is not supported yet. Pull request welcome!

=head2 gid

  $file = $file->gid(1001);
  $int = $file->gid;

The numeric representation of L</group>.

=head2 group

  $file = $file->group('users')
  $str = $file->group;

The string representation of L</gid>.

=head2 is_complete

  $bool = $file->is_complete;

Returns true if L</add_block> has added enough blocks to match L</size>.

=head2 mode

  $file = $file->mode(0644); # 0644 == 420
  $int = $file->mode;

The file mode. Note that this is 10-base, meaning C<$int> will be something
like "420" and not "644".

=head2 mtime

  $file = $file->mtime(time);
  $epoch = $file->mtime;

Epoch timestamp for this file.

=head2 owner

  $file = $file->owner('jhthorsen')
  $str = $file->owner;

The string representation of L</uid>.

=head2 path

  $file = $file->path('some/file/or/directory');
  $str = $file->path;

The path from the tar file. This is constructed with both the filename and
prefix (if any) in the ustar tar format.

=head2 size

  $file = $file->size(42);
  $int = $file->size;

The size of the file in bytes.

=head2 symlink

  $file = $file->symlink('path/for/symlink');
  $str = $file->symlink;

This attribute is not fully supported yet. Pull request welcome!

=head2 type

  $file = $file->type(5);
  $str = $file->type;

The tar file type.

This attribute is currently EXPERIMENTAL and might change from raw
representation to something more readable.

=head2 uid

  $file = $file->uid(1001);
  $int = $file->uid;

The numeric representation of L</owner>.

=head1 METHODS

=head2 add_block

  $file = $file->add_block($bytes);

Used to add a block from of bytes from the tar file to the L</asset>.

=head2 from_header

  $file = $file->from_header($bytes);

Will parse the header chunk from the tar file and set the L</ATTRIBUTES>.

=head2 to_header

  $bytes = $file->to_header;

Will construct a header chunk from the L</ATTRIBUTES>.

=head1 SEE ALSO

L<Mojo::Tar>.

=cut