[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->{p}{title} =~ s/\s+$//;
   $self->{p}{title} =~ s/\n/_/g;

   $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}~";
   rename "$self->{job}~", $self->{job};
}

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

   delete $job{$self->{id}};
   $self->save;
   rename $self->{job}, "$DONE_HOME/$self->{p}{title}.job";
   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};
         aio_fsync $self->{fh};
         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;

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