[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 = 350; # use max. this many transactions in parallel
our @HTL = (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})[\/ ]+(.*)$/) {
            $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++;
         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" .
   "";
}

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

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

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