view release on metacpan or  search on metacpan

lib/AnyEvent/BitTorrent.pm  view on Meta::CPAN

sub _build_files {
    my $s = shift;
    defined $s->metadata->{info}{files} ?
        [
        map {
            {priority => 1,
             fh       => undef,
             mode     => 'c',
             length   => $_->{length},
             timeout  => undef,
             path =>
                 File::Spec->rel2abs(
                     File::Spec->catfile($s->basedir, $s->name, @{$_->{path}})
                 )
            }
        } @{$s->metadata->{info}{files}}
        ]
        : [
          {priority => 1,
           fh       => undef,
           mode     => 'c',
           length   => $s->metadata->{info}{length},
           timeout  => undef,
           path =>
               File::Spec->rel2abs(File::Spec->catfile($s->basedir, $s->name))
          }
        ];
}
has size => (is       => 'ro',
             lazy     => 1,
             builder  => '_build_size',
             isa      => Int,
             init_arg => undef
);

sub _build_size {
    my $s   = shift;
    my $ret = 0;
    $ret += $_->{length} for @{$s->files};
    AE::log debug => '_build_size() => ' . $ret;
    $ret;
}

sub _open {
    my ($s, $i, $m) = @_;
    AE::log
        trace => 'Opening file #%d (%s) for %s',
        $i, $s->files->[$i]->{path}, $m;
    return 1 if $s->files->[$i]->{mode} eq $m;
    if (defined $s->files->[$i]->{fh}) {
        AE::log trace => 'Closing %s', $s->files->[$i]->{fh};
        flock $s->files->[$i]->{fh}, LOCK_UN;
        close $s->files->[$i]->{fh};
        $s->files->[$i]->{fh} = ();
    }
    if ($m eq 'r') {
        AE::log trace => 'Opening %s to read', $s->files->[$i]->{path};
        sysopen($s->files->[$i]->{fh}, $s->files->[$i]->{path}, O_RDONLY)
            || return;
        flock($s->files->[$i]->{fh}, LOCK_SH) || return;
        weaken $s unless isweak $s;
        my $x = $i;
        $s->files->[$x]->{timeout}
            = AE::timer(500, 0, sub { $s // return; $s->_open($x, 'c') });
    }
    elsif ($m eq 'w') {
        AE::log trace => 'Opening %s to write', $s->files->[$i]->{path};
        my @split = File::Spec->splitdir($s->files->[$i]->{path});
        pop @split;    # File name itself
        my $dir = File::Spec->catdir(@split);
        File::Path::mkpath($dir) if !-d $dir;
        sysopen($s->files->[$i]->{fh},
                $s->files->[$i]->{path},
                O_WRONLY | O_CREAT)
            || return;
        flock $s->files->[$i]->{fh}, LOCK_EX;
        truncate $s->files->[$i]->{fh}, $s->files->[$i]->{length}
            if -s $s->files->[$i]->{fh}
            != $s->files->[$i]->{length};    # XXX - pre-allocate files
        weaken $s unless isweak $s;
        my $x = $i;
        $s->files->[$x]->{timeout}
            = AE::timer(60, 0, sub { $s // return; $s->_open($x, 'c') });
    }
    elsif ($m eq 'c') { $s->files->[$i]->{timeout} = () }
    else              {return}
    return $s->files->[$i]->{mode} = $m;
}
has piece_cache => (is => 'ro', isa => HashRef, default => sub { {} });

sub _cache_path {
    my $s = shift;
    File::Spec->catfile($s->basedir,
                        (scalar @{$s->files} == 1 ? () : $s->name),
                        '~ABPartFile_-'
                            . uc(substr(unpack('H*', $s->infohash), 0, 10))
                            . '.dat'
    );
}

sub _write_cache {
    my ($s, $f, $o, $d) = @_;
    my $path = $s->_cache_path;
    AE::log
        debug =>
        'Attempting to store %d bytes to cache file (%s) [$f=%s, $o=%s]',
        length($d), $path, $f, $o;
    my @split = File::Spec->splitdir($path);
    pop @split;    # File name itself
    my $dir = File::Spec->catdir(@split);
    File::Path::mkpath($dir) if !-d $dir;
    sysopen(my ($fh), $path, O_WRONLY | O_CREAT)
        || return;
    flock $fh, LOCK_EX;
    my $pos = sysseek $fh, 0, SEEK_CUR;
    my $w = syswrite $fh, $d;
    flock $fh, LOCK_UN;
    close $fh;
    $s->piece_cache->{$f}{$o} = $pos;
    AE::log debug => 'Wrote %d bytes to cache file', $w;
    return $w;
}

sub _read_cache {
    my ($s, $f, $o, $l) = @_;
    $s->piece_cache->{$f} // return;
    $s->piece_cache->{$f}{$o} // return;
    my $path = $s->_cache_path;
    AE::log
        debug =>
        'Attempting to read %d bytes from cache file (%s) [$f=%s, $o=%s]',
        $l, $path, $f, $o;
    sysopen(my ($fh), $path, O_RDONLY)
        || return;
    flock $fh, LOCK_SH;
    sysseek $fh, $s->piece_cache->{$f}{$o}, SEEK_SET;
    my $w = sysread $fh, my ($d), $l;
    flock $fh, LOCK_UN;
    close $fh;
    return $d;
}

sub _read {
    my ($s, $index, $offset, $length) = @_;
    AE::log
        debug =>
        'Attempting to read %d bytes from piece %d starting at %d bytes',
        $length, $index, $offset;
    my $data         = '';
    my $file_index   = 0;
    my $total_offset = ($index * $s->piece_length) + $offset;
SEARCH:
    while ($total_offset > $s->files->[$file_index]->{length}) {
        $total_offset -= $s->files->[$file_index]->{length};
        $file_index++;
        AE::log
            trace =>
            'Searching for location... $total_offset = %d, $file_index = %d',
            $total_offset, $file_index;
        last SEARCH    # XXX - return?
            if not defined $s->files->[$file_index]->{length};
    }
READ: while ((defined $length) && ($length > 0)) {
        my $this_read
            = (
              ($total_offset + $length) >= $s->files->[$file_index]->{length})
            ?
            ($s->files->[$file_index]->{length} - $total_offset)
            : $length;
        AE::log
            trace =>
            'Attempting to read %d bytes from file #%d (%s), starting at %d',
            $this_read,
            $file_index, $s->files->[$file_index]->{path}, $total_offset;
        if (   (!-f $s->files->[$file_index]->{path})
            || (!$s->_open($file_index, 'r')))
        {   $data .= $s->_read_cache($file_index, $total_offset, $this_read)
                // ("\0" x $this_read);
            AE::log note => 'Failed to open file. Using null chars instead.';
        }
        else {
            sysseek $s->files->[$file_index]->{fh}, $total_offset, SEEK_SET;
            sysread $s->files->[$file_index]->{fh}, my ($_data), $this_read;
            $data .= $_data if $_data;
            AE::log
                trace =>
                'Read %d bytes of data from file (%d bytes collected so far)',
                length $_data, length $data;
            weaken $s unless isweak $s;
            my $x = $file_index;
            $s->files->[$x]->{timeout}
                = AE::timer(500, 0, sub { $s // return; $s->_open($x, 'c') });
        }
        $file_index++;
        $length -= $this_read;
        AE::log
            trace => 'Still need to read %d bytes',
            $length;
        last READ if not defined $s->files->[$file_index];
        $total_offset = 0;
    }
    AE::log trace => 'Returning %d bytes of data', length $data;
    return $data;
}

sub _write {
    my ($s, $index, $offset, $data) = @_;
    AE::log
        debug =>
        'Attempting to write %d bytes from piece %d starting at %d bytes',
        length($data), $index, $offset;
    my $file_index = 0;
    my $total_offset = int(($index * $s->piece_length) + ($offset || 0));
    AE::log
        debug => '...calculated offset == %d',
        $total_offset;
SEARCH:
    while ($total_offset > $s->files->[$file_index]->{length}) {
        $total_offset -= $s->files->[$file_index]->{length};
        $file_index++;
        AE::log
            trace =>
            'Searching for location... $total_offset = %d, $file_index = %d',
            $total_offset, $file_index;
        last SEARCH    # XXX - return?
            if not defined $s->files->[$file_index]->{length};
    }
WRITE: while ((defined $data) && (length $data > 0)) {
        my $this_write
            = (($total_offset + length $data)
               >= $s->files->[$file_index]->{length})
            ?
            ($s->files->[$file_index]->{length} - $total_offset)
            : length $data;
        AE::log
            trace =>
            'Attempting to write %d bytes from file #%d (%s), starting at %d',
            $this_write,
            $file_index, $s->files->[$file_index]->{path}, $total_offset;
        if ($s->files->[$file_index]->{priority} == 0) {
            $s->_write_cache($file_index, $total_offset, substr $data, 0,
                             $this_write, '');
            AE::log trace => 'Wrote data to cache...';
        }
        else {
            $s->_open($file_index, 'w');
            sysseek $s->files->[$file_index]->{fh}, $total_offset, SEEK_SET;
            my $w = syswrite $s->files->[$file_index]->{fh}, substr $data, 0,
                $this_write, '';
            AE::log
                trace => 'Wrote %d bytes of data to file (%d bytes left)',
                $w, length $data;
            weaken $s unless isweak $s;
            my $x = $file_index;
            $s->files->[$x]->{timeout}
                = AE::timer(120, 0, sub { $s // return; $s->_open($x, 'c') });
        }
        $file_index++;
        last WRITE if not defined $s->files->[$file_index];
        $total_offset = 0;
    }
    return length $data;
}

sub hashcheck (;@) {
    my $s = shift;
    my @indexes = @_ ? @_ : (0 .. $s->piece_count);
    AE::log trace => sub {
        require Data::Dump;
        'Hashcheck of : ' . Data::Dump::dump(\@indexes);
    };
    $s->bitfield;    # Makes sure it's built
    my $total_size = $s->size;
    for my $index (@indexes) {
        next if $index < 0 || $index > $s->piece_count;
        my $piece = $s->_read($index,
                              0,
                              $index == $s->piece_count
                              ?
                                  $total_size % $s->piece_length
                              : $s->piece_length
        );
        my $expected = substr($s->pieces, $index * 20, 20);
        my $reality  = sha1($piece);
        my $ok       = defined($piece)
            && ($expected eq $reality);
        vec($s->{bitfield}, $index, 1) = $ok;
        AE::log trace => sub {
            "Validate piece #%06d %s, Expected: %s\n"
                . "                         Reality:  %s",
                $index, ($ok ? 'PASS' : 'FAIL'), unpack('H*', $expected),
                unpack('H*', $reality);
        };
        $ok ?
            $s->_trigger_hash_pass($index)
            : $s->_trigger_hash_fail($index);
    }
}
has peers => (is      => 'ro',
              lazy    => 1,
              isa     => HashRef,
              clearer => '_clear_peers',
              builder => '_build_peers'
);
sub _build_peers { {} }

sub _add_peer {
    my ($s, $h) = @_;
    $s->{peers}{+$h} = {
        handle            => $h,
        peerid            => '',
        bitfield          => (pack 'b*', "\0" x $s->piece_count),
        remote_choked     => 1,

lib/AnyEvent/BitTorrent.pm  view on Meta::CPAN

        elsif ($s->bitfield() !~ m[[^\0]]) {
            return $s->_send_encrypted($h, build_havenone());
        }
    }

    # XXX - If it's cheaper to send HAVE packets than a full BITFIELD, do it
    $s->_send_encrypted($h, build_bitfield($s->bitfield));
}

sub _broadcast {
    my ($s, $data, $qualifier) = @_;
    $qualifier //= sub {1};
    $s->_send_encrypted($_->{handle}, $data)
        for grep { $qualifier->() } values %{$s->peers};
}

sub _consider_peer {    # Figure out whether or not we find a peer interesting
    my ($s, $p) = @_;
    return if $s->state ne 'active';
    return if $s->complete;
    my $relevence = $p->{bitfield} & $s->wanted;
    my $interesting
        = (
         index(substr(unpack('b*', $relevence), 0, $s->piece_count + 1), 1, 0)
             != -1) ? 1 : 0;
    if ($interesting) {
        if (!$p->{local_interested}) {
            $p->{local_interested} = 1;
            $s->_send_encrypted($p->{handle}, build_interested());
        }
    }
    else {
        if ($p->{local_interested}) {
            $p->{local_interested} = 0;
            $s->_send_encrypted($p->{handle}, build_not_interested());
        }
    }
}
has working_pieces => (is       => 'ro',
                       lazy     => 1,
                       isa      => HashRef,
                       init_arg => undef,
                       default  => sub { {} }
);

sub _file_to_range {
    my ($s, $file) = @_;
    my $start = 0;
    for (0 .. $file - 1) {
        $start += $s->files->[$_]->{length};
    }
    my $end = $start + $s->files->[$file]->{length};
    $start = $start / $s->piece_length;
    $end   = $end / $s->piece_length;
    (int($start) .. int $end + ($end != int($end) ? 0 : +1));
}

sub _request_pieces {
    my ($s, $p) = @_;
    return if $s->state ne 'active';
    weaken $p unless isweak $p;
    $p // return;
    $p->{handle} // return;
    my @indexes;
    if (scalar keys %{$s->working_pieces} < 10) {   # XXX - Max working pieces
        for my $findex (0 .. $#{$s->files}) {
            for my $index ($s->_file_to_range($findex)) {
                next
                    if !(vec($p->{bitfield}, $index, 1)
                         && !vec($s->bitfield, $index, 1));
                push @indexes,
                    map {$index} 1 .. $s->{files}[$findex]{priority};
            }
        }
    }
    else {
        @indexes = keys %{$s->working_pieces};
    }
    return if !@indexes;
    my $index = $indexes[rand @indexes];  # XXX - Weighted random/Rarest first
    my $piece_size
        = $index == $s->piece_count ?
        $s->size % $s->piece_length
        : $s->piece_length;
    my $block_count = $piece_size / $block_size;
    my @offsets = map { $_ * $block_size }
        0 .. $block_count - ((int($block_count) == $block_count) ? 1 : 0);
    $s->working_pieces->{$index} //= {map { $_ => {} } @offsets};
    my @unrequested = sort { $a <=> $b }
        grep {    # XXX - If there are no unrequested blocks, pick a new index
        (!ref $s->working_pieces->{$index}{$_})
            || (   (!defined $s->working_pieces->{$index}{$_}[4])
                && (!defined $s->working_pieces->{$index}{$_}[3]))
        } @offsets;
    my @unfilled_local_requests
        = grep { !defined $_->[4] } @{$p->{local_requests}};
    for (scalar @unfilled_local_requests .. 12) {
        my $offset = shift @unrequested;
        $offset // return;    # XXX - Start working on another piece
        my $_block_size
            = ($offset == $offsets[-1]) ?
            ($piece_size % $block_size) || $block_size
            : $block_size;
        next if !$_block_size;

        # XXX - Limit to x req per peer (future: based on bandwidth)
        AE::log
            trace => 'Requesting %d, %d, %d',
            $index, $offset, $_block_size;
        $s->_send_encrypted($p->{handle},
                            build_request($index, $offset, $_block_size))
            ;    # XXX - len for last piece
        $s->working_pieces->{$index}{$offset} = [
            $index, $offset,
            $_block_size,
            $p,     undef,
            AE::timer(
                60, 0,
                sub {
                    $p // return;
                    $p->{handle} // return;
                    $s->_send_encrypted($p->{handle},
                                 build_cancel($index, $offset, $_block_size));
                    $s->working_pieces->{$index}{$offset}[3] = ();
                    $p->{local_requests} = [
                        grep {
                                   $_->[0] != $index
                                || $_->[1] != $offset
                                || $_->[2] != $_block_size
                        } @{$p->{local_requests}}
                    ];
                    $p->{timeout} = AE::timer(45, 0,
                                         sub { $s->_del_peer($p->{handle}) });

                    #$s->_request_pieces( $p) #  XXX - Ask a different peer
                }
            )
        ];
        weaken($s->working_pieces->{$index}{$offset}[3])
            unless isweak($s->working_pieces->{$index}{$offset}[3]);
        push @{$p->{local_requests}}, [$index, $offset, $_block_size];
    }
}

# Cheap callback system
has on_hash_pass => (
    is      => 'rw',
    isa     => CodeRef,
    default => sub {
        sub { !!1 }
    },
    clearer => '_no_hash_pass'
);
sub _trigger_hash_pass { shift->on_hash_pass()->(@_) }
has on_hash_fail => (
    is      => 'rw',
    isa     => CodeRef,
    default => sub {
        sub { !!1 }
    },
    clearer => '_no_hash_fail'
);
sub _trigger_hash_fail { shift->on_hash_fail()->(@_) }
#
has state => (is      => 'ro',
              isa     => Enum [qw[active stopped paused]],
              writer  => '_set_state',
              default => sub {'active'}
);

sub stop {
    my $s = shift;
    AE::log debug => 'Stopping...';
    return if $s->state eq 'stopped';
    AE::log trace => 'Announcing "stopped" event to trackers...';
    $s->announce('stopped');
    AE::log trace => 'Disconnecting peers...';
    $s->_clear_peers;
    AE::log trace => 'Stopping new peers ticker...';
    $s->_clear_peer_timer;
    AE::log trace => 'Closing files...';
    $s->_open($_, 'c') for 0 .. $#{$s->files};
    AE::log trace => 'Setting internal status...';
    $s->_set_state('stopped');
}

sub start {
    my $s = shift;
    AE::log debug => 'Starting...';
    $s->announce('started') unless $s->state eq 'active';
    $s->peers;
    AE::log trace => 'Starting new peers ticker...';
    $s->_peer_timer;
    AE::log trace => 'Setting internal status...';
    $s->_set_state('active');
}

sub pause {
    my $s = shift;