3 // get_peers and announce_peers.
9 "github.com/anacrolix/missinggo"
10 "github.com/anacrolix/sync"
11 "github.com/willf/bloom"
13 "github.com/anacrolix/torrent/logonce"
16 // Maintains state for an ongoing Announce operation. An Announce is started
17 // by calling Server.Announce.
18 type Announce struct {
20 Peers chan PeersValues
21 // Inner chan is set to nil when on close.
22 values chan PeersValues
24 triedAddrs *bloom.BloomFilter
30 announcePortImplied bool
33 // Returns the number of distinct remote addresses the announce has queried.
34 func (me *Announce) NumContacted() int {
37 return me.numContacted
40 // This is kind of the main thing you want to do with DHT. It traverses the
41 // graph toward nodes that store peers for the infohash, streaming them to the
42 // caller, and announcing the local node to each node if allowed and
44 func (s *Server) Announce(infoHash string, port int, impliedPort bool) (*Announce, error) {
46 startAddrs := func() (ret []dHTAddr) {
47 for _, n := range s.closestGoodNodes(160, infoHash) {
48 ret = append(ret, n.addr)
53 if len(startAddrs) == 0 {
54 addrs, err := bootstrapAddrs(s.bootstrapNodes)
58 for _, addr := range addrs {
59 startAddrs = append(startAddrs, newDHTAddr(addr))
63 Peers: make(chan PeersValues, 100),
64 stop: make(chan struct{}),
65 values: make(chan PeersValues),
66 triedAddrs: bloom.NewWithEstimates(1000, 0.5),
70 announcePortImplied: impliedPort,
72 // Function ferries from values to Values until discovery is halted.
74 defer close(disc.Peers)
77 case psv := <-disc.values:
79 case disc.Peers <- psv:
88 for i, addr := range startAddrs {
90 time.Sleep(time.Millisecond)
99 func (me *Announce) gotNodeAddr(addr dHTAddr) {
100 if missinggo.AddrPort(addr) == 0 {
101 // Not a contactable address.
104 if me.triedAddrs.Test([]byte(addr.String())) {
107 if me.server.ipBlocked(addr.UDPAddr().IP) {
111 if me.server.badNodes.Test([]byte(addr.String())) {
112 me.server.mu.Unlock()
115 me.server.mu.Unlock()
119 func (me *Announce) contact(addr dHTAddr) {
121 me.triedAddrs.Add([]byte(addr.String()))
122 if err := me.getPeers(addr); err != nil {
123 log.Printf("error sending get_peers request to %s: %#v", addr, err)
129 func (me *Announce) transactionClosed() {
137 func (me *Announce) responseNode(node NodeInfo) {
138 me.gotNodeAddr(node.Addr)
141 func (me *Announce) closingCh() chan struct{} {
145 func (me *Announce) announcePeer(to dHTAddr, token string) {
147 err := me.server.announcePeer(to, me.infoHash, me.announcePort, token, me.announcePortImplied)
148 me.server.mu.Unlock()
150 logonce.Stderr.Printf("error announcing peer: %s", err)
154 func (me *Announce) getPeers(addr dHTAddr) error {
156 defer me.server.mu.Unlock()
157 t, err := me.server.getPeers(addr, me.infoHash)
161 t.SetResponseHandler(func(m Msg) {
162 // Register suggested nodes closer to the target info-hash.
164 for _, n := range m.Nodes() {
169 if vs := m.Values(); vs != nil {
170 for _, cp := range vs {
173 me.server.badNode(addr)
174 me.server.mu.Unlock()
178 nodeInfo := NodeInfo{
181 copy(nodeInfo.ID[:], m.SenderID())
183 case me.values <- PeersValues{
191 if at, ok := m.AnnounceToken(); ok {
192 me.announcePeer(addr, at)
196 me.transactionClosed()
202 // Corresponds to the "values" key in a get_peers KRPC response. A list of
203 // peers that a node has reported as being in the swarm for a queried info
205 type PeersValues struct {
206 Peers []Peer // Peers given in get_peers response.
207 NodeInfo // The node that gave the response.
210 // Stop the announce.
211 func (me *Announce) Close() {
217 func (ps *Announce) close() {