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

#!/opt/bin/perl

system "echo $$ >fmd.pid";

=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 Net::FCP;
use Storable;
use Time::HiRes;
use Coro 3.0;
use Coro::Channel;
use Coro::Event;
use Coro::Handle;
use Coro::Signal;
use Coro::Timer;
use Coro::AIO;
use List::Util;
use Digest::SHA1;
use Algorithm::FEC;
use Digest::SHA1;
use Net::FCP::Util;
use POSIX ();

Coro::AIO::max_poll_time 0.01;
Coro::AIO::min_parallel 1;
Coro::AIO::max_parallel 2;

$|=1;

our $MAX_TXN = 250; # use max. this many transactions in parallel
our @HTL = (5,12,20);
our $VERIFY_CHK = 1; # verify all blocks, again and again (only useful to debugging).
our $FMD_HOME = $ENV{FMD_HOME} || "$ENV{HOME}/fmd";

our $FCP = new Net::FCP;

defined $FMD_HOME
   or die "you currently must define FMD_HOME to a persistent directory";

mkdir $FMD_HOME, 0700;

our $QUEUE_HOME = "$FMD_HOME/tmp";
mkdir $QUEUE_HOME, 0700;
our $DONE_HOME = "$FMD_HOME/done";
mkdir $DONE_HOME, 0700;

our %job;

sub push_key {
   my ($key, $title) = @_;

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

   my $job = job->new_from_key ($key, $title);
   warn "added as job $job->{id}\n";
   $job;
}

sub cmdline {
   my ($i, $o) = @_;

   my $cmd = async {
      my $job;
      while (print $o "> " and defined ($_ = <$i>)) {
         chomp;
         if (/<attach(?:ment|ed|)>(.*) \* (CHK[^<]+)<\/attach/) {
            $job = push_key $2, $1;
         } elsif (/(CHK\@[a-zA-Z0-9,~\-]{54})[\/ ]+(.*?)\s*$/) {
            $job = push_key $1, $2;
         } elsif (s/^(\d+)//) {
            if ($job = $job{$1}) {
               print $o delete $job->{log};
               if ($job->{input}) {
               }
            }
            redo;
         } elsif (/^q/) {
            close $i;
            close $o;
         } elsif (/^l/) {
            for my $job (sort { $a->{id} <=> $b->{id} } values %job) {
               print $o "$_ $job->{id}: $job->{p}{key} $job->{p}{title} $job->{status}\n";
            }
         } elsif (/^pri\s*(\d+)/) {
            if ($job) {
               $job->{p}{pri} = $1;
               $job->save;
            }
         } elsif (/^s$/) {
            print $o $job->show if $job;
         } elsif (/^k$/) {
            $job->kill if $job;
         } elsif (/^p(.*)/) {
            print +(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;

   $self->{p}{title} !~ /\//
      or die "$self->{p}{title}: malformed key\n";

   $self->{p}{title} =~ s/\s+$//;
   $self->{p}{title} =~ s/\n/_/g;

   $self->{id} = ++$count;
   $self->{job} ||= "$QUEUE_HOME/" . Time::HiRes::time . ":$count.j";

   $job{$self->{id}} = $self;

   $self->save;
   $self->start;
   $self;
}

sub new_from_key {
   my ($class, $key, $title) = @_;
   $class->new (p => { key => $key, title => $title, state => "examine" });
}

sub new_from_file {
   my ($class, $path) = @_;
   $class->new (job => $path, p => Storable::retrieve $path);
}

sub save {
   my ($self) = @_;
   Storable::nstore $self->{p}, "$self->{job}~";
   rename "$self->{job}~", $self->{job};
}

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

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

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

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

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;

   print "+";#d#
   #warn "txn_begin $pri\n";#d#
   push_heap @queue, [$pri, $sig];
   $queue_change->send;
   $sig->wait;
}

sub txn_end {
   $queue_alloc--;
   $queue_change->send;
   print "-";#d#
}

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->{log} .= "$time $text\n";
}

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

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

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

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

   $self->{status} = "starting";
   $self->{coro} = async {
      for(;;) {
         my ($state, @args) = ref $self->{p}{state} ? @{$self->{p}{state}} : $self->{p}{state};
         my $next = eval { $self->can ("state_$state")->($self, @args) };
         if ($@) {
            $self->log ($@);
            $next = $self->feedback ("continue with state: ");
         }
         $self->log ($self->{status} = "STATE CHANGE: ". join ", ", ref $next ? @$next : $next);
         $self->{p}{state} = $next;
         $self->save;
      }
   };
}

sub state_finish {
   my ($self, $save) = @_;

   if ($save) {
      aio_fsync $self->{fh};
      close $self->{fh};

      aio_unlink "$DONE_HOME/$self->{p}{title}";
      aio_link $self->{file}, "$DONE_HOME/$self->{p}{title}"
         and die "link: $self->{file} => $DONE_HOME/$self->{p}{title}: $!";
      system "sync";
   }
   $self->clean;

   aio_unlink $self->{file};

   $self->{status} = "finished";
   $self->feedback ("finished");
   terminate;
}

sub state_examine {
   my ($self) = @_;
   my $p = $self->{p};

   $self->{status} = "initial fetch";

   for (;;) {
      $self->log ("fetching $p->{key} (=$p->{title})");

      ($p->{meta}, $p->{data}) = fetch_uri 100, "freenet:$p->{key}";
      $self->save;
      #use PApp::Util; print STDERR PApp::Util::dumpval [keys %{$meta->{document}[0]{split_file}}];
      $self->log ("type $p->{meta}{document}[0]{info}{format}");

      if (my $splitfile = $p->{meta}{document}[0]{split_file}) {
         return "splitfile";
      } elsif ((defined $p->{data}) and (length $p->{data})) {
         syswrite $self->{fh}, delete $p->{data};
         aio_fsync $self->{fh};
         return ["finish", 1];
      }

      $self->log ("EMPTY, retrying in an hour");
      Coro::Timer::sleep 3600;
   }

}

sub state_splitfile {
   my ($self) = @_;
   my $p = $self->{p};

   my $splitfile = $p->{meta}{document}[0]{split_file};
   my $filesize = hex $splitfile->{size};

   if ($splitfile->{algo_name} eq "OnionFEC_a_1_2") {
      my $data_packets = hex $splitfile->{block_count};
      my $check_packets = hex $splitfile->{check_block_count};

      my $blk = ($p->{blk} ||= []);

      unless (@$blk) {
         for (1..$data_packets) {
            push @$blk, {
               uri => $splitfile->{block}{sprintf "%x", $_},
            };
         }
         for (1..$check_packets) {
            push @$blk, {
               uri => $splitfile->{check_block}{sprintf "%x", $_},
            };
         }
      }

      my @segments;
      my $segments = 0;

      {
         # that is a horrible algorithm :(, these freenet freaks are... java-disabled
         # hardcoding lots of magic parameters is soo dumb.
         my $size = $filesize;
         my $offset = 0;
         my $offset2 = ($filesize & ~(1024*1024-1)) + 1024*1024; # leave enough space after last data block
         my $idx  = 0;
         my $idx2 = $data_packets;
         my @redundandy = (0,1,2); # maybe OnionFAC_a_1_2 means 1/2 redundancy(?)

         while ($size > 0) {
            my $segsize = $size >= MAXSEG ? MAXSEG : $size <= MINSEG ? MINSEG : $size;
            my $blksize = blocksize $segsize;
            my $seg    = 
               {
                  id      => $segments++,
                  todo    => int (($segsize + $blksize - 1) / $blksize),
                  done    => 0,
                  blk     => [],
                  blksize => $blksize,
               };

            push @segments, $seg;
            $size -= $segsize;

            while ($segsize > 0) {
               push @{$seg->{blk}}, $idx;
               for ($blk->[$idx++]) {
                  $_->{offset} = $offset;
                  #$_->{size}   = $blksize > $segsize ? $segsize : $blksize; # WRONG
                  $_->{size}   = $blksize;
                  $_->{seg}    = $seg;
               }

               $segsize -= $blksize;
               $offset  += $blksize;

               if (($redundandy[0] += $redundandy[1]) >= $redundandy[2]) {
                  $redundandy[0] -= $redundandy[2];

                  push @{$seg->{blk}}, $idx2;
                  for ($blk->[$idx2++]) {
                     $_->{offset} = $offset2;
                     $_->{size}   = $blksize;
                     $_->{seg}    = $seg;
                  }

                  $offset2 += $blksize;
               }
            }
         }

         $idx == $data_packets
            or die "$self->{id}/$p->{tile} $self->{job}\nidx $idx != data_packets $data_packets";
         $idx2 == $data_packets + $check_packets
            or die "$self->{id}/$p->{tile} $self->{job}\nidx2 $idx2 != data_packets $data_packets + check_packets $check_packets";
      }

      for (@$blk) {
         ++$_->{seg}{done} if $_->{done};
         delete $_->{htl};
      }

      my $fail = 0;
      my $sig = new Coro::Signal;

      $self->{status} = "splitfile fetch (" .  @$blk . " blocks)";

      my @txn;

      for (;;) {
         for my $id (0 .. $#$blk) {
            my $blk = $blk->[$id];

            next if $txn[$id] || $blk->{done} || $blk->{seg}{todo} <= $blk->{seg}{done};

            my $htl = $HTL[$blk->{htl}++ % @HTL];
            my $pri = int time + $htl / 20 * 7200 * rand;

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

async {
   for (<\Q$QUEUE_HOME\E/*.j>) {
      job->new_from_file ($_);
      print "J";
   }
   print "\n";
};

open my $stdin , "<&0" or die;
open my $stdout, ">&1" or die;
cmdline unblock $stdin, unblock $stdout;

&Coro::Event::loop;

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