Redo the compact peer types

Author: anacrolix

client.go

@@ -35,12 +35,10 @@ import (
 	"github.com/anacrolix/torrent/dht"
 	"github.com/anacrolix/torrent/internal/pieceordering"
 	"github.com/anacrolix/torrent/iplist"
-	"github.com/anacrolix/torrent/logonce"
 	"github.com/anacrolix/torrent/metainfo"
 	"github.com/anacrolix/torrent/mse"
 	pp "github.com/anacrolix/torrent/peer_protocol"
 	"github.com/anacrolix/torrent/tracker"
-	. "github.com/anacrolix/torrent/util"
 )
 
 var (
@@ -1403,12 +1401,6 @@ func (cl *Client) gotMetadataExtensionMsg(payload []byte, t *torrent, c *connect
 	return
 }
 
-type peerExchangeMessage struct {
-	Added      CompactPeers `bencode:"added"`
-	AddedFlags []byte       `bencode:"added.f"`
-	Dropped    CompactPeers `bencode:"dropped"`
-}
-
 // Extracts the port as an integer from an address string.
 func addrPort(addr net.Addr) int {
 	return AddrPort(addr)
@@ -2434,8 +2426,8 @@ newAnnounce:
 			for trIndex, tr := range tier {
 				numTrackersTried++
 				err := cl.announceTorrentSingleTracker(tr, &req, t)
-				if err != nil {
-					logonce.Stderr.Printf("%s: error announcing to %s: %s", t, tr, err)
+				if err != nil && missinggo.CryHeard() {
+					log.Printf("%s: error announcing to %s: %s", t, tr, err)
 					continue
 				}
 				// Float the successful announce to the top of the tier. If

dht/announce.go

@@ -11,7 +11,6 @@ import (
 	"github.com/willf/bloom"
 
 	"github.com/anacrolix/torrent/logonce"
-	"github.com/anacrolix/torrent/util"
 )
 
 // Maintains state for an ongoing Announce operation. An Announce is started
@@ -204,8 +203,8 @@ func (me *Announce) getPeers(addr dHTAddr) error {
 // peers that a node has reported as being in the swarm for a queried info
 // hash.
 type PeersValues struct {
-	Peers    []util.CompactPeer // Peers given in get_peers response.
-	NodeInfo                    // The node that gave the response.
+	Peers    []Peer // Peers given in get_peers response.
+	NodeInfo        // The node that gave the response.
 }
 
 // Stop the announce.

dht/dht.go

@@ -18,6 +18,7 @@ import (
 	"math/rand"
 	"net"
 	"os"
+	"strconv"
 	"time"
 
 	"github.com/anacrolix/missinggo"
@@ -1028,19 +1029,25 @@ func (s *Server) findNode(addr dHTAddr, targetID string) (t *Transaction, err er
 	return
 }
 
+type Peer struct {
+	IP   net.IP
+	Port int
+}
+
+func (me *Peer) String() string {
+	return net.JoinHostPort(me.IP.String(), strconv.FormatInt(int64(me.Port), 10))
+}
+
 // In a get_peers response, the addresses of torrent clients involved with the
 // queried info-hash.
-func (m Msg) Values() (vs []util.CompactPeer) {
-	r, ok := m["r"]
-	if !ok {
-		return
-	}
-	rd, ok := r.(map[string]interface{})
-	if !ok {
-		return
-	}
-	v, ok := rd["values"]
-	if !ok {
+func (m Msg) Values() (vs []Peer) {
+	v := func() interface{} {
+		defer func() {
+			recover()
+		}()
+		return m["r"].(map[string]interface{})["values"]
+	}()
+	if v == nil {
 		return
 	}
 	vl, ok := v.([]interface{})
@@ -1050,19 +1057,21 @@ func (m Msg) Values() (vs []util.CompactPeer) {
 		}
 		return
 	}
-	vs = make([]util.CompactPeer, 0, len(vl))
+	vs = make([]Peer, 0, len(vl))
 	for _, i := range vl {
 		s, ok := i.(string)
 		if !ok {
 			panic(i)
 		}
+		// Because it's a list of strings, we can let the length of the string
+		// determine the IP version of the compact peer.
 		var cp util.CompactPeer
 		err := cp.UnmarshalBinary([]byte(s))
 		if err != nil {
 			log.Printf("error decoding values list element: %s", err)
 			continue
 		}
-		vs = append(vs, cp)
+		vs = append(vs, Peer{cp.IP[:], int(cp.Port)})
 	}
 	return
 }

pex.go

@@ -0,0 +1,9 @@
+package torrent
+
+import "github.com/anacrolix/torrent/util"
+
+type peerExchangeMessage struct {
+	Added      util.CompactIPv4Peers `bencode:"added"`
+	AddedFlags []byte                `bencode:"added.f"`
+	Dropped    util.CompactIPv4Peers `bencode:"dropped"`
+}

pex_test.go

@@ -0,0 +1,18 @@
+package torrent
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/anacrolix/torrent/bencode"
+)
+
+func TestUnmarshalPex(t *testing.T) {
+	var pem peerExchangeMessage
+	err := bencode.Unmarshal([]byte("d5:added12:\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0ce"), &pem)
+	require.NoError(t, err)
+	require.EqualValues(t, 2, len(pem.Added))
+	require.EqualValues(t, 1286, pem.Added[0].Port)
+	require.EqualValues(t, 0x100*0xb+0xc, pem.Added[1].Port)
+}

tracker/http.go

@@ -43,8 +43,7 @@ func (r *response) UnmarshalPeers() (ret []Peer, err error) {
 		err = fmt.Errorf("unsupported peers value type: %T", r.Peers)
 		return
 	}
-	cp := make(util.CompactPeers, 0, len(s)/6)
-	err = cp.UnmarshalBinary([]byte(s))
+	cp, err := util.UnmarshalIPv4CompactPeers([]byte(s))
 	if err != nil {
 		return
 	}

tracker/server.go

@@ -0,0 +1,115 @@
+package tracker
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"math/rand"
+	"net"
+
+	"github.com/anacrolix/torrent/util"
+)
+
+type torrent struct {
+	Leechers int32
+	Seeders  int32
+	Peers    util.CompactIPv4Peers
+}
+
+type server struct {
+	pc    net.PacketConn
+	conns map[int64]struct{}
+	t     map[[20]byte]torrent
+}
+
+func marshal(parts ...interface{}) (ret []byte, err error) {
+	var buf bytes.Buffer
+	for _, p := range parts {
+		err = binary.Write(&buf, binary.BigEndian, p)
+		if err != nil {
+			return
+		}
+	}
+	ret = buf.Bytes()
+	return
+}
+
+func (me *server) respond(addr net.Addr, rh ResponseHeader, parts ...interface{}) (err error) {
+	b, err := marshal(append([]interface{}{rh}, parts...)...)
+	if err != nil {
+		return
+	}
+	_, err = me.pc.WriteTo(b, addr)
+	return
+}
+
+func (me *server) newConn() (ret int64) {
+	ret = rand.Int63()
+	if me.conns == nil {
+		me.conns = make(map[int64]struct{})
+	}
+	me.conns[ret] = struct{}{}
+	return
+}
+
+func (me *server) serveOne() (err error) {
+	b := make([]byte, 0x10000)
+	n, addr, err := me.pc.ReadFrom(b)
+	if err != nil {
+		return
+	}
+	r := bytes.NewReader(b[:n])
+	var h RequestHeader
+	err = readBody(r, &h)
+	if err != nil {
+		return
+	}
+	switch h.Action {
+	case Connect:
+		if h.ConnectionId != connectRequestConnectionId {
+			return
+		}
+		connId := me.newConn()
+		err = me.respond(addr, ResponseHeader{
+			Connect,
+			h.TransactionId,
+		}, ConnectionResponse{
+			connId,
+		})
+		return
+	case Announce:
+		if _, ok := me.conns[h.ConnectionId]; !ok {
+			me.respond(addr, ResponseHeader{
+				TransactionId: h.TransactionId,
+				Action:        Error,
+			}, []byte("not connected"))
+			return
+		}
+		var ar AnnounceRequest
+		err = readBody(r, &ar)
+		if err != nil {
+			return
+		}
+		t := me.t[ar.InfoHash]
+		b, err = t.Peers.MarshalBinary()
+		if err != nil {
+			panic(err)
+		}
+		err = me.respond(addr, ResponseHeader{
+			TransactionId: h.TransactionId,
+			Action:        Announce,
+		}, AnnounceResponseHeader{
+			Interval: 900,
+			Leechers: t.Leechers,
+			Seeders:  t.Seeders,
+		}, b)
+		return
+	default:
+		err = fmt.Errorf("unhandled action: %d", h.Action)
+		me.respond(addr, ResponseHeader{
+			TransactionId: h.TransactionId,
+			Action:        Error,
+		}, []byte("unhandled action"))
+		return
+	}
+}

tracker/udp.go

@@ -11,6 +11,8 @@ import (
 	"net/url"
 	"time"
 
+	"github.com/anacrolix/missinggo"
+
 	"github.com/anacrolix/torrent/util"
 )
 
@@ -22,6 +24,8 @@ const (
 	Scrape
 	Error
 
+	connectRequestConnectionId = 0x41727101980
+
 	// BEP 41
 	optionTypeEndOfOptions = 0
 	optionTypeNOP          = 1
@@ -121,34 +125,29 @@ func (c *udpClient) Announce(req *AnnounceRequest) (res AnnounceResponse, err er
 	res.Interval = h.Interval
 	res.Leechers = h.Leechers
 	res.Seeders = h.Seeders
-	for {
-		var p util.CompactPeer
-		err = binary.Read(b, binary.BigEndian, &p)
-		switch err {
-		case nil:
-		case io.EOF:
-			err = nil
-			fallthrough
-		default:
-			return
-		}
+	cps, err := util.UnmarshalIPv4CompactPeers(b.Bytes())
+	if err != nil {
+		return
+	}
+	for _, cp := range cps {
 		res.Peers = append(res.Peers, Peer{
-			IP:   p.IP[:],
-			Port: int(p.Port),
+			IP:   cp.IP[:],
+			Port: int(cp.Port),
 		})
 	}
+	return
 }
 
 // body is the binary serializable request body. trailer is optional data
 // following it, such as for BEP 41.
 func (c *udpClient) write(h *RequestHeader, body interface{}, trailer []byte) (err error) {
-	buf := &bytes.Buffer{}
-	err = binary.Write(buf, binary.BigEndian, h)
+	var buf bytes.Buffer
+	err = binary.Write(&buf, binary.BigEndian, h)
 	if err != nil {
 		panic(err)
 	}
 	if body != nil {
-		err = binary.Write(buf, binary.BigEndian, body)
+		err = binary.Write(&buf, binary.BigEndian, body)
 		if err != nil {
 			panic(err)
 		}
@@ -177,7 +176,7 @@ func write(w io.Writer, data interface{}) error {
 
 // args is the binary serializable request body. trailer is optional data
 // following it, such as for BEP 41.
-func (c *udpClient) request(action Action, args interface{}, options []byte) (responseBody *bytes.Reader, err error) {
+func (c *udpClient) request(action Action, args interface{}, options []byte) (responseBody *bytes.Buffer, err error) {
 	tid := newTransactionId()
 	err = c.write(&RequestHeader{
 		ConnectionId:  c.connectionId,
@@ -218,12 +217,12 @@ func (c *udpClient) request(action Action, args interface{}, options []byte) (re
 		if h.Action == Error {
 			err = errors.New(buf.String())
 		}
-		responseBody = bytes.NewReader(buf.Bytes())
+		responseBody = buf
 		return
 	}
 }
 
-func readBody(r *bytes.Reader, data ...interface{}) (err error) {
+func readBody(r io.Reader, data ...interface{}) (err error) {
 	for _, datum := range data {
 		err = binary.Read(r, binary.BigEndian, datum)
 		if err != nil {
@@ -241,9 +240,14 @@ func (c *udpClient) Connect() (err error) {
 	if c.connected() {
 		return nil
 	}
-	c.connectionId = 0x41727101980
+	c.connectionId = connectRequestConnectionId
 	if c.socket == nil {
-		c.socket, err = net.Dial("udp", c.url.Host)
+		hmp := missinggo.SplitHostPort(c.url.Host)
+		if hmp.NoPort {
+			hmp.NoPort = false
+			hmp.Port = 80
+		}
+		c.socket, err = net.Dial("udp", hmp.String())
 		if err != nil {
 			return
 		}