nprasad2077/gopeed
1
0 Fork 0
mirror of https://github.com/GopeedLab/gopeed.git synced 2026-11-06 08:30:29 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: download panic

Browse Source
This commit is contained in:
liwei committed 2020-03-01 23:35:51 +08:00
1 parent b0bd2a776a
commit 753830accf
9 files changed
+181 -67
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.vscode/settings.json
Vendored
+1 -9
View File
@@ -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)
}
}
}
protocol/bt/metainfo/metainfo.go
+5 -4
View File
@@ -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 {
protocol/bt/torrent/main/main.go
+74
View File
@@ -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)
}
protocol/bt/torrent/peer_conn.go
+63 -27
View File
@@ -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)
protocol/bt/torrent/peer_pool.go
+5 -4
View File
@@ -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)
}
}()
}
protocol/bt/torrent/piece_state.go
+14 -1
View File
@@ -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)
}
protocol/bt/torrent/torrent.go
+12 -16
View File
@@ -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
}
}
}
protocol/bt/torrent/torrent_test.go
+3 -3
View File
@@ -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) {
protocol/bt/tracker/tracker.go
+4 -3
View File
@@ -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