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

#!/opt/bin/perl

system "echo $$ >fmd.pid";

=head1 fmd - the freenet mass downloader

Fmd is at a very early stage of development (and very hackish, too), as I
am learning the basics of freenet myself.

However, I use it in production, and since it verifies everything it
decodes etc., it seems to be quite safe to use, that is, if you know how
to debug perl :)

=head2 FEATURES

 - decoding is done "in place", i.e. all non-checkblock blocks are
   stored in-place and will not be moved, only checkblocks will
   be moved to their final position in the file before decoding.
 - high resistance against failures of fred or fmd
 - extremely persistent retry behaviour - there is no such thing
   as a permanent failure.
 - handles hundreds of simultaneous downloads with grace.

=head2 ENVIRONMENT

Set FMD_HOME to a directory (default ~/fmd) where fmd will store it's
files. It will not store your porn files or other freenet data outside
that directory.

The subdirectory db will contain a database (soon to go away), while tmp
contains queue files and partial splitfiles. All finished files will be
moved to the done subdir.

FREDHOST and FREDPORT do the obvious. Fix these docs if you disagree.

Also edit the fmd executable for the number of threads and other
non-useful constants. The default number (250) works for me, probably not
for you.

=head2 COMMANDS

=over 4

=item CHK@... (space or "/") filename

Just pasting a CHK and a filename sperated by one or more spaces or
slashes will add files to the queue. The line can have leading garbage,
i.e. you can paste full uris.

=item <attachment>...</attachment>

=item <attach>...</attach>

=item <attached>...</attached>

Will add frosts mentally deranged pseudo-xml format, soon to be replaced
by something even more horrible.

=item l

List all jobs by number.

=item <number> (optionally trailing command)

Selects the job with the given number for further comamnds that require a current job.

=item s

Show the current job. Bot useful right now.

=item k

Kill the current job. This will keep the temp. file around. Sorry. (But it
could be used to reconstruct the download.. hmm..)

=item pri<number>

Set the current job priority to <pri>. The default is 1. A job with higher
priority will (on average) get more requests.

The useful range is probably 0..100, but be careful and limit yourself to
small values (<10), otherwise your other downloads might starve!

=item q

Kill the command prompt. Yupp.

=back

=cut

use strict;

use Net::FCP;
use Storable;
use Time::HiRes;
use Coro 3.0;
use Coro::Channel;
use Coro::Event;
use Coro::Handle;
use Coro::Signal;
use Coro::Timer;
use Coro::AIO;
use List::Util;
use Digest::SHA1;
use Algorithm::FEC;
use Digest::SHA1;
use Net::FCP::Util;
use POSIX ();

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

$|=1;

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

our $FCP = new Net::FCP;

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

mkdir $FMD_HOME, 0700;

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

our %job;

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

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

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

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

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

package job;

use strict;

use Coro;
use Coro::AIO;
use Fcntl;
use IO::Handle;
use Array::Heap2;

our $count = 0;

sub new {
   my $class = shift;

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

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

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

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

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

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

   $self->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 txn_client_get {
   my %arg = @_;

   txn_begin $arg{pri};
   $FCP->txn_client_get ($arg{uri}, $arg{htl})->cb (unblock_sub {
      txn_end;

      $arg{cb}->(@_);
   });
}

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

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

   die;
}

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

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

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

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

our $id;

sub MAXSEG (){ 128*1024*1024 }
sub MINSEG (){   6* 128*1024 }

sub blocksize($) {
   return
        $_[0] >= 64*1024*1024 ? 1024*1024
      : $_[0] >= 32*1024*1024 ?  512*1024
      : $_[0] >=    1024*1024 ?  256*1024
                              :  128*1024;
}

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

   $self->{p}{pri} ||= 1;
   $self->{p}{state} ||= "examine";

   $self->{job} =~ /\/([^\/]*)\.j$/ or die "$self->{job}: missing .j";
   $self->{file} = "$DATA_HOME/$1.d";
   $self->{fh} = aio_open $self->{file}, O_RDWR|O_CREAT, 0600
      or die "$self->{file}: $!";

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

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

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

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

   aio_unlink $self->{file};

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

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

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

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

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

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

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

}

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

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

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

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

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

      my @segments;
      my $segments = 0;

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

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

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

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

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

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

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

                  $offset2 += $blksize;
               }
            }
         }

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

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

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

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

      my @txn;

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

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

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

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

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

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

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

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

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

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

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

               $sig->send;
            };
         }

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

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

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

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

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

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

         last unless $segments;

         $sig->wait;
      }

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

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

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

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

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

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

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

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

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

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

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

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

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

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

package main;

$|=1;

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