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

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

sub txn_end {
   $queue_alloc--;
   $queue_change->send;
}

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

   for(my $count = 1; ; $count += 0.3) {
      for my $htl (@HTL) {
         txn_begin time + $htl + $count;
         my $sig = new Coro::Signal;
         my ($meta, $data) = eval { @{ $FCP->client_get ($uri, $htl) } };
         txn_end;
         if ($@) {
            if (UNIVERSAL::isa ($@, Net::FCP::Exception::)) {
               if ($@->type ("data_not_found")
                   || $@->type ("route_not_found")) {
                  next;
               }
               if ($@->type ("short_data")) {
                  warn "(short_data, redo)\n";
                  redo;
               }
               die;
            }
         }
         if (defined $data) {
            return ($meta, $data);
         }
      }
   }

   die;
}

sub log {
   my ($self, $text) = @_;
   my $time = POSIX::strftime "%H:%M:%S", localtime time;
   warn "$time $self->{id},$self->{p}{pri}: $text\n";
   $self->{log} .= "$time $text\n";
}

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

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

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

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;

            txn_begin $pri;
            #warn $self->{id} . ", GET<$htl, $pri>\n";#d#
            $txn[$id] ||= $FCP->txn_client_get ($blk->{uri}, $htl)->cb(sub {
               undef $txn[$id];

               my $seg = $blk->{seg};

               my ($meta, $data) = eval { @{ $_[0]->result } };

               if (defined $data) {
                  $blk->{size} == length $data
                     or die sprintf "block $id expected size %d, got %d\n", $blk->{size}, length $data;

                  (length $data) == (aio_write $self->{fh}, $blk->{offset}, (length $data), $data, 0)
                     or die "unable to write chunk to disk, not setting valid flag";
                  aio_fsync $self->{fh};

                  $blk->{done} = 1;
                  $blk->{meta} = $meta->{raw} if length $meta->{raw};#d#
                  $seg->{done}++;
                  $self->save;

                  $::htl_sum += $htl;
                  $::htl_cnt++;

                  $self->log (sprintf "got block $seg->{id}.$id %d ($seg->{done}/$seg->{todo}) at htl $htl (%f) and pri $pri",
                                       length $data, $::htl_sum / $::htl_cnt);
               } else {
                  if ($@) {
                     if ($@->type ("data_not_found")) {
                        # nop
                     } elsif ($@->type ("network_error")) {
                        $self->log ("$@, retrying in 1s");
                        CORE::sleep 1;
                     } else {
                        $self->log ("$@");
                     }
                  }
                  ++$fail;
               }
               $self->{status} = "splitfile fetch ($seg->{done}/$seg->{todo}, $fail failed)";

               txn_end;
               $sig->send;
            });
         }

         for my $seg (@segments) {
            if ($seg->{done} >= $seg->{todo} && !$seg->{finished}) {

               $self->log ("segment done, cancelling segment $seg->{id}");
               for my $id (@{$seg->{blk}}) {
                  (delete $txn[$id])->cancel if $txn[$id];
               }

               my $verify;
               for my $id (@{$seg->{blk}}) {
                  my $blk = $blk->[$id];
                  if ($blk->{done}) {
                     aio_read $self->{fh}, $blk->{offset}, $blk->{size}, my $buf, 0;

                     my $k1 = Net::FCP::Util::extract_chk_hash $blk->{uri};
                     my $k2 = Net::FCP::Util::generate_chk_hash $blk->{meta}, $buf;

                     if ($k1 ne $k2) {
                        $verify .= "v";
                        #warn sprintf "$p->{title} block $id BROKEN (%s != %s)", (unpack "H*", $k1), (unpack "H*", $k2);
                        $blk->{done} = 0;
                        $seg->{done}--;
                        $self->save;
                     } else {
                        $verify .= "V";
                     }
                  }
               }
               $self->log ("verified segment $seg->{id}");
               $self->log ($verify);

               if ($seg->{done} >= $seg->{todo} && !$seg->{finished}) {
                  $self->log ("verified segment OK $seg->{id}");
                  $seg->{finished}++;
                  $segments--;
               } else {
                  $self->log ("verified segment NOT OK $seg->{id}");
               }
            }
         }

         last unless $segments;

         $sig->wait;
      }

      $self->log ("decoding < $self->{job} $self->{file} $filesize >");

      for my $seg (@segments) {
         my @part;
         my @idx;
         my @blk = map $blk->[$_], sort { $a <=> $b } @{$seg->{blk}};

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

            push @part, [$self->{fh}, $blk->{offset}];
            push @idx, $id;

            last if @idx == $seg->{todo};
         }

         my $fec = new Algorithm::FEC
                          $seg->{todo},
                          scalar @blk,
                          $seg->{blksize};

         $fec->shuffle (\@part, \@idx);

         # now copy check blocks to their destination position
         for my $i (0 .. $#idx) {
            next if $idx[$i] == $i;

            my $src = $part[$i];
            $part[$i] = [$self->{fh}, $blk[$i]{offset}];
            $fec->copy ($src, $part[$i]);
         }

         $fec->set_decode_blocks (\@part, \@idx);
         $fec->decode;
      }

      my $sha1 = new Digest::SHA1;
      open my $dd, "-|", "head -c$filesize \Q$self->{file}\E"
         or die "DD: $!";
      #$dd = Coro::Handle::unblock $dd;
      $sha1->addfile ($dd);
      $sha1 = $sha1->hexdigest;

      if (exists $p->{meta}{document}[0]{info}{checksum}
          and $p->{meta}{document}[0]{info}{checksum} ne $sha1) {
         $self->log ("META: $p->{meta}{document}[0]{info}{checksum} and real checksum $sha1 for $filesize DIFFER");
#         $self->feedback ("CHECKSUM ERROR");
#         terminate;
      }

      truncate $self->{fh}, $filesize;
      sysseek $self->{fh}, 0, 0;

      return ["finish", 1];
   } else {
      $self->log ("splitfile algo '$splitfile->{algo_name}' unknown");
      $self->feedback ("algo unknown");
      terminate;
   }
}

package main;

$|=1;

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