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

#!/opt/bin/perl

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

$|=1;

our $MAX_TXN = 300; # use max. this many transactions in parallel
our @HTL = (0, 1, 3, 5, 0, 8, 0, 10, 0, 15, 0, 20, 0, 25);
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})[\/ ]+(.*)$/) {
            $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 (/\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 Coro;
use Fcntl;
use IO::Handle;
use Array::Heap;

my $count = 0;

sub new {
   my $class = shift;

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

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

my @queue;
my $queue_change = new Coro::Signal;
my $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++;
      }
      $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 $pri / ($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" .
   "";
}

my $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->{file} = "$QUEUE_HOME/tmp.$self->{p}{title}";
   sysopen $self->{fh}, $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: $next");
         $self->{p}{state} = $next;
         $self->save;
      }
   };
}

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

   if ($save) {
      IO::Handle::sync $self->{fh};
      close $self->{fh};

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

   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}, $p->{data};
         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 = ($p->{pri} + 10) * 0.1 * ($blk->{seg}{done} + 1) / $blk->{seg}{todo} * rand;

            txn_begin $pri;
            #warn +($self->{id}+1000) . ", GET<$blk->{uri}, $htl, $pri>\n";
            $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;

                  $blk->{offset} == sysseek $self->{fh}, $blk->{offset}, 0
                     or die "sysseek: $!";
                  (length $data) == (syswrite $self->{fh}, $data)
                     or die "unable to write chunk to disk, not setting valid flag";
                  IO::Handle::sync $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 ($seg->{done}/$seg->{todo}) at htl $htl (%f) and pri $pri", $::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];
               }

               $self->log ("verifying segment $seg->{id}");
               for my $id (@{$seg->{blk}}) {
                  my $blk = $blk->[$id];
                  if ($blk->{done}) {
                     sysseek $self->{fh}, $blk->{offset}, 0;
                     sysread $self->{fh}, my $buf, $blk->{size};

                     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) {
                        print "v";#d#
                        #warn sprintf "$p->{title} block $id BROKEN (%s != %s)", (unpack "H*", $k1), (unpack "H*", $k2);
                        $blk->{done} = 0;
                        $seg->{done}--;
                        $self->save;
                     } else {
                        print "V";#d#
                     }
                  }
               }
               print "\n";

               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;

&Coro::Event::loop;

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