diff --git a/.vscode/settings.json b/.vscode/settings.json index ed7427c7..0c9921c3 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -1,17 +1,9 @@ { "go.useLanguageServer": true, - "go.languageServerExperimentalFeatures": { - "diagnostics": true // for diagnostics as you type - }, "[go]": { - "editor.snippetSuggestions": "none", "editor.formatOnSave": true, "editor.codeActionsOnSave": { "source.organizeImports": true } }, - "gopls": { - "usePlaceholders": true, // add parameter placeholders when completing a function - "enhancedHover": true // experimental to improve quality of hover (will be on by default soon) - } -} +} \ No newline at end of file diff --git a/protocol/bt/metainfo/metainfo.go b/protocol/bt/metainfo/metainfo.go index 0c10bae5..2a7f83a6 100644 --- a/protocol/bt/metainfo/metainfo.go +++ b/protocol/bt/metainfo/metainfo.go @@ -3,8 +3,9 @@ package metainfo import ( "crypto/sha1" "encoding/json" - "github.com/marksamman/bencode" "os" + + "github.com/marksamman/bencode" ) type MetaInfo struct { @@ -22,9 +23,9 @@ type MetaInfo struct { UrlList []string `json:"url-list"` Info *Info `json:"info"` - infoHash [20]byte `json:"-"` - totalSize uint64 `json:"-"` - fileDetails []FileDetail `json:"-"` + infoHash [20]byte + totalSize uint64 + fileDetails []FileDetail } type Info struct { diff --git a/protocol/bt/torrent/main/main.go b/protocol/bt/torrent/main/main.go new file mode 100644 index 00000000..eac5299b --- /dev/null +++ b/protocol/bt/torrent/main/main.go @@ -0,0 +1,74 @@ +package main + +import ( + "github.com/monkeyWie/gopeed/protocol/bt/metainfo" + "github.com/monkeyWie/gopeed/protocol/bt/peer" + "github.com/monkeyWie/gopeed/protocol/bt/torrent" + log "github.com/sirupsen/logrus" +) + +func main() { + log.SetLevel(log.DebugLevel) + torrent := buildTorrent() + torrent.Download("e:/testbt/download") +} + +func buildTorrent() *torrent.Torrent { + // metaInfo, err := metainfo.ParseFromFile("../testdata/Game.of.Thrones.S08E05.720p.WEB.H264-MEMENTO.torrent") + metaInfo, err := metainfo.ParseFromFile("../../testdata/office.torrent") + if err != nil { + panic(err) + } + + metaInfo.AnnounceList[0] = append(metaInfo.AnnounceList[0], []string{ + "udp://tracker.coppersurfer.tk:6969/announce", + "udp://tracker.leechers-paradise.org:6969/announce", + "udp://tracker.opentrackr.org:1337/announce", + "udp://p4p.arenabg.com:1337/announce", + "udp://9.rarbg.to:2710/announce", + "udp://9.rarbg.me:2710/announce", + "udp://tracker.internetwarriors.net:1337/announce", + "udp://exodus.desync.com:6969/announce", + "udp://tracker.tiny-vps.com:6969/announce", + "udp://tracker.sbsub.com:2710/announce", + "udp://retracker.lanta-net.ru:2710/announce", + "udp://open.stealth.si:80/announce", + "udp://open.demonii.si:1337/announce", + "udp://tracker.torrent.eu.org:451/announce", + "udp://tracker.moeking.me:6969/announce", + "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 torrent.NewTorrent(peer.GenPeerID(), metaInfo) +} diff --git a/protocol/bt/torrent/peer_conn.go b/protocol/bt/torrent/peer_conn.go index 2f800a07..975f13b0 100644 --- a/protocol/bt/torrent/peer_conn.go +++ b/protocol/bt/torrent/peer_conn.go @@ -6,11 +6,6 @@ import ( "encoding/binary" "errors" "fmt" - "github.com/cenkalti/mse" - "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" @@ -18,12 +13,20 @@ import ( "path" "path/filepath" "time" + + "github.com/cenkalti/mse" + "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" ) // 现在主流客户端的block大小都是16KB const blockSize = 2 << 13 +const keepaliveTimeout = 60 * 2 var errPieceCheckFailed = errors.New("piece check failed") +var keepaliveData = make([]byte, 4) type peerConn struct { torrent *Torrent @@ -38,12 +41,15 @@ type peerConn struct { // peer is interested in this client peerInterested bool - bitfield *message.Bitfield - readyEnd bool - downloadEnd bool + // 上一次收到包的时间 + lastReciveTime int64 + + bitfield *message.Bitfield + readyEnd bool downloadedCh chan error disconnectCh chan error + keepaliveCh chan error // block个数 blockCount int // block下载队列,官方推荐为5 @@ -112,6 +118,26 @@ func (pc *peerConn) handshake() (*peer.Handshake, error) { return handshakeRes, nil } +func (pc *peerConn) handleKeepalive() { + pc.lastReciveTime = time.Now().Unix() + for { + time.Sleep(time.Minute * 2) + // 如果超过两分钟没有响应,则断开连接 + if time.Now().Unix()-pc.lastReciveTime > keepaliveTimeout { + pc.conn.Close() + break + } + // 两分钟发送一次心跳包 + _, err := pc.conn.Write(keepaliveData) + if err != nil { + pc.conn.Close() + break + } + } + pc.keepaliveCh <- fmt.Errorf("keepalive fail") + close(pc.keepaliveCh) +} + // 准备下载 func (pc *peerConn) ready() error { if err := pc.dialMse(); err != nil { @@ -121,13 +147,15 @@ func (pc *peerConn) ready() error { return fmt.Errorf("handshake error %w", err) } pc.readyEnd = false - pc.downloadEnd = false readyCh := make(chan bool) pc.disconnectCh = make(chan error) + pc.keepaliveCh = make(chan error) + go pc.handleKeepalive() go func() { scanner := bufio.NewScanner(pc.conn) scanner.Split(message.SplitMessage) for scanner.Scan() { + pc.lastReciveTime = time.Now().Unix() buf := scanner.Bytes() length := binary.BigEndian.Uint32(buf[:4]) if length == 0 { @@ -172,9 +200,8 @@ func (pc *peerConn) ready() error { case status := <-readyCh: if status { return nil - } else { - return errors.New("ready fail") } + return errors.New("ready fail") case <-time.After(time.Second * 30): // 30秒之后超时 return errors.New("ready time out") @@ -184,15 +211,16 @@ func (pc *peerConn) ready() error { pc.conn.Close() return err } - pc.blockQueueCh = make(chan interface{}, 5) return nil } // 下载指定piece -func (pc *peerConn) downloadPiece(index int) error { +func (pc *peerConn) downloadPiece(index int) (err error) { pieceLength := pc.torrent.MetaInfo.GetPieceLength(index) pc.blockCount = int(math.Ceil(float64(pieceLength) / blockSize)) pc.downloadedCh = make(chan error) + pc.blockQueueCh = make(chan interface{}, 5) + defer close(pc.blockQueueCh) // 按块下载分片 for i := 0; i < pc.blockCount; i++ { offset := blockSize * i @@ -212,23 +240,35 @@ func (pc *peerConn) downloadPiece(index int) error { select { case pc.blockQueueCh <- nil: break - case err := <-pc.downloadedCh: - return err - case err := <-pc.disconnectCh: - return err + case err = <-pc.downloadedCh: + break + case err = <-pc.disconnectCh: + break + case err = <-pc.keepaliveCh: + break + } + if err != nil { + pc.conn.Close() + return } // 发起request,对方会响应piece - _, err := pc.conn.Write(message.NewRequest(uint32(index), uint32(offset), blockLength).Encode()) + _, err = pc.conn.Write(message.NewRequest(uint32(index), uint32(offset), blockLength).Encode()) if err != nil { - return err + break } } select { - case err := <-pc.downloadedCh: - return err - case err := <-pc.disconnectCh: - return err + case err = <-pc.downloadedCh: + break + case err = <-pc.disconnectCh: + break + case err = <-pc.keepaliveCh: + break } + if err != nil { + pc.conn.Close() + } + return } func (pc *peerConn) handleUnchoke(readyCh chan<- bool) { @@ -367,8 +407,6 @@ func (pc *peerConn) handlePiece(buf []byte) { break } } - // 标记下载完成 - pc.downloadEnd = true // 断开连接 TODO 用连接池进行复用 pc.conn.Close() @@ -376,11 +414,9 @@ func (pc *peerConn) handlePiece(buf []byte) { downHash := [20]byte{} copy(downHash[:], sha1.Sum(nil)) if downHash == pc.torrent.MetaInfo.Info.Pieces[piece.Index] { - log.Infof("piece %d 校验通过", piece.Index) // piece下载完成 pc.downloadedCh <- nil } else { - log.Infof("piece %d 校验失败", piece.Index) pc.downloadedCh <- errPieceCheckFailed } close(pc.downloadedCh) diff --git a/protocol/bt/torrent/peer_pool.go b/protocol/bt/torrent/peer_pool.go index beab0508..2b2d2b97 100644 --- a/protocol/bt/torrent/peer_pool.go +++ b/protocol/bt/torrent/peer_pool.go @@ -1,10 +1,11 @@ package torrent import ( - "github.com/monkeyWie/gopeed/protocol/bt/peer" - "github.com/monkeyWie/gopeed/protocol/bt/tracker" "sync" "time" + + "github.com/monkeyWie/gopeed/protocol/bt/peer" + "github.com/monkeyWie/gopeed/protocol/bt/tracker" ) type peerPool struct { @@ -45,8 +46,8 @@ func (pp *peerPool) fetch() { } }() } - // 每5分钟检测一次 - time.Sleep(time.Minute * 5) + // 每2分钟检测一次 + time.Sleep(time.Minute * 2) } }() } diff --git a/protocol/bt/torrent/piece_state.go b/protocol/bt/torrent/piece_state.go index e485a205..91f3e5da 100644 --- a/protocol/bt/torrent/piece_state.go +++ b/protocol/bt/torrent/piece_state.go @@ -1,8 +1,9 @@ package torrent import ( - "github.com/RoaringBitmap/roaring" "sync" + + "github.com/RoaringBitmap/roaring" ) type state int @@ -62,6 +63,18 @@ func (ps *pieces) getReady() int { return -1 } +func (ps *pieces) getLeft() int { + ps.rw.RLock() + defer ps.rw.RUnlock() + count := 0 + for _, s := range ps.arr { + if s.state != stateFinished { + count++ + } + } + return count +} + func (ps *pieces) size() int { return len(ps.arr) } diff --git a/protocol/bt/torrent/torrent.go b/protocol/bt/torrent/torrent.go index 8ce928fd..72847fdb 100644 --- a/protocol/bt/torrent/torrent.go +++ b/protocol/bt/torrent/torrent.go @@ -2,9 +2,10 @@ package torrent import ( "errors" + "time" + "github.com/monkeyWie/gopeed/protocol/bt/metainfo" log "github.com/sirupsen/logrus" - "time" ) type Torrent struct { @@ -13,7 +14,7 @@ type Torrent struct { pieces *pieces peerPool *peerPool - completeCh chan bool + completeCh chan interface{} Path string } @@ -32,9 +33,9 @@ func (t *Torrent) Download(path string) { t.peerPool = newPeerPool(t) t.peerPool.fetch() - pieceQueueCh := make(chan interface{}, 128) + pieceQueueCh := make(chan interface{}, 256) defer close(pieceQueueCh) - t.completeCh = make(chan bool) + t.completeCh = make(chan interface{}) defer close(t.completeCh) for { pieceQueueCh <- nil @@ -42,13 +43,10 @@ func (t *Torrent) Download(path string) { index := t.pieces.getReady() // 没有待下载的piece了 if index == -1 { - if <-t.completeCh { - // 下载完成 - break - } else { - // 继续下载 - continue - } + log.Debugf("no piece to download") + <-t.completeCh + // 下载完成 + break } go func(index int) { peer := t.peerPool.get() @@ -59,16 +57,14 @@ func (t *Torrent) Download(path string) { return } 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 { - log.Debugf("peer ready success:%s,download %d", peer.Address(), index) err = pc.downloadPiece(index) if err != nil { + log.Debugf("piece %d download fail:%s", index, err.Error()) // 下载失败 t.pieces.setState(index, stateReady) t.peerPool.unavailable(peer) @@ -76,14 +72,14 @@ func (t *Torrent) Download(path string) { // piece校验失败,需要重新下载过一遍,清空block记录 t.pieces.clearBlocks(index) } - t.completeCh <- false } else { // 下载完成 t.pieces.setState(index, stateFinished) t.peerPool.release(peer) + log.Debugf("piece %d download success,total:%d,left:%d", index, t.pieces.size(), t.pieces.getLeft()) // 检查是否所有piece下载完成 if t.pieces.isDone() { - t.completeCh <- true + t.completeCh <- nil } } } diff --git a/protocol/bt/torrent/torrent_test.go b/protocol/bt/torrent/torrent_test.go index a03ce138..66360f33 100644 --- a/protocol/bt/torrent/torrent_test.go +++ b/protocol/bt/torrent/torrent_test.go @@ -1,15 +1,15 @@ package torrent import ( + "testing" + "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) - + log.SetLevel(log.DebugLevel) } func TestTorrent_Download(t *testing.T) { diff --git a/protocol/bt/tracker/tracker.go b/protocol/bt/tracker/tracker.go index 81d70287..f5fa6932 100644 --- a/protocol/bt/tracker/tracker.go +++ b/protocol/bt/tracker/tracker.go @@ -4,8 +4,6 @@ import ( "encoding/binary" "errors" "fmt" - "github.com/monkeyWie/gopeed/protocol/bt/metainfo" - "github.com/monkeyWie/gopeed/protocol/bt/peer" "io" "io/ioutil" "math" @@ -16,12 +14,15 @@ import ( "strconv" "sync" "time" + + "github.com/monkeyWie/gopeed/protocol/bt/metainfo" + "github.com/monkeyWie/gopeed/protocol/bt/peer" ) const ( udpConnectRequestMagic = 0x41727101980 udpConnectTimeout = 15 - udpConnectRetries = 8 + udpConnectRetries = 3 udpActionConnect = 0 udpActionAnnounce = 1 udpActionScrape = 2