[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.05;
Coro::AIO::min_parallel 4;
Coro::AIO::max_parallel 4;

$|=1;

our $MAX_TXN = 50; # use max. this many transactions in parallel
our @HTL = (3,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) {
      my $data = Coro::Storable::blocking_nfreeze $self->{p};
      aio_write $fh, 0, undef, $data, 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 "%Y-%m-%d %H:%M:%S", localtime time;
   warn "$time $self->{id},$self->{p}{pri}: $text\n";

   unless ($text =~ /Net::FCP::Exception/) {
      $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 EV::now + $htl / 20 * 7200 * rand;

            #warn $self->{id} . ", GET<$htl, $pri>\n";#d#
            my $start = EV::now;

            $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 (%.1fs)",
                                       length $data, $::htl_sum / $::htl_cnt, EV::now - $start);
               } 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::set_io_collect_interval 0.1;

EV::loop;

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