diff --git a/README.md b/README.md index 95908020..342ee267 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ - [ ] BitTorrent下载实现 - [x] .torrent文件解析 - [x] tracker协议实现 - - [ ] peer wire protocol协议实现 + - [x] peer wire protocol协议实现 - [ ] DHT协议实现 - [ ] 磁力链接支持 - [ ] uTP协议实现 diff --git a/go.mod b/go.mod index 294d22c2..7351aa56 100644 --- a/go.mod +++ b/go.mod @@ -6,4 +6,5 @@ require ( github.com/RoaringBitmap/roaring v0.4.21 github.com/cenkalti/mse v0.0.0-20140930130441-6ef65f170972 github.com/marksamman/bencode v0.0.0-20150821143521-dc84f26e086e + github.com/sirupsen/logrus v1.4.2 ) diff --git a/go.sum b/go.sum index d334cd5d..ac334005 100644 --- a/go.sum +++ b/go.sum @@ -4,6 +4,8 @@ github.com/cenkalti/mse v0.0.0-20140930130441-6ef65f170972 h1:FvAasVrPm/VUO8M0od github.com/cenkalti/mse v0.0.0-20140930130441-6ef65f170972/go.mod h1:wGf99xKTJ8f/a8cC1inMqFz9cGEzISLLid5F6i5fI/g= github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/glycerine/go-unsnap-stream v0.0.0-20181221182339-f9677308dec2 h1:Ujru1hufTHVb++eG6OuNDKMxZnGIvF6o/u8q/8h2+I4= github.com/glycerine/go-unsnap-stream v0.0.0-20181221182339-f9677308dec2/go.mod h1:/20jfyN9Y5QPEAprSgKAUr+glWDY39ZiUEAYOEv5dsE= github.com/glycerine/goconvey v0.0.0-20190410193231-58a59202ab31 h1:gclg6gY70GLy3PbkQ1AERPfmLMMagS60DKF78eWwLn8= @@ -14,6 +16,8 @@ github.com/gopherjs/gopherjs v0.0.0-20190910122728-9d188e94fb99 h1:twflg0XRTjwKp github.com/gopherjs/gopherjs v0.0.0-20190910122728-9d188e94fb99/go.mod h1:wJfORRmW1u3UXTncJ5qlYoELFm8eSnnEO6hX4iZ3EWY= github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo= github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU= +github.com/konsorten/go-windows-terminal-sequences v1.0.1 h1:mweAR1A6xJ3oS2pRaGiHgQ4OO8tzTaLawm8vnODuwDk= +github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/marksamman/bencode v0.0.0-20150821143521-dc84f26e086e h1:KMs6SK8iDSR1+ZzOK10L5wGPpWDByyvOe5nrqk51g2U= github.com/marksamman/bencode v0.0.0-20150821143521-dc84f26e086e/go.mod h1:+AHfJo5+69p+fjvMJTmYajNP9rFBHQcaTDFmuXRRATI= github.com/mschoch/smat v0.0.0-20160514031455-90eadee771ae h1:VeRdUYdCw49yizlSbMEn2SZ+gT+3IUKx8BqxyQdz+BY= @@ -22,13 +26,19 @@ github.com/philhofer/fwd v1.0.0 h1:UbZqGr5Y38ApvM/V/jEljVxwocdweyH+vmYvRPBnbqQ= github.com/philhofer/fwd v1.0.0/go.mod h1:gk3iGcWd9+svBvR0sR+KPcfE+RNWozjowpeBVG3ZVNU= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/sirupsen/logrus v1.4.2 h1:SPIRibHv4MatM3XXNO2BJeFLZwZ2LvZgfQ5+UNI2im4= +github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk= github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= github.com/tinylib/msgp v1.1.0 h1:9fQd+ICuRIu/ue4vxJZu6/LzxN0HwMds2nq/0cFvxHU= github.com/tinylib/msgp v1.1.0/go.mod h1:+d+yLhGm8mzTaHzB+wgMYrodPfmZrzkirds8fDWklFE= github.com/willf/bitset v1.1.10 h1:NotGKqX0KwQ72NUzqrjZq5ipPNDQex9lo3WpaS8L2sc= github.com/willf/bitset v1.1.10/go.mod h1:RjeCKbqT1RxIR/KWY6phxZiaY1IyutSBfGjNPySAYV4= +golang.org/x/sys v0.0.0-20190422165155-953cdadca894 h1:Cz4ceDQGXuKRnVBDTS23GTn/pU5OE2C0WrNTOYK1Uuc= +golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw= diff --git a/main.go b/main.go index 31489a77..7b95bf0b 100644 --- a/main.go +++ b/main.go @@ -2,10 +2,6 @@ package main import ( "fmt" - "github.com/monkeyWie/gopeed/protocol/bt/metainfo" - "github.com/monkeyWie/gopeed/protocol/bt/peer" - "github.com/monkeyWie/gopeed/protocol/bt/torrent" - "github.com/monkeyWie/gopeed/protocol/bt/tracker" "github.com/monkeyWie/gopeed/protocol/http" ) @@ -24,26 +20,4 @@ func main() { } got, _ := http.Resolve(request) fmt.Println(got) - getUsablePeer() - // webview.Open("Minimal webview example", "https://www.baidu.com", 800, 600, true) -} - -func buildTorrent() *torrent.Torrent { - metaInfo, err := metainfo.ParseFromFile("../testdata/Game.of.Thrones.S08E05.720p.WEB.H264-MEMENTO.torrent") - if err != nil { - panic(err) - } - metaInfo.Announce = "udp://tracker.opentrackr.org:1337/announce" - metaInfo.AnnounceList = [][]string{} - return torrent.NewTorrent(peer.GenPeerID(), metaInfo) -} - -// 获取一个能连接上的peer -func getUsablePeer() { - torrent := buildTorrent() - tracker := &tracker.Tracker{ - PeerID: torrent.PeerID, - MetaInfo: torrent.MetaInfo, - } - tracker.Tracker() } diff --git a/protocol/bt/torrent/peer_conn.go b/protocol/bt/torrent/peer_conn.go index d43e04ab..23a29d9b 100644 --- a/protocol/bt/torrent/peer_conn.go +++ b/protocol/bt/torrent/peer_conn.go @@ -10,6 +10,7 @@ import ( "github.com/monkeyWie/gopeed/protocol/bt/metainfo" "github.com/monkeyWie/gopeed/protocol/bt/peer" "github.com/monkeyWie/gopeed/protocol/bt/peer/message" + log "github.com/sirupsen/logrus" "io" "math" "net" @@ -20,7 +21,9 @@ import ( ) // 现在主流客户端的block大小都是16KB -const BlockSize = 2 << 13 +const blockSize = 2 << 13 + +var errPieceCheckFailed = errors.New("piece check failed") type peerConn struct { torrent *Torrent @@ -35,18 +38,19 @@ type peerConn struct { // peer is interested in this client peerInterested bool - bitfield *message.Bitfield - readyComplete bool - downloadComplete bool + bitfield *message.Bitfield + readyEnd bool + downloadEnd bool downloadedCh chan error + disconnectCh chan error // block个数 blockCount int // block下载队列,官方推荐为5 - blockQueue chan interface{} + blockQueueCh chan interface{} } -func NewPeerConn(torrent *Torrent, peer *peer.Peer) *peerConn { +func newPeerConn(torrent *Torrent, peer *peer.Peer) *peerConn { return &peerConn{ torrent: torrent, peer: peer, @@ -116,10 +120,10 @@ func (pc *peerConn) ready() error { if _, err := pc.handshake(); err != nil { return fmt.Errorf("handshake error %w", err) } - pc.readyComplete = false - pc.downloadComplete = false + pc.readyEnd = false + pc.downloadEnd = false readyCh := make(chan bool) - defer close(readyCh) + pc.disconnectCh = make(chan error) go func() { scanner := bufio.NewScanner(pc.conn) scanner.Split(message.SplitMessage) @@ -156,13 +160,12 @@ func (pc *peerConn) ready() error { } } // 还未下载完成时连接断开 - if pc.downloadedCh != nil && !pc.downloadComplete { - err := scanner.Err() - if err == nil { - err = io.EOF - } - pc.downloadedCh <- err + err := scanner.Err() + if err == nil { + err = io.EOF } + pc.disconnectCh <- err + close(pc.disconnectCh) }() err := func() error { select { @@ -177,24 +180,22 @@ func (pc *peerConn) ready() error { return errors.New("ready time out") } }() - pc.readyComplete = true if err != nil { pc.conn.Close() return err } - pc.blockQueue = make(chan interface{}, 5) + pc.blockQueueCh = make(chan interface{}, 5) return nil } // 下载指定piece func (pc *peerConn) downloadPiece(index int) error { pieceLength := pc.torrent.MetaInfo.GetPieceLength(index) - pc.blockCount = int(math.Ceil(float64(pieceLength) / BlockSize)) + pc.blockCount = int(math.Ceil(float64(pieceLength) / blockSize)) pc.downloadedCh = make(chan error) - defer close(pc.downloadedCh) // 按块下载分片 for i := 0; i < pc.blockCount; i++ { - offset := BlockSize * i + offset := blockSize * i // 如果已经下载过就跳过 if pc.torrent.pieces.isBlockDownloaded(index, offset) { continue @@ -205,36 +206,43 @@ func (pc *peerConn) downloadPiece(index int) error { // 最后一个block大小需要计算出来 blockLength = uint32(pieceLength - offset) } else { - blockLength = BlockSize + blockLength = blockSize } // block下载排队,如果连接出现问题直接返回异常 select { - case pc.blockQueue <- nil: + case pc.blockQueueCh <- nil: break - case err := <-pc.downloadedCh: + case err := <-pc.disconnectCh: return err } // 发起request,对方会响应piece _, err := pc.conn.Write(message.NewRequest(uint32(index), uint32(offset), blockLength).Encode()) if err != nil { - fmt.Println(err) + return err } } - return <-pc.downloadedCh + select { + case err := <-pc.downloadedCh: + return err + case err := <-pc.disconnectCh: + return err + } } func (pc *peerConn) handleUnchoke(readyCh chan<- bool) { pc.peerChoking = false // 已经处理过Unchoke信号 - if pc.readyComplete { + if pc.readyEnd { return } + pc.readyEnd = true // 如果客户端对peer感兴趣并且peer没有choked客户端,就可以开始下载了 if pc.amInterested { readyCh <- true } else { readyCh <- false } + close(readyCh) } func (pc *peerConn) handleBitfield(buf []byte) { @@ -319,9 +327,10 @@ func (pc *peerConn) handlePiece(buf []byte) { }() } pc.torrent.pieces.setBlockDownloaded(int(piece.Index), int(piece.Begin)) - fmt.Printf("piece %d downloaded block %d,count %d\n", piece.Index, piece.Begin, pc.torrent.pieces.blockSize(int(piece.Index))) // 出队 - <-pc.blockQueue + <-pc.blockQueueCh + + log.Debugf("piece:%d block:%d already:%d", piece.Index, piece.Begin, pc.torrent.pieces.blockSize(int(piece.Index))) // piece全部下载完 if pc.torrent.pieces.blockSize(int(piece.Index)) == pc.blockCount { @@ -356,22 +365,23 @@ func (pc *peerConn) handlePiece(buf []byte) { break } } + // 标记下载完成 + pc.downloadEnd = true + // 断开连接 TODO 用连接池进行复用 + pc.conn.Close() + // 校验piece SHA-1 hash downHash := [20]byte{} copy(downHash[:], sha1.Sum(nil)) if downHash == pc.torrent.MetaInfo.Info.Pieces[piece.Index] { - fmt.Printf("piece %d 校验通过\n", piece.Index) + log.Infof("piece %d 校验通过", piece.Index) + // piece下载完成 + pc.downloadedCh <- nil } else { - fmt.Printf("piece %d 校验失败\n", piece.Index) - // 清空该piece上所有的block下载记录 - pc.torrent.pieces.clearBlocks(int(piece.Index)) + log.Infof("piece %d 校验失败", piece.Index) + pc.downloadedCh <- errPieceCheckFailed } - // 标记下载完成 - pc.downloadComplete = true - // 断开连接 TODO 用连接池进行复用 - pc.conn.Close() - // piece下载完成 - pc.downloadedCh <- nil + close(pc.downloadedCh) } } diff --git a/protocol/bt/torrent/peer_conn_test.go b/protocol/bt/torrent/peer_conn_test.go index 31e41c0e..288e4549 100644 --- a/protocol/bt/torrent/peer_conn_test.go +++ b/protocol/bt/torrent/peer_conn_test.go @@ -101,10 +101,7 @@ func getUsablePeerMore() chan *peerConn { PeerID: torrent.PeerID, MetaInfo: torrent.MetaInfo, } - peers, err := tracker.Tracker() - if err != nil { - panic(err) - } + peers := <-tracker.Tracker() ch := make(chan *peerConn) for i := range peers { go peerTest(torrent, &peers[i], ch) @@ -162,5 +159,4 @@ func TestSome2(t *testing.T) { }(i) } wg.Wait() - } diff --git a/protocol/bt/torrent/peer_pool.go b/protocol/bt/torrent/peer_pool.go index ae03637f..beab0508 100644 --- a/protocol/bt/torrent/peer_pool.go +++ b/protocol/bt/torrent/peer_pool.go @@ -2,10 +2,14 @@ package torrent import ( "github.com/monkeyWie/gopeed/protocol/bt/peer" + "github.com/monkeyWie/gopeed/protocol/bt/tracker" "sync" + "time" ) type peerPool struct { + torrent *Torrent + states map[peer.Peer]*peerState lock *sync.Mutex } @@ -17,13 +21,36 @@ type peerState struct { errors int } -func newPeerPool() *peerPool { +func newPeerPool(torrent *Torrent) *peerPool { return &peerPool{ - lock: &sync.Mutex{}, - states: map[peer.Peer]*peerState{}, + torrent: torrent, + lock: &sync.Mutex{}, + states: map[peer.Peer]*peerState{}, } } +func (pp *peerPool) fetch() { + tracker := &tracker.Tracker{ + PeerID: pp.torrent.PeerID, + MetaInfo: pp.torrent.MetaInfo, + } + + go func() { + for { + // 当peer数量少于200个时重新发起一次tracker + if len(pp.states) < 200 { + go func() { + for peers := range tracker.Tracker() { + pp.put(peers) + } + }() + } + // 每5分钟检测一次 + time.Sleep(time.Minute * 5) + } + }() +} + func (pp *peerPool) put(peers []peer.Peer) { pp.lock.Lock() defer pp.lock.Unlock() @@ -50,6 +77,7 @@ func (pp *peerPool) get() *peer.Peer { return nil } +// 将peer重新放回池中,等待使用 func (pp *peerPool) release(peer *peer.Peer) { pp.lock.Lock() defer pp.lock.Unlock() @@ -57,9 +85,11 @@ func (pp *peerPool) release(peer *peer.Peer) { pp.states[*peer].errors = 0 } +// 标记peer为不可用,超过3次则剔除该peer func (pp *peerPool) unavailable(peer *peer.Peer) { pp.lock.Lock() defer pp.lock.Unlock() + pp.states[*peer].using = false pp.states[*peer].errors++ if pp.states[*peer].errors > 3 { delete(pp.states, *peer) diff --git a/protocol/bt/torrent/peer_pool_test.go b/protocol/bt/torrent/peer_pool_test.go deleted file mode 100644 index 64926789..00000000 --- a/protocol/bt/torrent/peer_pool_test.go +++ /dev/null @@ -1,40 +0,0 @@ -package torrent - -import ( - "fmt" - "github.com/monkeyWie/gopeed/protocol/bt/peer" - "testing" -) - -func Test_PeerPool(t *testing.T) { - pool := newPeerPool() - pool.put([]peer.Peer{ - { - IP: 0, - Port: 0, - }, - { - IP: 1, - Port: 0, - }, - { - IP: 2, - Port: 0, - }, - { - IP: 3, - Port: 0, - }, - { - IP: 4, - Port: 0, - }, - }) - get := pool.get() - pool.release(get) - pool.unavailable(get) - for i := 0; i < 5; i++ { - fmt.Println(pool.get()) - } - fmt.Println(pool.get()) -} diff --git a/protocol/bt/torrent/piece_state.go b/protocol/bt/torrent/piece_state.go index 7d507912..e485a205 100644 --- a/protocol/bt/torrent/piece_state.go +++ b/protocol/bt/torrent/piece_state.go @@ -66,6 +66,17 @@ func (ps *pieces) size() int { return len(ps.arr) } +func (ps *pieces) isDone() bool { + ps.rw.RLock() + defer ps.rw.RUnlock() + for _, s := range ps.arr { + if s.state != stateFinished { + return false + } + } + return true +} + // 指定piece下的某个block是否下载完成 func (ps *pieces) isBlockDownloaded(pieceIndex int, blockIndex int) bool { return ps.arr[pieceIndex].blocks.ContainsInt(blockIndex) diff --git a/protocol/bt/torrent/torrent.go b/protocol/bt/torrent/torrent.go index a18713cf..6f69627d 100644 --- a/protocol/bt/torrent/torrent.go +++ b/protocol/bt/torrent/torrent.go @@ -1,9 +1,10 @@ package torrent import ( + "errors" "fmt" "github.com/monkeyWie/gopeed/protocol/bt/metainfo" - "github.com/monkeyWie/gopeed/protocol/bt/tracker" + log "github.com/sirupsen/logrus" "time" ) @@ -11,8 +12,9 @@ type Torrent struct { PeerID [20]byte MetaInfo *metainfo.MetaInfo - pieces *pieces - peerPool *peerPool + pieces *pieces + peerPool *peerPool + completeCh chan bool Path string } @@ -28,64 +30,66 @@ func NewTorrent(peerID [20]byte, metaInfo *metainfo.MetaInfo) *Torrent { func (t *Torrent) Download(path string) { t.Path = path - t.peerPool = newPeerPool() - t.fetchPeers() + t.peerPool = newPeerPool(t) + t.peerPool.fetch() - taskCh := make(chan interface{}, 128) - defer close(taskCh) + pieceQueueCh := make(chan interface{}, 128) + defer close(pieceQueueCh) + t.completeCh = make(chan bool) + defer close(t.completeCh) for { + pieceQueueCh <- nil // 获取一个待下载的piece index := t.pieces.getReady() + // 没有待下载的piece了 if index == -1 { - return + fmt.Println("wait piece") + if <-t.completeCh { + // 下载完成 + break + } else { + // 继续下载 + continue + } } - taskCh <- nil go func(index int) { peer := t.peerPool.get() if peer == nil { - fmt.Println("no peer") time.Sleep(time.Second * 10) t.pieces.setState(index, stateReady) - <-taskCh + <-pieceQueueCh return } - pc := NewPeerConn(t, peer) + pc := newPeerConn(t, peer) + log.Debugf("peer ready start:%s,download %d", peer.Address(), index) err := pc.ready() if err != nil { + log.Debugf("peer ready error:%s,download %d,error:%s", peer.Address(), index, err.Error()) t.pieces.setState(index, stateReady) t.peerPool.unavailable(peer) } else { - fmt.Printf("peer ready:%s,download %d\n", peer.Address(), index) + log.Debugf("peer ready success:%s,download %d", peer.Address(), index) err = pc.downloadPiece(index) if err != nil { - fmt.Println("down piece error:" + err.Error()) // 下载失败 t.pieces.setState(index, stateReady) t.peerPool.unavailable(peer) + if errors.Is(err, errPieceCheckFailed) { + // piece校验失败,需要重新下载过一遍,清空block记录 + t.pieces.clearBlocks(index) + } + t.completeCh <- false } else { // 下载完成 t.pieces.setState(index, stateFinished) t.peerPool.release(peer) + // 检查是否所有piece下载完成 + if t.pieces.isDone() { + t.completeCh <- true + } } } - <-taskCh + <-pieceQueueCh }(index) } } - -// 获取可用的peer -func (t *Torrent) fetchPeers() { - tracker := &tracker.Tracker{ - PeerID: t.PeerID, - MetaInfo: t.MetaInfo, - } - urls := tracker.MetaInfo.AnnounceList[0] - for _, url := range urls { - go func(url string) { - peers, err := tracker.DoTracker(url) - if err == nil { - t.peerPool.put(peers) - } - }(url) - } -} diff --git a/protocol/bt/torrent/torrent_test.go b/protocol/bt/torrent/torrent_test.go index feb799c9..a03ce138 100644 --- a/protocol/bt/torrent/torrent_test.go +++ b/protocol/bt/torrent/torrent_test.go @@ -3,9 +3,15 @@ package torrent import ( "github.com/monkeyWie/gopeed/protocol/bt/metainfo" "github.com/monkeyWie/gopeed/protocol/bt/peer" + log "github.com/sirupsen/logrus" "testing" ) +func init() { + log.SetLevel(log.InfoLevel) + +} + func TestTorrent_Download(t *testing.T) { torrent := buildTorrent() torrent.Download("e:/testbt/download") @@ -37,8 +43,36 @@ func buildTorrent() *Torrent { "udp://tracker.cyberia.is:6969/announce", "udp://denis.stalker.upeer.me:6969/announce", "udp://ipv4.tracker.harry.lu:80/announce", + "udp://explodie.org:6969/announce", "udp://zephir.monocul.us:6969/announce", "udp://xxxtor.com:2710/announce", + "udp://valakas.rollo.dnsabr.com:2710/announce", + "udp://tracker.zerobytes.xyz:1337/announce", + "udp://tracker.yoshi210.com:6969/announce", + "udp://tracker.uw0.xyz:6969/announce", + "udp://tracker.swateam.org.uk:2710/announce", + "udp://tracker.nyaa.uk:6969/announce", + "udp://tracker.filemail.com:6969/announce", + "udp://tracker.ds.is:6969/announce", + "udp://retracker.sevstar.net:2710/announce", + "udp://retracker.netbynet.ru:2710/announce", + "udp://retracker.akado-ural.ru:80/announce", + "udp://opentracker.i2p.rocks:6969/announce", + "udp://opentor.org:2710/announce", + "udp://open.nyap2p.com:6969/announce", + "udp://chihaya.toss.li:9696/announce", + "udp://bt2.archive.org:6969/announce", + "udp://bt1.archive.org:6969/announce", + "udp://www.loushao.net:8080/announce", + "udp://tracker4.itzmx.com:2710/announce", + "udp://tracker3.itzmx.com:6961/announce", + "udp://tracker2.itzmx.com:6961/announce", + "udp://tracker.lelux.fi:6969/announce", + "udp://tracker.kamigami.org:2710/announce", + "udp://tracker.dler.org:6969/announce", + "udp://tr.bangumi.moe:6969/announce", + "udp://qg.lorzl.gq:2710/announce", + "udp://bt2.54new.com:8080/announce", }...) return NewTorrent(peer.GenPeerID(), metaInfo) } diff --git a/protocol/bt/tracker/tracker.go b/protocol/bt/tracker/tracker.go index df9417c7..81d70287 100644 --- a/protocol/bt/tracker/tracker.go +++ b/protocol/bt/tracker/tracker.go @@ -14,6 +14,7 @@ import ( "net/http" "net/url" "strconv" + "sync" "time" ) @@ -108,7 +109,7 @@ type udpAnnounceRequest struct { event uint32 ip uint32 key uint32 - numWant int32 + numWant uint32 port uint16 } @@ -133,7 +134,7 @@ func (req *udpAnnounceRequest) encode() []byte { binary.BigEndian.PutUint32(buf[80:84], req.event) binary.BigEndian.PutUint32(buf[84:88], req.ip) binary.BigEndian.PutUint32(buf[88:92], req.key) - binary.BigEndian.PutUint32(buf[92:96], uint32(req.numWant)) + binary.BigEndian.PutUint32(buf[92:96], req.numWant) binary.BigEndian.PutUint16(buf[96:98], req.port) return buf } @@ -212,22 +213,27 @@ func (tracker *Tracker) announce(conn *net.UDPConn, timeout int64, connectionId request.event = 0 request.ip = 0 request.key = 0 - request.numWant = 50 + request.numWant = 200 request.port = 6882 encode := request.encode() conn.SetReadDeadline(time.Now().Add(time.Second * time.Duration(timeout))) conn.Write(encode) - buf := make([]byte, 512) + buf := make([]byte, 20+request.numWant*6) len, err := conn.Read(buf) if err != nil { return } + if len < 20 { + err = errors.New("announce fail") + return + } response = newUdpAnnounceResponse(buf[:len]) return } // 通过tracker服务器获取peer信息 -func (tracker *Tracker) Tracker() (peers []peer.Peer, err error) { +func (tracker *Tracker) Tracker() <-chan []peer.Peer { + peersCh := make(chan []peer.Peer) metaInfo := tracker.MetaInfo checkAnnounceList := func(announceList [][]string) bool { @@ -243,20 +249,35 @@ func (tracker *Tracker) Tracker() (peers []peer.Peer, err error) { // 只要获取到了peers就直接返回 if checkAnnounceList(metaInfo.AnnounceList) { + wg := sync.WaitGroup{} for _, announceArr := range metaInfo.AnnounceList { if len(announceArr) > 0 { for _, announce := range announceArr { - peers, err = tracker.DoTracker(announce) - if err == nil { - return - } + wg.Add(1) + go func(announce string) { + peers, err := tracker.DoTracker(announce) + if err == nil { + peersCh <- peers + } + wg.Done() + }(announce) } } } + go func() { + wg.Wait() + close(peersCh) + }() } else { - peers, err = tracker.DoTracker(metaInfo.Announce) + go func() { + peers, err := tracker.DoTracker(metaInfo.Announce) + if err == nil { + peersCh <- peers + } + close(peersCh) + }() } - return + return peersCh } func (tracker *Tracker) DoTracker(announce string) (peers []peer.Peer, err error) { @@ -323,13 +344,13 @@ func (tracker *Tracker) udpTracker(url *url.URL) (peers []peer.Peer, err error) announceResponse, err := func() (announceResponse *udpAnnounceResponse, err error) { var connectResponse *udpConnectResponse // 0:connect 1:announce - state := 0 + state := udpActionConnect // 访问超时最多重试8次,当有请求成功则将重试次数重置为0 for n := 0; n < udpConnectRetries; n++ { // 15 * 2 ^ n seconds timeout := int64(udpConnectTimeout * math.Pow(2, float64(n))) switch state { - case 0: + case udpActionConnect: connectResponse, err = tracker.connect(conn, timeout) if err != nil { if oe, ok := err.(*net.OpError); ok && oe.Timeout() { @@ -339,9 +360,9 @@ func (tracker *Tracker) udpTracker(url *url.URL) (peers []peer.Peer, err error) } } n = -1 - state = 1 + state = udpActionAnnounce break - case 1: + case udpActionAnnounce: announceResponse, err = tracker.announce(conn, timeout, connectResponse.connectionId) if err != nil { if oe, ok := err.(*net.OpError); ok && oe.Timeout() { diff --git a/protocol/bt/tracker/tracker_test.go b/protocol/bt/tracker/tracker_test.go index 123601b2..09119a2e 100644 --- a/protocol/bt/tracker/tracker_test.go +++ b/protocol/bt/tracker/tracker_test.go @@ -17,10 +17,6 @@ func TestTracker_Tracker(t *testing.T) { } tracker.MetaInfo.Announce = "udp://tracker.opentrackr.org:1337/announce" tracker.MetaInfo.AnnounceList = [][]string{} - peers, err := tracker.Tracker() - if err != nil { - panic(err) - return - } + peers := <-tracker.Tracker() fmt.Println(len(peers)) }