[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 4.23;
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;

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