view release on metacpan or  search on metacpan

script/torrent  view on Meta::CPAN

            ? ($global_offset + $written) : $fstart;
        my $write_end = ($global_offset + $data_len < $fend)
            ? ($global_offset + $data_len) : $fend;
        my $file_off = $write_start - $fstart;
        my $data_off = $write_start - $global_offset;
        my $wlen     = $write_end - $write_start;

        my $fh = $file_handles->[$i];
        sysseek($fh, $file_off, 0);
        syswrite($fh, $piece_data, $wlen, $data_off);

        $written = $write_end - $global_offset;
    }
}

# ─── Formatting ───────────────────────────────────────────────────────────────

sub format_size {
    my ($b) = @_;
    return sprintf("%.2f GB", $b / 1073741824) if $b >= 1073741824;
    return sprintf("%.2f MB", $b / 1048576)    if $b >= 1048576;
    return sprintf("%.2f KB", $b / 1024)       if $b >= 1024;
    return "$b B";
}

# ─── Thread-Safe Shared State ────────────────────────────────────────────────

my @g_have    :shared;    # 1 if piece is verified and written
my @g_pending :shared;    # 1 if piece is being downloaded by a thread
my $g_done    :shared = 0;
my $g_dl      :shared = 0;    # bytes downloaded
my $g_lock    :shared;        # mutex for piece selection

# ─── Peer Worker (runs in its own thread) ────────────────────────────────────

sub peer_worker {
    my ($ip, $port, $info_hash, $peer_id, $piece_length, $total_size,
        $piece_hashes, $num_pieces, $files, $output_dir, $start_time) = @_;

    my $tag = "$ip:$port";

    # Early exit if download is already complete
    { lock($g_lock); return if $g_done >= $num_pieces; }

    # Each thread opens its own file handles for writing
    my @fhs;
    for my $file (@$files) {
        my $path = "$output_dir/$file->{path}";
        open my $fh, '+<:raw', $path or do {
            warn "  [$tag] Cannot open $path: $!\n";
            return;
        };
        push @fhs, $fh;
    }

    # Connect
    my $socket = IO::Socket::INET->new(
        PeerAddr => $ip,
        PeerPort => $port,
        Proto    => 'tcp',
        Timeout  => CONNECT_TIMEOUT,
    );
    unless ($socket) {
        print "  [$tag] connect failed\n";
        close $_ for @fhs;
        return;
    }
    binmode($socket);

    # Handshake
    my $rpid = do_handshake($socket, $info_hash, $peer_id);
    unless ($rpid) {
        print "  [$tag] handshake failed\n";
        close $socket;
        close $_ for @fhs;
        return;
    }
    print "  [$tag] connected\n";

    # Per-peer state
    my $peer_choking = 1;
    my $peer_bf      = "\0" x ceil($num_pieces / 8);

    # Send interested
    send_msg($socket, MSG_INTERESTED, '');

    # Wait for unchoke (up to 50 messages)
    eval {
        for (1 .. 50) {
            last unless $peer_choking;
            my $msg = read_msg($socket);
            last unless defined $msg;
            my ($id, $payload) = @$msg;
            next unless defined $id;

            if ($id == MSG_UNCHOKE) {
                $peer_choking = 0;
            } elsif ($id == MSG_BITFIELD) {
                $peer_bf = $payload;
            } elsif ($id == MSG_HAVE) {
                if (length($payload) >= 4) {
                    my $idx = unpack('N', $payload);
                    bf_set(\$peer_bf, $idx);
                }
            }
        }
    };

    if ($peer_choking) {
        print "  [$tag] never unchoked\n";
        close $socket;
        close $_ for @fhs;
        return;
    }

    # Download loop
    eval {
        while (1) {
            # Check if all pieces are done
            { lock($g_lock); last if $g_done >= $num_pieces; }