[ViewVC] Annotation of: cvs/Net-FCP/bin/fmd

#!/opt/bin/perl

=head1 fmd - the freenet mass downloader

Fmd is at a very early stage of development (and very hackish, too), as I
am learning the basics of freenet myself.

However, I use it in production, and since it verifies everything it
decodes etc., it seems to be quite safe to use, that is, if you know how
to debug perl :)

=head2 FEATURES

 - decoding is done "in place", i.e. all non-checkblock blocks are
   stored in-place and will not be moved, only checkblocks will
   be moved to their final position in the file before decoding.
 - high resistance against failures of fred or fmd
 - extremely persistent retry behaviour - there is no such thing
   as a permanent failure.
 - handles hundreds of simultaneous downloads with grace.

=head2 ENVIRONMENT

Set FMD_HOME to a directory (default ~/fmd) where fmd will store it's
files. It will not store your porn files or other freenet data outside
that directory.

The subdirectory db will contain a database (soon to go away), while tmp
contains queue files and partial splitfiles. All finished files will be
moved to the done subdir.

FREDHOST and FREDPORT do the obvious. Fix these docs if you disagree.

Also edit the fmd executable for the number of threads and other
non-useful constants. The default number (200) works for me, probably not
for you.

=head2 COMMANDS

=over 4

=item CHK@... (space or "/") filename

Just pasting a CHK and a filename sperated by one or more spaces or
slashes will add files to the queue. The line can have leading garbage,
i.e. you can paste full uris.

=item <attachment>...</attachment>

=item <attach>...</attach>

=item <attached>...</attached>

Will add frosts mentally deranged pseudo-xml format, soon to be replaced
by something even more horrible.

=item l

List all jobs by number.

=item <number> (optionally trailing command)

Selects the job with the given number for further comamnds that require a current job.

=item s

Show the current job. Bot useful right now.

=item k

Kill the current job. This will keep the temp. file around. Sorry. (But it
could be used to reconstruct the download.. hmm..)

=item pri<number>

Set the current job priority to <pri>. The default is 1. A job with higher
priority will (on average) get more requests.

The useful range is probably 0..100, but be careful and limit yourself to
small values (<10), otherwise your other downloads might starve!

=item q

Kill the command prompt. Yupp.

=back

=cut

use strict;

use Net::FCP;
use Storable;
use Time::HiRes;
use Coro;
use Coro::Event;
use Coro::Handle;
use Coro::Signal;
use Coro::Timer;
use Coro::AIO;
use List::Util;
use Digest::SHA1;
use Algorithm::FEC;
use Digest::SHA1;
use Net::FCP::Util;
use POSIX ();

Coro::AIO::max_poll_time 0.01;
Coro::AIO::min_parallel 1;
Coro::AIO::max_parallel 2;

$|=1;

our $MAX_TXN = 150; # use max. this many transactions in parallel
our @HTL = (5,12,20);
our $VERIFY_CHK = 1; # verify all blocks, again and again (only useful to debugging).
our $FMD_HOME = $ENV{FMD_HOME} || "$ENV{HOME}/fmd";

our $FCP = new Net::FCP;

defined $FMD_HOME
   or die "you currently must define FMD_HOME to a persistent directory";

mkdir $FMD_HOME, 0700;

our $QUEUE_HOME = "$FMD_HOME/tmp";
mkdir $QUEUE_HOME, 0700;
our $DONE_HOME = "$FMD_HOME/done";
mkdir $DONE_HOME, 0700;

our %job;

sub push_key {
   my ($key, $title) = @_;

   for my $job (values %job) {
      if ($job->{p}{key} eq $key) {
         warn "job $job->{id} already works on this, not adding";
         return;
      }
   }

   my $job = job->new_from_key ($key, $title);
   warn "added as job $job->{id}\n";
   $job;
}

sub cmdline {
   my ($i, $o) = @_;

   my $cmd = async {
      my $job;
      while (print $o "> " and defined ($_ = <$i>)) {
         chomp;
         if (/<attach(?:ment|ed|)>(.*) \* (CHK[^<]+)<\/attach/) {
            $job = push_key $2, $1;
         } elsif (/(CHK\@[a-zA-Z0-9,~\-]{54})[\/ ]+(.*?)\s*$/) {
            $job = push_key $1, $2;
         } elsif (s/^(\d+)//) {
            if ($job = $job{$1}) {
               print $o delete $job->{log};
               if ($job->{input}) {
               }
            }
            redo;
         } elsif (/^q/) {
            close $i;
            close $o;
         } elsif (/^l/) {
            for my $job (sort { $a->{id} <=> $b->{id} } values %job) {
               print $o "$_ $job->{id}: $job->{p}{key} $job->{p}{title} $job->{status}\n";
            }
         } elsif (/^pri\s*(\d+)/) {
            if ($job) {
               $job->{p}{pri} = $1;
               $job->save;
            }
         } elsif (/^s$/) {
            print $o $job->show if $job;
         } elsif (/^k$/) {
            $job->kill if $job;
         } elsif (/\S/) {
            #
         } else {
            print $o "?\n";
         }
         for my $job (sort { $a->{id} <=> $b->{id} } values %job) {
            if ($job->{_input}) {
                print $o "> $job->{id} $job->{title} $job->{status}\n";
            }
         }
      }
   };
}

package job;

use strict;

use Coro;
use Coro::AIO;
use Fcntl;
use IO::Handle;
use Array::Heap2;

my $count = 0;

sub new {
   my $class = shift;

   my $self = bless { @_ }, $class;

   $self->{p}{title} !~ /\//
      or die "$self->{p}{title}: malformed key\n";

   $self->{id} = ++$count;
   $self->{job} ||= "$QUEUE_HOME/" . Time::HiRes::time . ":$count.j";

   $job{$self->{id}} = $self;

   $self->save;
   $self->start;
   $self;
}

sub new_from_key {
   my ($class, $key, $title) = @_;
   $class->new (p => { key => $key, title => $title, state => "examine" });
}

sub new_from_file {
   my ($class, $path) = @_;
   $class->new (job => $path, p => Storable::retrieve $path);
}

sub save {
   my ($self) = @_;
   Storable::nstore $self->{p}, "$self->{job}~";
   aio_rename "$self->{job}~", $self->{job};
}

sub clean {
   my ($self) = @_;

   delete $job{$self->{id}};
   $self->save;
   aio_rename $self->{job}, "$DONE_HOME/$self->{p}{title}.job";
   aio_unlink $self->{job};
}

sub kill {
   my ($self) = @_;

   $self->clean;
   $self->{coro}->cancel;
}

my @queue;
my $queue_change = new Coro::Signal;
my $queue_alloc = 0;

async {
   for (;;) {
      while (@queue
             and (($queue[0][0] > 10 and $queue_alloc < $MAX_TXN)
               or ($queue[0][0] >  1 and $queue_alloc < $MAX_TXN - 3)
               or $queue_alloc < $MAX_TXN - 5)) {
         (pop_heap @queue)->[1]->send;
         $queue_alloc++;
         Coro::Timer::sleep 0.05;
      }
      $queue_change->wait;
   }
};

sub txn_begin {
   my ($pri) = @_;
   my $sig = new Coro::Signal;

   #warn "txn_begin $pri\n";#d#
   push_heap @queue, [$pri, $sig];
   $queue_change->send;
   $sig->wait;
}

sub txn_end {
   $queue_alloc--;
   $queue_change->send;
}

sub fetch_uri {
   my ($pri, $uri) = @_;

   for(my $count = 1; ; $count += 0.3) {
      for my $htl (@HTL) {
         txn_begin time + $htl + $count;
         my $sig = new Coro::Signal;
         my ($meta, $data) = eval { @{ $FCP->client_get ($uri, $htl) } };
         txn_end;
         if ($@) {
            if (UNIVERSAL::isa ($@, Net::FCP::Exception::)) {
               if ($@->type ("data_not_found")
                   || $@->type ("route_not_found")) {
                  next;
               }
               if ($@->type ("short_data")) {
                  warn "(short_data, redo)\n";
                  redo;
               }
               die;
            }
         }
         if (defined $data) {
            return ($meta, $data);
         }
      }
   }

   die;
}

sub log {
   my ($self, $text) = @_;
   my $time = POSIX::strftime "%H:%M:%S", localtime time;
   warn "$time $self->{id},$self->{p}{pri}: $text\n";
   $self->{log} .= "$time $text\n";
}

sub feedback {
   my ($self, $prompt) = @_;
   $self->{input} = [$Coro::current, $prompt];
   Coro::schedule;
}

sub show {
   my ($self) = @_;

   "ID: $self->{id}\n"
      . "Title: $self->{p}{title}\n"
      . "Blocks#: " . @{$self->{p}{blk}} . "\n"
      . "Blocks: " . (join "", map {
         $_->{done} ? "+" : "-"
      } @{$self->{p}{blk}}) . "\n" .
   ""
}

my $id;

sub MAXSEG (){ 128*1024*1024 }
sub MINSEG (){   6* 128*1024 }

sub blocksize($) {
   return
        $_[0] >= 64*1024*1024 ? 1024*1024
      : $_[0] >= 32*1024*1024 ?  512*1024
      : $_[0] >=    1024*1024 ?  256*1024
                              :  128*1024;
}

sub start {
   my ($self) = @_;

   $self->{p}{pri} ||= 1;
   $self->{p}{state} ||= "examine";

   $self->{job} =~ /^(.*)\.j$/ or die "$self->{job}: missing .j";
   $self->{file} = "$1.d";
   $self->{fh} = aio_open $self->{file}, O_RDWR|O_CREAT, 0600
      or die "$self->{file}: $!";

   $self->{status} = "starting";
   $self->{coro} = async {
      for(;;) {
         my ($state, @args) = ref $self->{p}{state} ? @{$self->{p}{state}} : $self->{p}{state};
         my $next = eval { $self->can ("state_$state")->($self, @args) };
         if ($@) {
            $self->log ($@);
            $next = $self->feedback ("continue with state: ");
         }
         $self->log ($self->{status} = "STATE CHANGE: ". join ", ", ref $next ? @$next : $next);
         $self->{p}{state} = $next;
         $self->save;
      }
   };
}

sub state_finish {
   my ($self, $save) = @_;

   if ($save) {
      aio_fsync $self->{fh};
      close $self->{fh};

      aio_unlink "$DONE_HOME/$self->{p}{title}";
      aio_link $self->{file}, "$DONE_HOME/$self->{p}{title}"
         and die "link: $self->{file} => $DONE_HOME/$self->{p}{title}: $!";
      system "sync";
   }
   $self->clean;

   aio_unlink $self->{file};

   $self->{status} = "finished";
   $self->feedback ("finished");
   terminate;
}

sub state_examine {
   my ($self) = @_;
   my $p = $self->{p};

   $self->{status} = "initial fetch";

   for (;;) {
      $self->log ("fetching $p->{key} (=$p->{title})");

      ($p->{meta}, $p->{data}) = fetch_uri 100, "freenet:$p->{key}";
      $self->save;
      #use PApp::Util; print STDERR PApp::Util::dumpval [keys %{$meta->{document}[0]{split_file}}];
      $self->log ("type $p->{meta}{document}[0]{info}{format}");

      if (my $splitfile = $p->{meta}{document}[0]{split_file}) {
         return "splitfile";
      } elsif ((defined $p->{data}) and (length $p->{data})) {
         syswrite $self->{fh}, delete $p->{data};
         return ["finish", 1];
      }

      $self->log ("EMPTY, retrying in an hour");
      Coro::Timer::sleep 3600;
   }

}

sub state_splitfile {
   my ($self) = @_;
   my $p = $self->{p};

   my $splitfile = $p->{meta}{document}[0]{split_file};
   my $filesize = hex $splitfile->{size};

   if ($splitfile->{algo_name} eq "OnionFEC_a_1_2") {
      my $data_packets = hex $splitfile->{block_count};
      my $check_packets = hex $splitfile->{check_block_count};

      my $blk = ($p->{blk} ||= []);

      unless (@$blk) {
         for (1..$data_packets) {
            push @$blk, {
               uri => $splitfile->{block}{sprintf "%x", $_},
            };
         }
         for (1..$check_packets) {
            push @$blk, {
               uri => $splitfile->{check_block}{sprintf "%x", $_},
            };
         }
      }

      my @segments;
      my $segments = 0;

      {
         # that is a horrible algorithm :(, these freenet freaks are... java-disabled
         # hardcoding lots of magic parameters is soo dumb.
         my $size = $filesize;
         my $offset = 0;
         my $offset2 = ($filesize & ~(1024*1024-1)) + 1024*1024; # leave enough space after last data block
         my $idx  = 0;
         my $idx2 = $data_packets;
         my @redundandy = (0,1,2); # maybe OnionFAC_a_1_2 means 1/2 redundancy(?)

         while ($size > 0) {
            my $segsize = $size >= MAXSEG ? MAXSEG : $size <= MINSEG ? MINSEG : $size;
            my $blksize = blocksize $segsize;
            my $seg    = 
               {
                  id      => $segments++,
                  todo    => int (($segsize + $blksize - 1) / $blksize),
                  done    => 0,
                  blk     => [],
                  blksize => $blksize,
               };

            push @segments, $seg;
            $size -= $segsize;

            while ($segsize > 0) {
               push @{$seg->{blk}}, $idx;
               for ($blk->[$idx++]) {
                  $_->{offset} = $offset;
                  #$_->{size}   = $blksize > $segsize ? $segsize : $blksize; # WRONG
                  $_->{size}   = $blksize;
                  $_->{seg}    = $seg;
               }

               $segsize -= $blksize;
               $offset  += $blksize;

               if (($redundandy[0] += $redundandy[1]) >= $redundandy[2]) {
                  $redundandy[0] -= $redundandy[2];

                  push @{$seg->{blk}}, $idx2;
                  for ($blk->[$idx2++]) {
                     $_->{offset} = $offset2;
                     $_->{size}   = $blksize;
                     $_->{seg}    = $seg;
                  }

                  $offset2 += $blksize;
               }
            }
         }

         $idx == $data_packets
            or die "$self->{id}/$p->{tile} $self->{job}\nidx $idx != data_packets $data_packets";
         $idx2 == $data_packets + $check_packets
            or die "$self->{id}/$p->{tile} $self->{job}\nidx2 $idx2 != data_packets $data_packets + check_packets $check_packets";
      }

      for (@$blk) {
         ++$_->{seg}{done} if $_->{done};
         delete $_->{htl};
      }

      my $fail = 0;
      my $sig = new Coro::Signal;

      $self->{status} = "splitfile fetch (" .  @$blk . " blocks)";

      my @txn;

      for (;;) {
         for my $id (0 .. $#$blk) {
            my $blk = $blk->[$id];

            next if $txn[$id] || $blk->{done} || $blk->{seg}{todo} <= $blk->{seg}{done};

            my $htl = $HTL[$blk->{htl}++ % @HTL];
            my $pri = int time + $htl / 20 * 7200 * rand;

            txn_begin $pri;
            #warn $self->{id} . ", GET<$htl, $pri>\n";#d#
            $txn[$id] ||= $FCP->txn_client_get ($blk->{uri}, $htl)->cb(sub {
               undef $txn[$id];

               my $seg = $blk->{seg};

               my ($meta, $data) = eval { @{ $_[0]->result } };

               if (defined $data) {
                  $blk->{size} == length $data
                     or die sprintf "block $id expected size %d, got %d\n", $blk->{size}, length $data;

                  (length $data) == (aio_write $self->{fh}, $blk->{offset}, (length $data), $data, 0)
                     or die "unable to write chunk to disk, not setting valid flag";
                  aio_fsync $self->{fh};

                  $blk->{done} = 1;
                  $blk->{meta} = $meta->{raw} if length $meta->{raw};#d#
                  $seg->{done}++;
                  $self->save;

                  $::htl_sum += $htl;
                  $::htl_cnt++;

                  $self->log (sprintf "got block $seg->{id}.$id %d ($seg->{done}/$seg->{todo}) at htl $htl (%f) and pri $pri",
                                       length $data, $::htl_sum / $::htl_cnt);
               } else {
                  if ($@) {
                     if ($@->type ("data_not_found")) {
                        # nop
                     } elsif ($@->type ("network_error")) {
                        $self->log ("$@, retrying in 1s");
                        CORE::sleep 1;
                     } else {
                        $self->log ("$@");
                     }
                  }
                  ++$fail;
               }
               $self->{status} = "splitfile fetch ($seg->{done}/$seg->{todo}, $fail failed)";

               txn_end;
               $sig->send;
            });
         }

         for my $seg (@segments) {
            if ($seg->{done} >= $seg->{todo} && !$seg->{finished}) {

               $self->log ("segment done, cancelling segment $seg->{id}");
               for my $id (@{$seg->{blk}}) {
                  (delete $txn[$id])->cancel if $txn[$id];
               }

               my $verify;
               for my $id (@{$seg->{blk}}) {
                  my $blk = $blk->[$id];
                  if ($blk->{done}) {
                     aio_read $self->{fh}, $blk->{offset}, $blk->{size}, my $buf, 0;

                     my $k1 = Net::FCP::Util::extract_chk_hash $blk->{uri};
                     my $k2 = Net::FCP::Util::generate_chk_hash $blk->{meta}, $buf;

                     if ($k1 ne $k2) {
                        $verify .= "v";
                        #warn sprintf "$p->{title} block $id BROKEN (%s != %s)", (unpack "H*", $k1), (unpack "H*", $k2);
                        $blk->{done} = 0;
                        $seg->{done}--;
                        $self->save;
                     } else {
                        $verify .= "V";
                     }
                  }
               }
               $self->log ("verified segment $seg->{id}");
               $self->log ($verify);

               if ($seg->{done} >= $seg->{todo} && !$seg->{finished}) {
                  $self->log ("verified segment OK $seg->{id}");
                  $seg->{finished}++;
                  $segments--;
               } else {
                  $self->log ("verified segment NOT OK $seg->{id}");
               }
            }
         }

         last unless $segments;

         $sig->wait;
      }

      $self->log ("decoding < $self->{job} $self->{file} $filesize >");

      for my $seg (@segments) {
         my @part;
         my @idx;
         my @blk = map $blk->[$_], sort { $a <=> $b } @{$seg->{blk}};

         for my $id (0 .. $#blk) {
            my $blk = $blk[$id];
            next unless $blk->{done};

            push @part, [$self->{fh}, $blk->{offset}];
            push @idx, $id;

            last if @idx == $seg->{todo};
         }

         my $fec = new Algorithm::FEC
                          $seg->{todo},
                          scalar @blk,
                          $seg->{blksize};

         $fec->shuffle (\@part, \@idx);

         # now copy check blocks to their destination position
         for my $i (0 .. $#idx) {
            next if $idx[$i] == $i;

            my $src = $part[$i];
            $part[$i] = [$self->{fh}, $blk[$i]{offset}];
            $fec->copy ($src, $part[$i]);
         }

         $fec->set_decode_blocks (\@part, \@idx);
         $fec->decode;
      }

      my $sha1 = new Digest::SHA1;
      open my $dd, "-|", "head -c$filesize \Q$self->{file}\E"
         or die "DD: $!";
      #$dd = Coro::Handle::unblock $dd;
      $sha1->addfile ($dd);
      $sha1 = $sha1->hexdigest;

      if (exists $p->{meta}{document}[0]{info}{checksum}
          and $p->{meta}{document}[0]{info}{checksum} ne $sha1) {
         $self->log ("META: $p->{meta}{document}[0]{info}{checksum} and real checksum $sha1 for $filesize DIFFER");
#         $self->feedback ("CHECKSUM ERROR");
#         terminate;
      }

      truncate $self->{fh}, $filesize;
      sysseek $self->{fh}, 0, 0;

      return ["finish", 1];
   } else {
      $self->log ("splitfile algo '$splitfile->{algo_name}' unknown");
      $self->feedback ("algo unknown");
      terminate;
   }
}

package main;

$|=1;

for (<\Q$QUEUE_HOME\E/*.j>) {
   job->new_from_file ($_);
   print "J";
}
print "\n";

open my $stdin , "<&0" or die;
open my $stdout, ">&1" or die;
cmdline unblock $stdin, unblock $stdout;

&Coro::Event::loop;

Revision:	1.11
Committed:	Mon Nov 20 20:20:51 2006 UTC (19 years, 9 months ago) by root
Branch:	MAIN
Changes since 1.10:	+1 -1 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	#!/opt/bin/perl
2
3			=head1 fmd - the freenet mass downloader
4
5			Fmd is at a very early stage of development (and very hackish, too), as I
6			am learning the basics of freenet myself.
7
8			However, I use it in production, and since it verifies everything it
9			decodes etc., it seems to be quite safe to use, that is, if you know how
10			to debug perl :)
11
12			=head2 FEATURES
13
14			- decoding is done "in place", i.e. all non-checkblock blocks are
15			stored in-place and will not be moved, only checkblocks will
16			be moved to their final position in the file before decoding.
17			- high resistance against failures of fred or fmd
18			- extremely persistent retry behaviour - there is no such thing
19			as a permanent failure.
20			- handles hundreds of simultaneous downloads with grace.
21
22			=head2 ENVIRONMENT
23
24			Set FMD_HOME to a directory (default ~/fmd) where fmd will store it's
25			files. It will not store your porn files or other freenet data outside
26			that directory.
27
28			The subdirectory db will contain a database (soon to go away), while tmp
29			contains queue files and partial splitfiles. All finished files will be
30			moved to the done subdir.
31
32			FREDHOST and FREDPORT do the obvious. Fix these docs if you disagree.
33
34			Also edit the fmd executable for the number of threads and other
35			non-useful constants. The default number (200) works for me, probably not
36			for you.
37
38			=head2 COMMANDS
39
40			=over 4
41
42			=item CHK@... (space or "/") filename
43
44			Just pasting a CHK and a filename sperated by one or more spaces or
45			slashes will add files to the queue. The line can have leading garbage,
46			i.e. you can paste full uris.
47
48			=item <attachment>...</attachment>
49
50			=item <attach>...</attach>
51
52			=item <attached>...</attached>
53
54			Will add frosts mentally deranged pseudo-xml format, soon to be replaced
55			by something even more horrible.
56
57			=item l
58
59			List all jobs by number.
60
61			=item <number> (optionally trailing command)
62
63			Selects the job with the given number for further comamnds that require a current job.
64
65			=item s
66
67			Show the current job. Bot useful right now.
68
69			=item k
70
71			Kill the current job. This will keep the temp. file around. Sorry. (But it
72			could be used to reconstruct the download.. hmm..)
73
74			=item pri<number>
75
76	root	1.2	Set the current job priority to <pri>. The default is 1. A job with higher
77			priority will (on average) get more requests.
78	root	1.1
79	root	1.2	The useful range is probably 0..100, but be careful and limit yourself to
80			small values (<10), otherwise your other downloads might starve!
81	root	1.1
82			=item q
83
84			Kill the command prompt. Yupp.
85
86			=back
87
88			=cut
89
90	root	1.10	use strict;
91
92	root	1.1	use Net::FCP;
93			use Storable;
94			use Time::HiRes;
95			use Coro;
96			use Coro::Event;
97			use Coro::Handle;
98			use Coro::Signal;
99			use Coro::Timer;
100	root	1.9	use Coro::AIO;
101	root	1.1	use List::Util;
102			use Digest::SHA1;
103			use Algorithm::FEC;
104			use Digest::SHA1;
105			use Net::FCP::Util;
106			use POSIX ();
107
108	root	1.9	Coro::AIO::max_poll_time 0.01;
109			Coro::AIO::min_parallel 1;
110			Coro::AIO::max_parallel 2;
111
112	root	1.1	$\|=1;
113
114	root	1.9	our $MAX_TXN = 150; # use max. this many transactions in parallel
115			our @HTL = (5,12,20);
116	root	1.1	our $VERIFY_CHK = 1; # verify all blocks, again and again (only useful to debugging).
117			our $FMD_HOME = $ENV{FMD_HOME} \|\| "$ENV{HOME}/fmd";
118
119			our $FCP = new Net::FCP;
120
121			defined $FMD_HOME
122			or die "you currently must define FMD_HOME to a persistent directory";
123
124			mkdir $FMD_HOME, 0700;
125
126			our $QUEUE_HOME = "$FMD_HOME/tmp";
127			mkdir $QUEUE_HOME, 0700;
128			our $DONE_HOME = "$FMD_HOME/done";
129			mkdir $DONE_HOME, 0700;
130
131			our %job;
132
133			sub push_key {
134			my ($key, $title) = @_;
135
136			for my $job (values %job) {
137			if ($job->{p}{key} eq $key) {
138			warn "job $job->{id} already works on this, not adding";
139			return;
140			}
141			}
142
143			my $job = job->new_from_key ($key, $title);
144			warn "added as job $job->{id}\n";
145			$job;
146			}
147
148			sub cmdline {
149			my ($i, $o) = @_;
150
151			my $cmd = async {
152			my $job;
153			while (print $o "> " and defined ($_ = <$i>)) {
154			chomp;
155			if (/<attach(?:ment\|ed\|)>(.) \ (CHK[^<]+)<\/attach/) {
156			$job = push_key $2, $1;
157	root	1.7	} elsif (/(CHK\@[a-zA-Z0-9,~\-]{54})[\/ ]+(.?)\s$/) {
158	root	1.1	$job = push_key $1, $2;
159			} elsif (s/^(\d+)//) {
160			if ($job = $job{$1}) {
161			print $o delete $job->{log};
162			if ($job->{input}) {
163			}
164			}
165			redo;
166			} elsif (/^q/) {
167			close $i;
168			close $o;
169			} elsif (/^l/) {
170			for my $job (sort { $a->{id} <=> $b->{id} } values %job) {
171			print $o "$_ $job->{id}: $job->{p}{key} $job->{p}{title} $job->{status}\n";
172			}
173			} elsif (/^pri\s*(\d+)/) {
174			if ($job) {
175			$job->{p}{pri} = $1;
176			$job->save;
177			}
178			} elsif (/^s$/) {
179			print $o $job->show if $job;
180			} elsif (/^k$/) {
181			$job->kill if $job;
182			} elsif (/\S/) {
183			#
184			} else {
185			print $o "?\n";
186			}
187			for my $job (sort { $a->{id} <=> $b->{id} } values %job) {
188			if ($job->{_input}) {
189			print $o "> $job->{id} $job->{title} $job->{status}\n";
190			}
191			}
192			}
193			};
194			}
195
196			package job;
197
198	root	1.10	use strict;
199
200	root	1.1	use Coro;
201	root	1.9	use Coro::AIO;
202	root	1.1	use Fcntl;
203			use IO::Handle;
204	root	1.6	use Array::Heap2;
205	root	1.1
206			my $count = 0;
207
208			sub new {
209			my $class = shift;
210
211			my $self = bless { @_ }, $class;
212
213	root	1.9	$self->{p}{title} !~ /\//
214			or die "$self->{p}{title}: malformed key\n";
215
216	root	1.1	$self->{id} = ++$count;
217			$self->{job} \|\|= "$QUEUE_HOME/" . Time::HiRes::time . ":$count.j";
218
219			$job{$self->{id}} = $self;
220
221			$self->save;
222			$self->start;
223			$self;
224			}
225
226			sub new_from_key {
227			my ($class, $key, $title) = @_;
228			$class->new (p => { key => $key, title => $title, state => "examine" });
229			}
230
231			sub new_from_file {
232			my ($class, $path) = @_;
233			$class->new (job => $path, p => Storable::retrieve $path);
234			}
235
236			sub save {
237			my ($self) = @_;
238			Storable::nstore $self->{p}, "$self->{job}~";
239	root	1.9	aio_rename "$self->{job}~", $self->{job};
240	root	1.1	}
241
242			sub clean {
243			my ($self) = @_;
244
245			delete $job{$self->{id}};
246			$self->save;
247	root	1.9	aio_rename $self->{job}, "$DONE_HOME/$self->{p}{title}.job";
248			aio_unlink $self->{job};
249	root	1.1	}
250
251			sub kill {
252			my ($self) = @_;
253
254			$self->clean;
255			$self->{coro}->cancel;
256			}
257
258			my @queue;
259			my $queue_change = new Coro::Signal;
260			my $queue_alloc = 0;
261
262			async {
263			for (;;) {
264			while (@queue
265			and (($queue[0][0] > 10 and $queue_alloc < $MAX_TXN)
266			or ($queue[0][0] > 1 and $queue_alloc < $MAX_TXN - 3)
267			or $queue_alloc < $MAX_TXN - 5)) {
268			(pop_heap @queue)->[1]->send;
269			$queue_alloc++;
270	root	1.4	Coro::Timer::sleep 0.05;
271	root	1.1	}
272			$queue_change->wait;
273			}
274			};
275
276			sub txn_begin {
277			my ($pri) = @_;
278			my $sig = new Coro::Signal;
279
280			#warn "txn_begin $pri\n";#d#
281			push_heap @queue, [$pri, $sig];
282			$queue_change->send;
283			$sig->wait;
284			}
285
286			sub txn_end {
287			$queue_alloc--;
288			$queue_change->send;
289			}
290
291			sub fetch_uri {
292			my ($pri, $uri) = @_;
293
294			for(my $count = 1; ; $count += 0.3) {
295			for my $htl (@HTL) {
296	root	1.4	txn_begin time + $htl + $count;
297	root	1.1	my $sig = new Coro::Signal;
298			my ($meta, $data) = eval { @{ $FCP->client_get ($uri, $htl) } };
299			txn_end;
300			if ($@) {
301			if (UNIVERSAL::isa ($@, Net::FCP::Exception::)) {
302			if ($@->type ("data_not_found")
303			\|\| $@->type ("route_not_found")) {
304			next;
305			}
306			if ($@->type ("short_data")) {
307			warn "(short_data, redo)\n";
308			redo;
309			}
310			die;
311			}
312			}
313			if (defined $data) {
314			return ($meta, $data);
315			}
316			}
317			}
318
319			die;
320			}
321
322			sub log {
323			my ($self, $text) = @_;
324			my $time = POSIX::strftime "%H:%M:%S", localtime time;
325	root	1.3	warn "$time $self->{id},$self->{p}{pri}: $text\n";
326	root	1.1	$self->{log} .= "$time $text\n";
327			}
328
329			sub feedback {
330			my ($self, $prompt) = @_;
331			$self->{input} = [$Coro::current, $prompt];
332			Coro::schedule;
333			}
334
335			sub show {
336			my ($self) = @_;
337
338			"ID: $self->{id}\n"
339			. "Title: $self->{p}{title}\n"
340			. "Blocks#: " . @{$self->{p}{blk}} . "\n"
341			. "Blocks: " . (join "", map {
342			$_->{done} ? "+" : "-"
343			} @{$self->{p}{blk}}) . "\n" .
344	root	1.11	""
345	root	1.1	}
346
347			my $id;
348
349			sub MAXSEG (){ 12810241024 }
350			sub MINSEG (){ 6* 128*1024 }
351
352			sub blocksize($) {
353			return
354			$_[0] >= 6410241024 ? 1024*1024
355			: $_[0] >= 3210241024 ? 512*1024
356			: $_[0] >= 10241024 ? 2561024
357			: 128*1024;
358			}
359
360			sub start {
361			my ($self) = @_;
362
363			$self->{p}{pri} \|\|= 1;
364			$self->{p}{state} \|\|= "examine";
365
366	root	1.9	$self->{job} =~ /^(.*)\.j$/ or die "$self->{job}: missing .j";
367			$self->{file} = "$1.d";
368			$self->{fh} = aio_open $self->{file}, O_RDWR\|O_CREAT, 0600
369	root	1.1	or die "$self->{file}: $!";
370
371			$self->{status} = "starting";
372			$self->{coro} = async {
373			for(;;) {
374			my ($state, @args) = ref $self->{p}{state} ? @{$self->{p}{state}} : $self->{p}{state};
375			my $next = eval { $self->can ("state_$state")->($self, @args) };
376			if ($@) {
377			$self->log ($@);
378			$next = $self->feedback ("continue with state: ");
379			}
380	root	1.10	$self->log ($self->{status} = "STATE CHANGE: ". join ", ", ref $next ? @$next : $next);
381	root	1.1	$self->{p}{state} = $next;
382			$self->save;
383			}
384			};
385			}
386
387			sub state_finish {
388			my ($self, $save) = @_;
389
390			if ($save) {
391	root	1.10	aio_fsync $self->{fh};
392	root	1.1	close $self->{fh};
393
394	root	1.10	aio_unlink "$DONE_HOME/$self->{p}{title}";
395			aio_link $self->{file}, "$DONE_HOME/$self->{p}{title}"
396			and die "link: $self->{file} => $DONE_HOME/$self->{p}{title}: $!";
397	root	1.1	system "sync";
398			}
399			$self->clean;
400
401	root	1.10	aio_unlink $self->{file};
402	root	1.1
403			$self->{status} = "finished";
404			$self->feedback ("finished");
405			terminate;
406			}
407
408			sub state_examine {
409			my ($self) = @_;
410			my $p = $self->{p};
411
412			$self->{status} = "initial fetch";
413
414			for (;;) {
415			$self->log ("fetching $p->{key} (=$p->{title})");
416
417			($p->{meta}, $p->{data}) = fetch_uri 100, "freenet:$p->{key}";
418			$self->save;
419			#use PApp::Util; print STDERR PApp::Util::dumpval [keys %{$meta->{document}[0]{split_file}}];
420			$self->log ("type $p->{meta}{document}[0]{info}{format}");
421
422			if (my $splitfile = $p->{meta}{document}[0]{split_file}) {
423			return "splitfile";
424			} elsif ((defined $p->{data}) and (length $p->{data})) {
425	root	1.8	syswrite $self->{fh}, delete $p->{data};
426	root	1.1	return ["finish", 1];
427			}
428
429			$self->log ("EMPTY, retrying in an hour");
430			Coro::Timer::sleep 3600;
431			}
432
433			}
434
435			sub state_splitfile {
436			my ($self) = @_;
437			my $p = $self->{p};
438
439			my $splitfile = $p->{meta}{document}[0]{split_file};
440			my $filesize = hex $splitfile->{size};
441
442			if ($splitfile->{algo_name} eq "OnionFEC_a_1_2") {
443			my $data_packets = hex $splitfile->{block_count};
444			my $check_packets = hex $splitfile->{check_block_count};
445
446			my $blk = ($p->{blk} \|\|= []);
447
448			unless (@$blk) {
449			for (1..$data_packets) {
450			push @$blk, {
451			uri => $splitfile->{block}{sprintf "%x", $_},
452			};
453			}
454			for (1..$check_packets) {
455			push @$blk, {
456			uri => $splitfile->{check_block}{sprintf "%x", $_},
457			};
458			}
459			}
460
461			my @segments;
462			my $segments = 0;
463
464			{
465			# that is a horrible algorithm :(, these freenet freaks are... java-disabled
466			# hardcoding lots of magic parameters is soo dumb.
467			my $size = $filesize;
468			my $offset = 0;
469			my $offset2 = ($filesize & ~(10241024-1)) + 10241024; # leave enough space after last data block
470			my $idx = 0;
471			my $idx2 = $data_packets;
472			my @redundandy = (0,1,2); # maybe OnionFAC_a_1_2 means 1/2 redundancy(?)
473
474			while ($size > 0) {
475			my $segsize = $size >= MAXSEG ? MAXSEG : $size <= MINSEG ? MINSEG : $size;
476			my $blksize = blocksize $segsize;
477			my $seg =
478			{
479			id => $segments++,
480			todo => int (($segsize + $blksize - 1) / $blksize),
481			done => 0,
482			blk => [],
483			blksize => $blksize,
484			};
485
486			push @segments, $seg;
487			$size -= $segsize;
488
489			while ($segsize > 0) {
490			push @{$seg->{blk}}, $idx;
491			for ($blk->[$idx++]) {
492			$_->{offset} = $offset;
493			#$_->{size} = $blksize > $segsize ? $segsize : $blksize; # WRONG
494			$_->{size} = $blksize;
495			$_->{seg} = $seg;
496			}
497
498			$segsize -= $blksize;
499			$offset += $blksize;
500
501			if (($redundandy[0] += $redundandy[1]) >= $redundandy[2]) {
502			$redundandy[0] -= $redundandy[2];
503
504			push @{$seg->{blk}}, $idx2;
505			for ($blk->[$idx2++]) {
506			$_->{offset} = $offset2;
507			$_->{size} = $blksize;
508			$_->{seg} = $seg;
509			}
510
511			$offset2 += $blksize;
512			}
513			}
514			}
515
516			$idx == $data_packets
517			or die "$self->{id}/$p->{tile} $self->{job}\nidx $idx != data_packets $data_packets";
518			$idx2 == $data_packets + $check_packets
519			or die "$self->{id}/$p->{tile} $self->{job}\nidx2 $idx2 != data_packets $data_packets + check_packets $check_packets";
520			}
521
522			for (@$blk) {
523			++$_->{seg}{done} if $_->{done};
524			delete $_->{htl};
525			}
526
527			my $fail = 0;
528			my $sig = new Coro::Signal;
529
530			$self->{status} = "splitfile fetch (" . @$blk . " blocks)";
531
532			my @txn;
533
534			for (;;) {
535			for my $id (0 .. $#$blk) {
536			my $blk = $blk->[$id];
537
538			next if $txn[$id] \|\| $blk->{done} \|\| $blk->{seg}{todo} <= $blk->{seg}{done};
539
540			my $htl = $HTL[$blk->{htl}++ % @HTL];
541	root	1.5	my $pri = int time + $htl / 20 * 7200 * rand;
542	root	1.1
543			txn_begin $pri;
544	root	1.5	#warn $self->{id} . ", GET<$htl, $pri>\n";#d#
545	root	1.1	$txn[$id] \|\|= $FCP->txn_client_get ($blk->{uri}, $htl)->cb(sub {
546			undef $txn[$id];
547
548			my $seg = $blk->{seg};
549
550			my ($meta, $data) = eval { @{ $_[0]->result } };
551
552			if (defined $data) {
553			$blk->{size} == length $data
554			or die sprintf "block $id expected size %d, got %d\n", $blk->{size}, length $data;
555
556	root	1.9	(length $data) == (aio_write $self->{fh}, $blk->{offset}, (length $data), $data, 0)
557	root	1.1	or die "unable to write chunk to disk, not setting valid flag";
558	root	1.9	aio_fsync $self->{fh};
559	root	1.1
560			$blk->{done} = 1;
561			$blk->{meta} = $meta->{raw} if length $meta->{raw};#d#
562			$seg->{done}++;
563			$self->save;
564
565			$::htl_sum += $htl;
566			$::htl_cnt++;
567
568	root	1.4	$self->log (sprintf "got block $seg->{id}.$id %d ($seg->{done}/$seg->{todo}) at htl $htl (%f) and pri $pri",
569			length $data, $::htl_sum / $::htl_cnt);
570	root	1.1	} else {
571			if ($@) {
572			if ($@->type ("data_not_found")) {
573			# nop
574			} elsif ($@->type ("network_error")) {
575			$self->log ("$@, retrying in 1s");
576			CORE::sleep 1;
577			} else {
578			$self->log ("$@");
579			}
580			}
581			++$fail;
582			}
583			$self->{status} = "splitfile fetch ($seg->{done}/$seg->{todo}, $fail failed)";
584
585			txn_end;
586			$sig->send;
587			});
588			}
589
590			for my $seg (@segments) {
591			if ($seg->{done} >= $seg->{todo} && !$seg->{finished}) {
592
593			$self->log ("segment done, cancelling segment $seg->{id}");
594			for my $id (@{$seg->{blk}}) {
595			(delete $txn[$id])->cancel if $txn[$id];
596			}
597
598	root	1.10	my $verify;
599	root	1.1	for my $id (@{$seg->{blk}}) {
600			my $blk = $blk->[$id];
601			if ($blk->{done}) {
602	root	1.10	aio_read $self->{fh}, $blk->{offset}, $blk->{size}, my $buf, 0;
603	root	1.1
604			my $k1 = Net::FCP::Util::extract_chk_hash $blk->{uri};
605			my $k2 = Net::FCP::Util::generate_chk_hash $blk->{meta}, $buf;
606
607			if ($k1 ne $k2) {
608	root	1.10	$verify .= "v";
609	root	1.1	#warn sprintf "$p->{title} block $id BROKEN (%s != %s)", (unpack "H", $k1), (unpack "H", $k2);
610			$blk->{done} = 0;
611			$seg->{done}--;
612			$self->save;
613			} else {
614	root	1.10	$verify .= "V";
615	root	1.1	}
616			}
617			}
618	root	1.10	$self->log ("verified segment $seg->{id}");
619			$self->log ($verify);
620	root	1.1
621			if ($seg->{done} >= $seg->{todo} && !$seg->{finished}) {
622			$self->log ("verified segment OK $seg->{id}");
623			$seg->{finished}++;
624			$segments--;
625			} else {
626			$self->log ("verified segment NOT OK $seg->{id}");
627			}
628			}
629			}
630
631			last unless $segments;
632
633			$sig->wait;
634			}
635
636			$self->log ("decoding < $self->{job} $self->{file} $filesize >");
637
638			for my $seg (@segments) {
639			my @part;
640			my @idx;
641			my @blk = map $blk->[$_], sort { $a <=> $b } @{$seg->{blk}};
642
643			for my $id (0 .. $#blk) {
644			my $blk = $blk[$id];
645			next unless $blk->{done};
646
647			push @part, [$self->{fh}, $blk->{offset}];
648			push @idx, $id;
649
650			last if @idx == $seg->{todo};
651			}
652
653			my $fec = new Algorithm::FEC
654			$seg->{todo},
655			scalar @blk,
656			$seg->{blksize};
657
658			$fec->shuffle (\@part, \@idx);
659
660			# now copy check blocks to their destination position
661			for my $i (0 .. $#idx) {
662			next if $idx[$i] == $i;
663
664			my $src = $part[$i];
665			$part[$i] = [$self->{fh}, $blk[$i]{offset}];
666			$fec->copy ($src, $part[$i]);
667			}
668
669			$fec->set_decode_blocks (\@part, \@idx);
670			$fec->decode;
671			}
672
673			my $sha1 = new Digest::SHA1;
674			open my $dd, "-\|", "head -c$filesize \Q$self->{file}\E"
675			or die "DD: $!";
676			#$dd = Coro::Handle::unblock $dd;
677			$sha1->addfile ($dd);
678			$sha1 = $sha1->hexdigest;
679
680			if (exists $p->{meta}{document}[0]{info}{checksum}
681			and $p->{meta}{document}[0]{info}{checksum} ne $sha1) {
682			$self->log ("META: $p->{meta}{document}[0]{info}{checksum} and real checksum $sha1 for $filesize DIFFER");
683	root	1.9	# $self->feedback ("CHECKSUM ERROR");
684			# terminate;
685	root	1.1	}
686
687			truncate $self->{fh}, $filesize;
688			sysseek $self->{fh}, 0, 0;
689
690			return ["finish", 1];
691			} else {
692			$self->log ("splitfile algo '$splitfile->{algo_name}' unknown");
693			$self->feedback ("algo unknown");
694			terminate;
695			}
696			}
697
698			package main;
699
700			$\|=1;
701
702			for (<\Q$QUEUE_HOME\E/*.j>) {
703			job->new_from_file ($_);
704			print "J";
705			}
706			print "\n";
707
708			open my $stdin , "<&0" or die;
709			open my $stdout, ">&1" or die;
710			cmdline unblock $stdin, unblock $stdout;
711
712			&Coro::Event::loop;
713