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

#!/opt/bin/perl

#system "echo $$ >fmd.pid";
#use Coro::Debug; our $server = new_unix_server Coro::Debug "debug";

=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 (250) 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 EV;
use Coro::EV;

use Net::FCP;
use Storable ();
use Time::HiRes;
use Coro;
use Coro::Channel;
use Coro::Handle;
use Coro::Signal;
use Coro::Timer;
use Coro::AIO;
use Coro::Storable ();
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 32;

EV::set_io_collect_interval 0.2;
EV::set_timeout_collect_interval 0.05;

$|=1;

our $MAX_TXN = 400; # use max. this many transactions in parallel
our @HTL = (5,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/job";
mkdir $QUEUE_HOME, 0700;
our $DATA_HOME = "$FMD_HOME/tmp";
mkdir $DATA_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 (/^p(.*)/) {
            print $o +(join " ", eval $1), "\n";
         } 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;

our $count = 0;

sub new {
   my $class = shift;

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

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

   $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->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) = @_;
   if (my $fh = aio_open "$self->{job}~", O_CREAT|O_TRUNC|O_WRONLY, 0600) {
      aio_write $fh, 0, undef, Coro::Storable::blocking_nfreeze $self->{p}, 0;
      aio_fsync $fh;
      aio_close $fh;
      aio_rename "$self->{job}~", $self->{job};
      aio_pathsync $QUEUE_HOME;
   }
}

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

   delete $job{$self->{id}};
   $self->save;
   system "mv", $self->{job}, "$DONE_HOME/$self->{p}{title}.job";
   unlink $self->{job};
}

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

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

our @queue;
our $queue_change = new Coro::Signal;
our $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 txn_client_get {
   my %arg = @_;

   txn_begin $arg{pri};
   $FCP->txn_client_get ($arg{uri}, $arg{htl})->cb (unblock_sub {
      txn_end;

      $arg{cb}->(@_);
   });
}

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

   for(my $count = 1; ; $count += 0.3) {
      for my $htl (@HTL) {
         txn_begin time + $htl + $count;
         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->{p}{log} .= "$time $text\n";
   $self->save;
}

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

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

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

our $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} = "$DATA_HOME/$1.d";
   sysopen $self->{fh}, $self->{file}, O_RDWR|O_CREAT, 0600
   #$self->{fh} = aio_open $self->{file}, O_RDWR|O_CREAT, 0600 
      or die "$self->{file}: $!";

   $self->{status} = "starting";
   $self->{coro} = async {
      $self->save;

      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}: $!";
      aio_pathsync $DONE_HOME;
   }
   $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;

            #warn $self->{id} . ", GET<$htl, $pri>\n";#d#
            $txn[$id] ||= txn_client_get pri => $pri, uri => $blk->{uri}, htl => $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)";

               $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;

EV::loop;

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