This tutorial introduces you to the Arvados file storage system.
- *This tutorial assumes that you are "logged into an Arvados VM instance":{{site.basedoc}}/user/getting_started/ssh-access.html#login, and have a "working environment.":{{site.basedoc}}/user/getting_started/check-environment.html*
+ *This tutorial assumes that you are "logged into an Arvados VM instance":{{site.baseurl}}/user/getting_started/ssh-access.html#login, and have a "working environment.":{{site.baseurl}}/user/getting_started/check-environment.html*
The Arvados distributed file system is called *Keep*. Keep is a content-addressable file system. This means that files are managed using special unique identifiers derived from the _contents_ of the file, rather than human-assigned file names (specifically, the md5 hash). This has a number of advantages:
* Files can be stored and replicated across a cluster of servers without requiring a central name server.
The command @arv keep get@ fetches the contents of the locator @c1bad4b39ca5a924e481008009d94e32+210@. This is a locator for a collection data block, so it fetches the contents of the collection. In this example, this collection consists of a single file @var-GS000016015-ASM.tsv.bz2@ which is 227212247 bytes long, and is stored using four sequential data blocks, <code>204e43b8a1185621ca55a94839582e6f+67108864</code>, <code>b9677abbac956bd3e86b1deb28dfac03+67108864</code>, <code>fc15aff2a762b13f521baf042140acec+67108864</code>, <code>323d2a3ce20370c4ca1d3462a344f8fd+25885655</code>.
-Let's use @arv keep get@ to download the first datablock:
-
-notextile. <pre><code>/scratch/<b>you</b>$ <span class="userinput">arv keep get 204e43b8a1185621ca55a94839582e6f+67108864 > block1</span></code></pre>
-
-Let's look at the size and compute the md5 hash of @block1@:
-
-<notextile>
-<pre><code>/scratch/<b>you</b>$ <span class="userinput">ls -l block1</span>
--rw-r--r-- 1 you group 67108864 Dec 9 20:14 block1
-/scratch/<b>you</b>$ <span class="userinput">md5sum block1</span>
-204e43b8a1185621ca55a94839582e6f block1
-</code></pre>
-</notextile>
-
-Notice that the block identifer <code>204e43b8a1185621ca55a94839582e6f+67108864</code> of:
+Notice that the block identifer <code>204e43b8a1185621ca55a94839582e6f+67108864</code> consists of:
* the md5 hash @204e43b8a1185621ca55a94839582e6f@ which matches the md5 hash of @block1@
* a size hint @67108864@ which matches the size of @block1@
<pre><code>/scratch/<b>you</b>$ <span class="userinput">arv keep get c1bad4b39ca5a924e481008009d94e32+210/var-GS000016015-ASM.tsv.bz2 .</span>
</code></pre>
-This downloads the file @var-GS000016015-ASM.tsv.bz2@ described by collection @c1bad4b39ca5a924e481008009d94e32+210@ from Keep and places it into the local directory. Now that we have the file, we can compute the md5 hash of the complete file:
+This downloads the file <code>var-GS000016015-ASM.tsv.bz2</code> described by collection <code>c1bad4b39ca5a924e481008009d94e32+210</code> from Keep and places it into the local directory. Now that we have the file, we can compute the md5 hash of the complete file:
<notextile>
<pre><code>/scratch/<b>you</b>$ <span class="userinput">md5sum var-GS000016015-ASM.tsv.bz2</span>
<notextile>
<pre><code>/scratch/<b>you</b>$ <span class="userinput">arv keep ls c1bad4b39ca5a924e481008009d94e32+210</span>
-var-GS000016015-ASM.tsv.bz2
+./var-GS000016015-ASM.tsv.bz2
/scratch/<b>you</b>$ <span class="userinput">arv keep ls -s c1bad4b39ca5a924e481008009d94e32+210</span>
-221887 var-GS000016015-ASM.tsv.bz2
+ 221886 ./var-GS000016015-ASM.tsv.bz2
</code></pre>
</notextile>
use Fcntl qw(F_GETFL F_SETFL O_NONBLOCK);
use Arvados;
use Getopt::Long;
-use Warehouse;
-use Warehouse::Stream;
-use IPC::System::Simple qw(capturex);
+use IPC::Open2;
+use IO::Select;
$ENV{"TMPDIR"} ||= "/tmp";
unless (defined $ENV{"CRUNCH_TMP"}) {
}
$job_id = $Job->{'uuid'};
-$metastream = Warehouse::Stream->new(whc => new Warehouse);
-$metastream->clear;
-$metastream->name('.');
-$metastream->write_start($job_id . '.log.txt');
+# $metastream = Warehouse::Stream->new(whc => new Warehouse);
+# $metastream->clear;
+# $metastream->name('.');
+# $metastream->write_start($job_id . '.log.txt');
$Job->{'runtime_constraints'} ||= {};
if ($skip_install)
{
$ENV{"CRUNCH_SRC"} = $Job->{script_version};
- system("virtualenv", "$ENV{CRUNCH_TMP}/opt") == 0
- or croak ("virtualenv $ENV{CRUNCH_TMP}/opt failed: exit ".($?>>8));
- for my $src_path ("$ENV{CRUNCH_SRC}/arvados/sdk/python",
- "$ENV{CRUNCH_SRC}/sdk/python") {
+ for my $src_path ("$ENV{CRUNCH_SRC}/arvados/sdk/python") {
if (-d $src_path) {
- system ("cd $src_path && \$CRUNCH_TMP/opt/bin/python setup.py install")
+ system("virtualenv", "$ENV{CRUNCH_TMP}/opt") == 0
+ or croak ("virtualenv $ENV{CRUNCH_TMP}/opt failed: exit ".($?>>8));
+ system ("cd $src_path && ./build.sh && \$CRUNCH_TMP/opt/bin/python setup.py install")
== 0
or croak ("setup.py in $src_path failed: exit ".($?>>8));
}
if ($Job->{'output'})
{
eval {
- my $manifest_text = capturex("whget", $Job->{'output'});
+ my $manifest_text = `arv keep get $Job->{'output'}`;
$arv->{'collections'}->{'create'}->execute('collection' => {
'uuid' => $Job->{'output'},
'manifest_text' => $manifest_text,
} split ("\n", $jobstep[$job]->{stderr});
}
+sub fetch_block
+{
+ my $hash = shift;
+ my ($child_out, $child_in, $output_block);
+
+ my $pid = open2($child_out, $child_in, 'arv', 'keep', 'get', $hash);
+ sysread($child_out, $output_block, 64 * 1024 * 1024);
+ waitpid($pid, 0);
+ return $output_block;
+}
sub collate_output
{
- my $whc = Warehouse->new;
Log (undef, "collate");
- $whc->write_start (1);
+
+ my ($child_out, $child_in);
+ my $pid = open2($child_out, $child_in, 'arv', 'keep', 'put', '--raw');
my $joboutput;
for (@jobstep)
{
if ($output !~ /^[0-9a-f]{32}(\+\S+)*$/)
{
$output_in_keep ||= $output =~ / [0-9a-f]{32}\S*\+K/;
- $whc->write_data ($output);
+ print $child_in $output;
}
elsif (@jobstep == 1)
{
$joboutput = $output;
- $whc->write_finish;
+ last;
}
- elsif (defined (my $outblock = $whc->fetch_block ($output)))
+ elsif (defined (my $outblock = fetch_block ($output)))
{
$output_in_keep ||= $outblock =~ / [0-9a-f]{32}\S*\+K/;
- $whc->write_data ($outblock);
+ print $child_in $outblock;
}
else
{
- my $errstr = $whc->errstr;
- $whc->write_data ("XXX fetch_block($output) failed: $errstr XXX\n");
+ print $child_in "XXX fetch_block($output) failed XXX\n";
$main::success = 0;
}
}
- $joboutput = $whc->write_finish if !defined $joboutput;
+ if (!defined $joboutput) {
+ my $s = IO::Select->new($child_out);
+ sysread($child_out, $joboutput, 64 * 1024 * 1024) if $s->can_read(0);
+ }
+ $child_in->close;
+ waitpid($pid, 0);
+
if ($joboutput)
{
Log (undef, "output $joboutput");
}
print STDERR ((-t STDERR) ? ($datetime." ".$message) : $message);
- return if !$metastream;
- $metastream->write_data ($datetime . " " . $message);
+ # return if !$metastream;
+ # $metastream->write_data ($datetime . " " . $message);
}
sub save_meta
{
- my $justcheckpoint = shift; # false if this will be the last meta saved
- my $m = $metastream;
- $m = $m->copy if $justcheckpoint;
- $m->write_finish;
- my $whc = Warehouse->new;
- my $loglocator = $whc->store_block ($m->as_string);
- $arv->{'collections'}->{'create'}->execute('collection' => {
- 'uuid' => $loglocator,
- 'manifest_text' => $m->as_string,
- });
- undef $metastream if !$justcheckpoint; # otherwise Log() will try to use it
- Log (undef, "log manifest is $loglocator");
- $Job->{'log'} = $loglocator;
- $Job->update_attributes('log', $loglocator) if $job_has_uuid;
+# my $justcheckpoint = shift; # false if this will be the last meta saved
+# my $m = $metastream;
+# $m = $m->copy if $justcheckpoint;
+# $m->write_finish;
+# my $whc = Warehouse->new;
+# my $loglocator = $whc->store_block ($m->as_string);
+# $arv->{'collections'}->{'create'}->execute('collection' => {
+# 'uuid' => $loglocator,
+# 'manifest_text' => $m->as_string,
+# });
+# undef $metastream if !$justcheckpoint; # otherwise Log() will try to use it
+# Log (undef, "log manifest is $loglocator");
+# $Job->{'log'} = $loglocator;
+# $Job->update_attributes('log', $loglocator) if $job_has_uuid;
}
{
croak ("Thaw not implemented");
- my $whc;
- my $key = shift;
- Log (undef, "thaw from $key");
-
- @jobstep = ();
- @jobstep_done = ();
- @jobstep_todo = ();
- @jobstep_tomerge = ();
- $jobstep_tomerge_level = 0;
- my $frozenjob = {};
-
- my $stream = new Warehouse::Stream ( whc => $whc,
- hash => [split (",", $key)] );
- $stream->rewind;
- while (my $dataref = $stream->read_until (undef, "\n\n"))
- {
- if ($$dataref =~ /^job /)
- {
- foreach (split ("\n", $$dataref))
- {
- my ($k, $v) = split ("=", $_, 2);
- $frozenjob->{$k} = freezeunquote ($v);
- }
- next;
- }
-
- if ($$dataref =~ /^merge (\d+) (.*)/)
- {
- $jobstep_tomerge_level = $1;
- @jobstep_tomerge
- = map { freezeunquote ($_) } split ("\n", freezeunquote($2));
- next;
- }
-
- my $Jobstep = { };
- foreach (split ("\n", $$dataref))
- {
- my ($k, $v) = split ("=", $_, 2);
- $Jobstep->{$k} = freezeunquote ($v) if $k;
- }
- $Jobstep->{'failures'} = 0;
- push @jobstep, $Jobstep;
-
- if ($Jobstep->{exitcode} eq "0")
- {
- push @jobstep_done, $#jobstep;
- }
- else
- {
- push @jobstep_todo, $#jobstep;
- }
- }
-
- foreach (qw (script script_version script_parameters))
- {
- $Job->{$_} = $frozenjob->{$_};
- }
- $Job->save if $job_has_uuid;
+ # my $whc;
+ # my $key = shift;
+ # Log (undef, "thaw from $key");
+
+ # @jobstep = ();
+ # @jobstep_done = ();
+ # @jobstep_todo = ();
+ # @jobstep_tomerge = ();
+ # $jobstep_tomerge_level = 0;
+ # my $frozenjob = {};
+
+ # my $stream = new Warehouse::Stream ( whc => $whc,
+ # hash => [split (",", $key)] );
+ # $stream->rewind;
+ # while (my $dataref = $stream->read_until (undef, "\n\n"))
+ # {
+ # if ($$dataref =~ /^job /)
+ # {
+ # foreach (split ("\n", $$dataref))
+ # {
+ # my ($k, $v) = split ("=", $_, 2);
+ # $frozenjob->{$k} = freezeunquote ($v);
+ # }
+ # next;
+ # }
+
+ # if ($$dataref =~ /^merge (\d+) (.*)/)
+ # {
+ # $jobstep_tomerge_level = $1;
+ # @jobstep_tomerge
+ # = map { freezeunquote ($_) } split ("\n", freezeunquote($2));
+ # next;
+ # }
+
+ # my $Jobstep = { };
+ # foreach (split ("\n", $$dataref))
+ # {
+ # my ($k, $v) = split ("=", $_, 2);
+ # $Jobstep->{$k} = freezeunquote ($v) if $k;
+ # }
+ # $Jobstep->{'failures'} = 0;
+ # push @jobstep, $Jobstep;
+
+ # if ($Jobstep->{exitcode} eq "0")
+ # {
+ # push @jobstep_done, $#jobstep;
+ # }
+ # else
+ # {
+ # push @jobstep_todo, $#jobstep;
+ # }
+ # }
+
+ # foreach (qw (script script_version script_parameters))
+ # {
+ # $Job->{$_} = $frozenjob->{$_};
+ # }
+ # $Job->save if $job_has_uuid;
}
open STDERR, ">&STDOUT";
mkdir $destdir;
- open TARX, "|-", "tar", "-C", $destdir, "-xf", "-";
- print TARX <DATA>;
- if(!close(TARX)) {
- die "'tar -C $destdir -xf -' exited $?: $!";
+ my @git_archive_data = <DATA>;
+ if (@git_archive_data) {
+ open TARX, "|-", "tar", "-C", $destdir, "-xf", "-";
+ print TARX @git_archive_data;
+ if(!close(TARX)) {
+ die "'tar -C $destdir -xf -' exited $?: $!";
+ }
}
my $pwd;
my $install_dir = $ENV{"CRUNCH_INSTALL"} || "$pwd/opt";
mkdir $install_dir;
- shell_or_die ("virtualenv", $install_dir);
- for my $src_path ("$destdir/arvados/sdk/python", "$destdir/sdk/python") {
+ for my $src_path ("$destdir/arvados/sdk/python") {
if (-d $src_path) {
- shell_or_die ("cd $src_path && $install_dir/bin/python setup.py install");
+ shell_or_die ("virtualenv", $install_dir);
+ shell_or_die ("cd $src_path && ./build.sh && $install_dir/bin/python setup.py install");
}
}
projects / arvados.git / commitdiff