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

#!/opt/bin/perl
use Carp;$SIG{QUIT} = sub { Carp::confess "QUIT" };#d#

=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}~";
   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;

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