[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 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 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: $next");
         $self->{p}{state} = $next;
         $self->save;
      }
   };
}

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

   if ($save) {
      IO::Handle::sync $self->{fh};
      close $self->{fh};

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

   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];
               }

               $self->log ("verifying segment $seg->{id}");
               for my $id (@{$seg->{blk}}) {
                  my $blk = $blk->[$id];
                  if ($blk->{done}) {
                     sysseek $self->{fh}, $blk->{offset}, 0;
                     sysread $self->{fh}, my $buf, $blk->{size};

                     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) {
                        print "v";#d#
                        #warn sprintf "$p->{title} block $id BROKEN (%s != %s)", (unpack "H*", $k1), (unpack "H*", $k2);
                        $blk->{done} = 0;
                        $seg->{done}--;
                        $self->save;
                     } else {
                        print "V";#d#
                     }
                  }
               }
               print "\n";

               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.9
Committed:	Mon Nov 20 19:24:16 2006 UTC (19 years, 9 months ago) by root
Branch:	MAIN
Changes since 1.8:	+21 -13 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			use Net::FCP;
91			use Storable;
92			use Time::HiRes;
93			use Coro;
94			use Coro::Event;
95			use Coro::Handle;
96			use Coro::Signal;
97			use Coro::Timer;
98	root	1.9	use Coro::AIO;
99	root	1.1	use List::Util;
100			use Digest::SHA1;
101			use Algorithm::FEC;
102			use Digest::SHA1;
103			use Net::FCP::Util;
104			use POSIX ();
105
106	root	1.9	Coro::AIO::max_poll_time 0.01;
107			Coro::AIO::min_parallel 1;
108			Coro::AIO::max_parallel 2;
109
110	root	1.1	$\|=1;
111
112	root	1.9	our $MAX_TXN = 150; # use max. this many transactions in parallel
113			our @HTL = (5,12,20);
114	root	1.1	our $VERIFY_CHK = 1; # verify all blocks, again and again (only useful to debugging).
115			our $FMD_HOME = $ENV{FMD_HOME} \|\| "$ENV{HOME}/fmd";
116
117			our $FCP = new Net::FCP;
118
119			defined $FMD_HOME
120			or die "you currently must define FMD_HOME to a persistent directory";
121
122			mkdir $FMD_HOME, 0700;
123
124			our $QUEUE_HOME = "$FMD_HOME/tmp";
125			mkdir $QUEUE_HOME, 0700;
126			our $DONE_HOME = "$FMD_HOME/done";
127			mkdir $DONE_HOME, 0700;
128
129			our %job;
130
131			sub push_key {
132			my ($key, $title) = @_;
133
134			for my $job (values %job) {
135			if ($job->{p}{key} eq $key) {
136			warn "job $job->{id} already works on this, not adding";
137			return;
138			}
139			}
140
141			my $job = job->new_from_key ($key, $title);
142			warn "added as job $job->{id}\n";
143			$job;
144			}
145
146			sub cmdline {
147			my ($i, $o) = @_;
148
149			my $cmd = async {
150			my $job;
151			while (print $o "> " and defined ($_ = <$i>)) {
152			chomp;
153			if (/<attach(?:ment\|ed\|)>(.) \ (CHK[^<]+)<\/attach/) {
154			$job = push_key $2, $1;
155	root	1.7	} elsif (/(CHK\@[a-zA-Z0-9,~\-]{54})[\/ ]+(.?)\s$/) {
156	root	1.1	$job = push_key $1, $2;
157			} elsif (s/^(\d+)//) {
158			if ($job = $job{$1}) {
159			print $o delete $job->{log};
160			if ($job->{input}) {
161			}
162			}
163			redo;
164			} elsif (/^q/) {
165			close $i;
166			close $o;
167			} elsif (/^l/) {
168			for my $job (sort { $a->{id} <=> $b->{id} } values %job) {
169			print $o "$_ $job->{id}: $job->{p}{key} $job->{p}{title} $job->{status}\n";
170			}
171			} elsif (/^pri\s*(\d+)/) {
172			if ($job) {
173			$job->{p}{pri} = $1;
174			$job->save;
175			}
176			} elsif (/^s$/) {
177			print $o $job->show if $job;
178			} elsif (/^k$/) {
179			$job->kill if $job;
180			} elsif (/\S/) {
181			#
182			} else {
183			print $o "?\n";
184			}
185			for my $job (sort { $a->{id} <=> $b->{id} } values %job) {
186			if ($job->{_input}) {
187			print $o "> $job->{id} $job->{title} $job->{status}\n";
188			}
189			}
190			}
191			};
192			}
193
194			package job;
195
196			use Coro;
197	root	1.9	use Coro::AIO;
198	root	1.1	use Fcntl;
199			use IO::Handle;
200	root	1.6	use Array::Heap2;
201	root	1.1
202			my $count = 0;
203
204			sub new {
205			my $class = shift;
206
207			my $self = bless { @_ }, $class;
208
209	root	1.9	$self->{p}{title} !~ /\//
210			or die "$self->{p}{title}: malformed key\n";
211
212	root	1.1	$self->{id} = ++$count;
213			$self->{job} \|\|= "$QUEUE_HOME/" . Time::HiRes::time . ":$count.j";
214
215			$job{$self->{id}} = $self;
216
217			$self->save;
218			$self->start;
219			$self;
220			}
221
222			sub new_from_key {
223			my ($class, $key, $title) = @_;
224			$class->new (p => { key => $key, title => $title, state => "examine" });
225			}
226
227			sub new_from_file {
228			my ($class, $path) = @_;
229			$class->new (job => $path, p => Storable::retrieve $path);
230			}
231
232			sub save {
233			my ($self) = @_;
234			Storable::nstore $self->{p}, "$self->{job}~";
235	root	1.9	aio_rename "$self->{job}~", $self->{job};
236	root	1.1	}
237
238			sub clean {
239			my ($self) = @_;
240
241			delete $job{$self->{id}};
242			$self->save;
243	root	1.9	aio_rename $self->{job}, "$DONE_HOME/$self->{p}{title}.job";
244			aio_unlink $self->{job};
245	root	1.1	}
246
247			sub kill {
248			my ($self) = @_;
249
250			$self->clean;
251			$self->{coro}->cancel;
252			}
253
254			my @queue;
255			my $queue_change = new Coro::Signal;
256			my $queue_alloc = 0;
257
258			async {
259			for (;;) {
260			while (@queue
261			and (($queue[0][0] > 10 and $queue_alloc < $MAX_TXN)
262			or ($queue[0][0] > 1 and $queue_alloc < $MAX_TXN - 3)
263			or $queue_alloc < $MAX_TXN - 5)) {
264			(pop_heap @queue)->[1]->send;
265			$queue_alloc++;
266	root	1.4	Coro::Timer::sleep 0.05;
267	root	1.1	}
268			$queue_change->wait;
269			}
270			};
271
272			sub txn_begin {
273			my ($pri) = @_;
274			my $sig = new Coro::Signal;
275
276			#warn "txn_begin $pri\n";#d#
277			push_heap @queue, [$pri, $sig];
278			$queue_change->send;
279			$sig->wait;
280			}
281
282			sub txn_end {
283			$queue_alloc--;
284			$queue_change->send;
285			}
286
287			sub fetch_uri {
288			my ($pri, $uri) = @_;
289
290			for(my $count = 1; ; $count += 0.3) {
291			for my $htl (@HTL) {
292	root	1.4	txn_begin time + $htl + $count;
293	root	1.1	my $sig = new Coro::Signal;
294			my ($meta, $data) = eval { @{ $FCP->client_get ($uri, $htl) } };
295			txn_end;
296			if ($@) {
297			if (UNIVERSAL::isa ($@, Net::FCP::Exception::)) {
298			if ($@->type ("data_not_found")
299			\|\| $@->type ("route_not_found")) {
300			next;
301			}
302			if ($@->type ("short_data")) {
303			warn "(short_data, redo)\n";
304			redo;
305			}
306			die;
307			}
308			}
309			if (defined $data) {
310			return ($meta, $data);
311			}
312			}
313			}
314
315			die;
316			}
317
318			sub log {
319			my ($self, $text) = @_;
320			my $time = POSIX::strftime "%H:%M:%S", localtime time;
321	root	1.3	warn "$time $self->{id},$self->{p}{pri}: $text\n";
322	root	1.1	$self->{log} .= "$time $text\n";
323			}
324
325			sub feedback {
326			my ($self, $prompt) = @_;
327			$self->{input} = [$Coro::current, $prompt];
328			Coro::schedule;
329			}
330
331			sub show {
332			my ($self) = @_;
333
334			"ID: $self->{id}\n"
335			. "Title: $self->{p}{title}\n"
336			. "Blocks#: " . @{$self->{p}{blk}} . "\n"
337			. "Blocks: " . (join "", map {
338			$_->{done} ? "+" : "-"
339			} @{$self->{p}{blk}}) . "\n" .
340			"";
341			}
342
343			my $id;
344
345			sub MAXSEG (){ 12810241024 }
346			sub MINSEG (){ 6* 128*1024 }
347
348			sub blocksize($) {
349			return
350			$_[0] >= 6410241024 ? 1024*1024
351			: $_[0] >= 3210241024 ? 512*1024
352			: $_[0] >= 10241024 ? 2561024
353			: 128*1024;
354			}
355
356			sub start {
357			my ($self) = @_;
358
359			$self->{p}{pri} \|\|= 1;
360			$self->{p}{state} \|\|= "examine";
361
362	root	1.9	$self->{job} =~ /^(.*)\.j$/ or die "$self->{job}: missing .j";
363			$self->{file} = "$1.d";
364			$self->{fh} = aio_open $self->{file}, O_RDWR\|O_CREAT, 0600
365	root	1.1	or die "$self->{file}: $!";
366
367			$self->{status} = "starting";
368			$self->{coro} = async {
369			for(;;) {
370			my ($state, @args) = ref $self->{p}{state} ? @{$self->{p}{state}} : $self->{p}{state};
371			my $next = eval { $self->can ("state_$state")->($self, @args) };
372			if ($@) {
373			$self->log ($@);
374			$next = $self->feedback ("continue with state: ");
375			}
376			$self->log ($self->{status} = "STATE CHANGE: $next");
377			$self->{p}{state} = $next;
378			$self->save;
379			}
380			};
381			}
382
383			sub state_finish {
384			my ($self, $save) = @_;
385
386			if ($save) {
387			IO::Handle::sync $self->{fh};
388			close $self->{fh};
389
390			unlink "$DONE_HOME/$self->{p}{title}";
391			link $self->{file}, "$DONE_HOME/$self->{p}{title}"
392			or die "link: $self->{file} => $DONE_HOME/$self->{p}{title}: $!";
393			system "sync";
394			}
395			$self->clean;
396
397			unlink $self->{file};
398
399			$self->{status} = "finished";
400			$self->feedback ("finished");
401			terminate;
402			}
403
404			sub state_examine {
405			my ($self) = @_;
406			my $p = $self->{p};
407
408			$self->{status} = "initial fetch";
409
410			for (;;) {
411			$self->log ("fetching $p->{key} (=$p->{title})");
412
413			($p->{meta}, $p->{data}) = fetch_uri 100, "freenet:$p->{key}";
414			$self->save;
415			#use PApp::Util; print STDERR PApp::Util::dumpval [keys %{$meta->{document}[0]{split_file}}];
416			$self->log ("type $p->{meta}{document}[0]{info}{format}");
417
418			if (my $splitfile = $p->{meta}{document}[0]{split_file}) {
419			return "splitfile";
420			} elsif ((defined $p->{data}) and (length $p->{data})) {
421	root	1.8	syswrite $self->{fh}, delete $p->{data};
422	root	1.1	return ["finish", 1];
423			}
424
425			$self->log ("EMPTY, retrying in an hour");
426			Coro::Timer::sleep 3600;
427			}
428
429			}
430
431			sub state_splitfile {
432			my ($self) = @_;
433			my $p = $self->{p};
434
435			my $splitfile = $p->{meta}{document}[0]{split_file};
436			my $filesize = hex $splitfile->{size};
437
438			if ($splitfile->{algo_name} eq "OnionFEC_a_1_2") {
439			my $data_packets = hex $splitfile->{block_count};
440			my $check_packets = hex $splitfile->{check_block_count};
441
442			my $blk = ($p->{blk} \|\|= []);
443
444			unless (@$blk) {
445			for (1..$data_packets) {
446			push @$blk, {
447			uri => $splitfile->{block}{sprintf "%x", $_},
448			};
449			}
450			for (1..$check_packets) {
451			push @$blk, {
452			uri => $splitfile->{check_block}{sprintf "%x", $_},
453			};
454			}
455			}
456
457			my @segments;
458			my $segments = 0;
459
460			{
461			# that is a horrible algorithm :(, these freenet freaks are... java-disabled
462			# hardcoding lots of magic parameters is soo dumb.
463			my $size = $filesize;
464			my $offset = 0;
465			my $offset2 = ($filesize & ~(10241024-1)) + 10241024; # leave enough space after last data block
466			my $idx = 0;
467			my $idx2 = $data_packets;
468			my @redundandy = (0,1,2); # maybe OnionFAC_a_1_2 means 1/2 redundancy(?)
469
470			while ($size > 0) {
471			my $segsize = $size >= MAXSEG ? MAXSEG : $size <= MINSEG ? MINSEG : $size;
472			my $blksize = blocksize $segsize;
473			my $seg =
474			{
475			id => $segments++,
476			todo => int (($segsize + $blksize - 1) / $blksize),
477			done => 0,
478			blk => [],
479			blksize => $blksize,
480			};
481
482			push @segments, $seg;
483			$size -= $segsize;
484
485			while ($segsize > 0) {
486			push @{$seg->{blk}}, $idx;
487			for ($blk->[$idx++]) {
488			$_->{offset} = $offset;
489			#$_->{size} = $blksize > $segsize ? $segsize : $blksize; # WRONG
490			$_->{size} = $blksize;
491			$_->{seg} = $seg;
492			}
493
494			$segsize -= $blksize;
495			$offset += $blksize;
496
497			if (($redundandy[0] += $redundandy[1]) >= $redundandy[2]) {
498			$redundandy[0] -= $redundandy[2];
499
500			push @{$seg->{blk}}, $idx2;
501			for ($blk->[$idx2++]) {
502			$_->{offset} = $offset2;
503			$_->{size} = $blksize;
504			$_->{seg} = $seg;
505			}
506
507			$offset2 += $blksize;
508			}
509			}
510			}
511
512			$idx == $data_packets
513			or die "$self->{id}/$p->{tile} $self->{job}\nidx $idx != data_packets $data_packets";
514			$idx2 == $data_packets + $check_packets
515			or die "$self->{id}/$p->{tile} $self->{job}\nidx2 $idx2 != data_packets $data_packets + check_packets $check_packets";
516			}
517
518			for (@$blk) {
519			++$_->{seg}{done} if $_->{done};
520			delete $_->{htl};
521			}
522
523			my $fail = 0;
524			my $sig = new Coro::Signal;
525
526			$self->{status} = "splitfile fetch (" . @$blk . " blocks)";
527
528			my @txn;
529
530			for (;;) {
531			for my $id (0 .. $#$blk) {
532			my $blk = $blk->[$id];
533
534			next if $txn[$id] \|\| $blk->{done} \|\| $blk->{seg}{todo} <= $blk->{seg}{done};
535
536			my $htl = $HTL[$blk->{htl}++ % @HTL];
537	root	1.5	my $pri = int time + $htl / 20 * 7200 * rand;
538	root	1.1
539			txn_begin $pri;
540	root	1.5	#warn $self->{id} . ", GET<$htl, $pri>\n";#d#
541	root	1.1	$txn[$id] \|\|= $FCP->txn_client_get ($blk->{uri}, $htl)->cb(sub {
542			undef $txn[$id];
543
544			my $seg = $blk->{seg};
545
546			my ($meta, $data) = eval { @{ $_[0]->result } };
547
548			if (defined $data) {
549			$blk->{size} == length $data
550			or die sprintf "block $id expected size %d, got %d\n", $blk->{size}, length $data;
551
552	root	1.9	(length $data) == (aio_write $self->{fh}, $blk->{offset}, (length $data), $data, 0)
553	root	1.1	or die "unable to write chunk to disk, not setting valid flag";
554	root	1.9	aio_fsync $self->{fh};
555	root	1.1
556			$blk->{done} = 1;
557			$blk->{meta} = $meta->{raw} if length $meta->{raw};#d#
558			$seg->{done}++;
559			$self->save;
560
561			$::htl_sum += $htl;
562			$::htl_cnt++;
563
564	root	1.4	$self->log (sprintf "got block $seg->{id}.$id %d ($seg->{done}/$seg->{todo}) at htl $htl (%f) and pri $pri",
565			length $data, $::htl_sum / $::htl_cnt);
566	root	1.1	} else {
567			if ($@) {
568			if ($@->type ("data_not_found")) {
569			# nop
570			} elsif ($@->type ("network_error")) {
571			$self->log ("$@, retrying in 1s");
572			CORE::sleep 1;
573			} else {
574			$self->log ("$@");
575			}
576			}
577			++$fail;
578			}
579			$self->{status} = "splitfile fetch ($seg->{done}/$seg->{todo}, $fail failed)";
580
581			txn_end;
582			$sig->send;
583			});
584			}
585
586			for my $seg (@segments) {
587			if ($seg->{done} >= $seg->{todo} && !$seg->{finished}) {
588
589			$self->log ("segment done, cancelling segment $seg->{id}");
590			for my $id (@{$seg->{blk}}) {
591			(delete $txn[$id])->cancel if $txn[$id];
592			}
593
594			$self->log ("verifying segment $seg->{id}");
595			for my $id (@{$seg->{blk}}) {
596			my $blk = $blk->[$id];
597			if ($blk->{done}) {
598			sysseek $self->{fh}, $blk->{offset}, 0;
599			sysread $self->{fh}, my $buf, $blk->{size};
600
601			my $k1 = Net::FCP::Util::extract_chk_hash $blk->{uri};
602			my $k2 = Net::FCP::Util::generate_chk_hash $blk->{meta}, $buf;
603
604			if ($k1 ne $k2) {
605			print "v";#d#
606			#warn sprintf "$p->{title} block $id BROKEN (%s != %s)", (unpack "H", $k1), (unpack "H", $k2);
607			$blk->{done} = 0;
608			$seg->{done}--;
609			$self->save;
610			} else {
611			print "V";#d#
612			}
613			}
614			}
615			print "\n";
616
617			if ($seg->{done} >= $seg->{todo} && !$seg->{finished}) {
618			$self->log ("verified segment OK $seg->{id}");
619			$seg->{finished}++;
620			$segments--;
621			} else {
622			$self->log ("verified segment NOT OK $seg->{id}");
623			}
624			}
625			}
626
627			last unless $segments;
628
629			$sig->wait;
630			}
631
632			$self->log ("decoding < $self->{job} $self->{file} $filesize >");
633
634			for my $seg (@segments) {
635			my @part;
636			my @idx;
637			my @blk = map $blk->[$_], sort { $a <=> $b } @{$seg->{blk}};
638
639			for my $id (0 .. $#blk) {
640			my $blk = $blk[$id];
641			next unless $blk->{done};
642
643			push @part, [$self->{fh}, $blk->{offset}];
644			push @idx, $id;
645
646			last if @idx == $seg->{todo};
647			}
648
649			my $fec = new Algorithm::FEC
650			$seg->{todo},
651			scalar @blk,
652			$seg->{blksize};
653
654			$fec->shuffle (\@part, \@idx);
655
656			# now copy check blocks to their destination position
657			for my $i (0 .. $#idx) {
658			next if $idx[$i] == $i;
659
660			my $src = $part[$i];
661			$part[$i] = [$self->{fh}, $blk[$i]{offset}];
662			$fec->copy ($src, $part[$i]);
663			}
664
665			$fec->set_decode_blocks (\@part, \@idx);
666			$fec->decode;
667			}
668
669			my $sha1 = new Digest::SHA1;
670			open my $dd, "-\|", "head -c$filesize \Q$self->{file}\E"
671			or die "DD: $!";
672			#$dd = Coro::Handle::unblock $dd;
673			$sha1->addfile ($dd);
674			$sha1 = $sha1->hexdigest;
675
676			if (exists $p->{meta}{document}[0]{info}{checksum}
677			and $p->{meta}{document}[0]{info}{checksum} ne $sha1) {
678			$self->log ("META: $p->{meta}{document}[0]{info}{checksum} and real checksum $sha1 for $filesize DIFFER");
679	root	1.9	# $self->feedback ("CHECKSUM ERROR");
680			# terminate;
681	root	1.1	}
682
683			truncate $self->{fh}, $filesize;
684			sysseek $self->{fh}, 0, 0;
685
686			return ["finish", 1];
687			} else {
688			$self->log ("splitfile algo '$splitfile->{algo_name}' unknown");
689			$self->feedback ("algo unknown");
690			terminate;
691			}
692			}
693
694			package main;
695
696			$\|=1;
697
698			for (<\Q$QUEUE_HOME\E/*.j>) {
699			job->new_from_file ($_);
700			print "J";
701			}
702			print "\n";
703
704			open my $stdin , "<&0" or die;
705			open my $stdout, ">&1" or die;
706			cmdline unblock $stdin, unblock $stdout;
707
708			&Coro::Event::loop;
709