From fb4506622863d02dbba11a108fa950688feb6c61 Mon Sep 17 00:00:00 2001 From: liwei Date: Fri, 1 Nov 2019 15:05:25 +0800 Subject: [PATCH] =?UTF-8?q?=E7=AE=80=E5=8D=95=E5=AE=9E=E7=8E=B0peer=20wire?= =?UTF-8?q?=20protocol=E5=8D=8F=E8=AE=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- down/bt/metainfo/metainfo.go | 7 +- down/bt/metainfo/metainfo_test.go | 2 +- down/bt/peer/message/message.go | 18 +- down/bt/torrent/peer_conn.go | 403 +++++++++++++++++++++++++++++ down/bt/torrent/peer_conn_test.go | 147 +++++++++++ down/bt/torrent/peer_pool.go | 82 ++++++ down/bt/torrent/peer_pool_test.go | 40 +++ down/bt/torrent/peer_state.go | 292 --------------------- down/bt/torrent/peer_state_test.go | 85 ------ down/bt/torrent/piece_state.go | 55 +++- down/bt/torrent/torrent.go | 92 ++++++- down/bt/torrent/torrent_test.go | 32 +++ down/bt/tracker/tracker.go | 8 +- 13 files changed, 859 insertions(+), 404 deletions(-) create mode 100644 down/bt/torrent/peer_conn.go create mode 100644 down/bt/torrent/peer_conn_test.go create mode 100644 down/bt/torrent/peer_pool.go create mode 100644 down/bt/torrent/peer_pool_test.go delete mode 100644 down/bt/torrent/peer_state.go delete mode 100644 down/bt/torrent/peer_state_test.go create mode 100644 down/bt/torrent/torrent_test.go diff --git a/down/bt/metainfo/metainfo.go b/down/bt/metainfo/metainfo.go index a6584761..a8df3308 100644 --- a/down/bt/metainfo/metainfo.go +++ b/down/bt/metainfo/metainfo.go @@ -30,7 +30,7 @@ type MetaInfo struct { type Info struct { Name string `json:"name"` PieceLength uint64 `json:"piece length"` - Pieces []string + Pieces [][20]byte Length uint64 `json:"length"` Files []File `json:"files"` } @@ -71,11 +71,12 @@ func ParseFromFile(path string) (*MetaInfo, error) { // Split pieces hash pieces := info["pieces"].(string) if pieces != "" { - metaInfo.Info.Pieces = make([]string, len(pieces)/20) + metaInfo.Info.Pieces = make([][20]byte, len(pieces)/20) // pieces parse for i := range metaInfo.Info.Pieces { begin := i * 20 - metaInfo.Info.Pieces[i] = pieces[begin : begin+20] + metaInfo.Info.Pieces[i] = [20]byte{} + copy(metaInfo.Info.Pieces[i][:], pieces[begin:begin+20]) } } } diff --git a/down/bt/metainfo/metainfo_test.go b/down/bt/metainfo/metainfo_test.go index 2eb55ce3..042b09e3 100644 --- a/down/bt/metainfo/metainfo_test.go +++ b/down/bt/metainfo/metainfo_test.go @@ -10,6 +10,6 @@ func TestParseFromFile(t *testing.T) { if err != nil { panic(err) } - fmt.Printf("%+v\n", metaInfo) + fmt.Printf("%+v\n", metaInfo.AnnounceList) fmt.Printf("%+v\n", metaInfo.Info.Files) } diff --git a/down/bt/peer/message/message.go b/down/bt/peer/message/message.go index 99c9e9c5..b798f87f 100644 --- a/down/bt/peer/message/message.go +++ b/down/bt/peer/message/message.go @@ -8,15 +8,15 @@ import ( type ID byte const ( - IdChoke ID = 0 - IdUnchoke ID = 1 - IdInterested ID = 2 - IdNotinterested ID = 3 - IdHave ID = 4 - IdBitfield ID = 5 - IdRequest ID = 6 - IdPiece ID = 7 - IdCancel ID = 8 + IdChoke ID = iota + IdUnchoke + IdInterested + IdNotinterested + IdHave + IdBitfield + IdRequest + IdPiece + IdCancel ) /*type Message struct { diff --git a/down/bt/torrent/peer_conn.go b/down/bt/torrent/peer_conn.go new file mode 100644 index 00000000..1af8f463 --- /dev/null +++ b/down/bt/torrent/peer_conn.go @@ -0,0 +1,403 @@ +package torrent + +import ( + "bufio" + "crypto/sha1" + "encoding/binary" + "errors" + "fmt" + "github.com/cenkalti/mse" + "github.com/monkeyWie/gopeed/down/bt/metainfo" + "github.com/monkeyWie/gopeed/down/bt/peer" + "github.com/monkeyWie/gopeed/down/bt/peer/message" + "io" + "math" + "net" + "os" + "path" + "sync" + "time" +) + +type peerConn struct { + torrent *Torrent + peer *peer.Peer + conn net.Conn + // this client is choking the peer + amChoking bool + // this client is interested in the peer + amInterested bool + // peer is choking this client + peerChoking bool + // peer is interested in this client + peerInterested bool + + bitfield *message.Bitfield + readyComplete bool + pieceDownloadedCh chan bool + // block下载队列，官方推荐为5 + pieceBlockQueue chan interface{} + // piece上所有的block下载状态 + pieceBlockState sync.Map +} + +func NewPeerConn(torrent *Torrent, peer *peer.Peer) *peerConn { + return &peerConn{ + torrent: torrent, + peer: peer, + } +} + +// 使用MSE加密来避免运营商对bt流量的封锁，基本上现在市面上BT客户端都默认开启了，不用MSE的话很多Peer拒绝连接 +// see http://wiki.vuze.com/w/Message_Stream_Encryption +func (pc *peerConn) dialMse() error { + conn, err := net.DialTimeout("tcp", pc.peer.Address(), time.Second*time.Duration(30)) + if err != nil { + return err + } + mseConn := mse.WrapConn(conn) + infoHash := pc.torrent.MetaInfo.GetInfoHash() + _, err = mseConn.HandshakeOutgoing(infoHash[:], mse.PlainText, nil) + if err != nil { + mseConn.Close() + return err + } + pc.conn = mseConn + return nil +} + +// Handshake of Peer wire protocol +// see https://wiki.theory.org/index.php/BitTorrentSpecification#Handshake +func (pc *peerConn) handshake() (*peer.Handshake, error) { + handshakeRes, err := func() (*peer.Handshake, error) { + handshakeReq := peer.NewHandshake([8]byte{}, pc.torrent.MetaInfo.GetInfoHash(), pc.torrent.PeerID) + _, err := pc.conn.Write(handshakeReq.Encode()) + if err != nil { + return nil, err + } + var read [68]byte + _, err = io.ReadFull(pc.conn, read[:]) + if err != nil { + return nil, err + } + handshakeRes := &peer.Handshake{} + err = handshakeRes.Decode(read[:]) + if err != nil { + return nil, err + } + // InfoHash不匹配 + if handshakeRes.InfoHash != handshakeReq.InfoHash { + return nil, fmt.Errorf("info_hash not currently serving") + } + return handshakeRes, nil + }() + if err != nil { + pc.conn.Close() + return nil, err + } + // init state + pc.amChoking = true + pc.amInterested = false + pc.peerChoking = true + pc.peerInterested = false + return handshakeRes, nil +} + +// 准备下载 +func (pc *peerConn) ready() error { + if err := pc.dialMse(); err != nil { + return err + } + if _, err := pc.handshake(); err != nil { + return err + } + pc.readyComplete = false + readyCh := make(chan bool) + defer close(readyCh) + go func() { + scanner := bufio.NewScanner(pc.conn) + scanner.Split(message.SplitMessage) + for scanner.Scan() { + buf := scanner.Bytes() + length := binary.BigEndian.Uint32(buf[:4]) + if length == 0 { + // keepalive + } else { + messageID := message.ID(buf[4]) + switch messageID { + case message.IdChoke: + break + case message.IdUnchoke: + pc.handleUnchoke(readyCh) + break + case message.IdInterested: + break + case message.IdNotinterested: + break + case message.IdHave: + break + case message.IdBitfield: + pc.handleBitfield(buf) + break + case message.IdRequest: + break + case message.IdPiece: + pc.handlePiece(buf) + break + case message.IdCancel: + break + } + } + } + }() + err := func() error { + select { + case status := <-readyCh: + if status { + return nil + } else { + return errors.New("ready fail") + } + case <-time.After(time.Second * 30): + // 30秒之后超时 + return errors.New("ready time out") + } + }() + pc.readyComplete = true + if err != nil { + pc.conn.Close() + return err + } + pc.pieceBlockQueue = make(chan interface{}, 5) + return nil +} + +// 下载指定piece +func (pc *peerConn) downloadPiece(index int) error { + pc.pieceDownloadedCh = make(chan bool) + defer close(pc.pieceDownloadedCh) + pc.pieceBlockState = sync.Map{} + pieceLength := pc.torrent.MetaInfo.GetPieceSize(index) + lastOffset := uint64(0) + + waitDuration := time.Second * 30 + timer := time.NewTimer(waitDuration) + defer timer.Stop() + // 按块下载分片 + for lastOffset < pieceLength { + blockLength := math.Min(2<<13, float64(pieceLength-lastOffset)) + // 标记block开始下载 + lastOffset32 := uint32(lastOffset) + pc.pieceBlockState.Store(lastOffset32, false) + // block下载排队，并做超时处理 + timer.Reset(waitDuration) + select { + case pc.pieceBlockQueue <- nil: + case <-timer.C: + return errors.New("download timeout") + } + _, err := pc.conn.Write(message.NewRequest(uint32(index), lastOffset32, uint32(blockLength)).Encode()) + if err != nil { + fmt.Println(err) + } + lastOffset += uint64(blockLength) + } + // 监听是否下载完成 + <-pc.pieceDownloadedCh + return nil +} + +func (pc *peerConn) handleUnchoke(readyCh chan<- bool) { + pc.peerChoking = false + // 已经处理过Unchoke信号 + if pc.readyComplete { + return + } + // 如果客户端对peer感兴趣并且peer没有choked客户端，就可以开始下载了 + if pc.amInterested { + readyCh <- true + } else { + readyCh <- false + } +} + +func (pc *peerConn) handleBitfield(buf []byte) { + pc.bitfield = &message.Bitfield{} + pc.bitfield.Decode(buf) + have := pc.getHavePieces(pc.bitfield) + if len(have) > 0 { + // 表示对该peer感兴趣，并且不choked该peer + pc.conn.Write(message.NewInterested().Encode()) + pc.amInterested = true + + pc.conn.Write(message.NewUnchoke().Encode()) + pc.amChoking = false + } else { + pc.conn.Close() + } +} + +// 处理分片下载响应 +func (pc *peerConn) handlePiece(buf []byte) { + piece := &message.Piece{} + piece.Decode(buf) + info := pc.torrent.MetaInfo.Info + fds := pc.torrent.MetaInfo.GetFileDetails() + blockLength := int64(len(piece.Block)) + pieceBegin := int64(piece.Index) * int64(info.PieceLength) + blockBegin := pieceBegin + int64(piece.Begin) + // 获取对应要的文件 + var fileBlocks []fileBlock + if len(info.Files) == 0 { + // 单文件 + fileBlocks = append(fileBlocks, fileBlock{ + filepath: info.Name, + fileSeek: blockBegin, + blockBegin: 0, + blockEnd: blockLength, + }) + } else { + // 获取要写入的第一个文件和偏移 + writeIndex := getWriteFile(blockBegin, fds) + // block对应种子的偏移 + blockFileBegin := blockBegin + // block对应文件的偏移 + var blockSeek int64 = 0 + for _, f := range fds[writeIndex:] { + // 计算文件可剩余写入字节数 + fileWritable := f.End - blockFileBegin + // 计算block剩余写入字节数 + blockWritable := blockLength - blockSeek + fb := fileBlock{ + filepath: f.Path[0], + fileSeek: blockFileBegin - f.Begin, + blockBegin: blockSeek, + } + if fileWritable >= blockWritable { + // 若够block写入直接跳出循环 + fb.blockEnd = blockSeek + blockWritable + fileBlocks = append(fileBlocks, fb) + break + } else { + // 否则计算剩余可写字节数，写入到下一个文件中 + fb.blockEnd = blockSeek + fileWritable + fileBlocks = append(fileBlocks, fb) + blockFileBegin = f.End + blockSeek += fileWritable + } + } + } + // fmt.Println(fileBlocks) + for _, f := range fileBlocks { + func() { + name := path.Join(pc.torrent.Path, f.filepath) + file, err := os.OpenFile(name, os.O_RDWR|os.O_CREATE, 0644) + if err != nil { + panic(err) + } + defer file.Close() + _, err = file.WriteAt(piece.Block[f.blockBegin:f.blockEnd], f.fileSeek) + if err != nil { + fmt.Println(err) + } + }() + } + pc.pieceBlockState.Store(piece.Begin, true) + // 出队 + <-pc.pieceBlockQueue + flag := true + // fmt.Printf("index %d: %v\n", piece.Index, pc.pieceBlockState) + pc.pieceBlockState.Range(func(key, value interface{}) bool { + // 如果有还没下载完的block + if !value.(bool) { + flag = false + return false + } + return true + }) + // piece全部下载完 + if flag { + // 计算piece对应的文件偏移 + sha1 := sha1.New() + pieceLength := pc.torrent.MetaInfo.GetPieceSize(int(piece.Index)) + writeIndex := getWriteFile(pieceBegin, fds) + fileBegin := pieceBegin - fds[writeIndex].Begin + for _, fd := range fds[writeIndex:] { + func() { + file, err := os.Open(path.Join(pc.torrent.Path, fd.Path[0])) + if err != nil { + panic(err) + } + defer file.Close() + _, err = file.Seek(fileBegin, 0) + if err != nil { + panic(err) + } + written, err := io.CopyN(sha1, file, int64(pieceLength)) + if err != nil { + if err != io.EOF { + panic(err) + } + } + pieceLength -= uint64(written) + }() + if pieceLength > 0 { + // 继续读下个文件 + fileBegin = 0 + } else { + break + } + } + // 校验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) + } else { + fmt.Printf("piece %d 校验失败\n", piece.Index) + } + // piece下载完成 + pc.pieceDownloadedCh <- true + } +} + +// 获取piece对应要写入的文件 +func getWriteFile(pieceBegin int64, fds []metainfo.FileDetail) int { + for i, f := range fds { + if f.Begin <= pieceBegin && f.End > pieceBegin { + return i + } + } + return -1 +} + +type fileBlock struct { + filepath string + fileSeek int64 + blockBegin int64 + blockEnd int64 +} + +// 获取peer能提供需要下载的文件分片 +func (pc *peerConn) getHavePieces(bitfield *message.Bitfield) []int { + states := make([]bool, pc.torrent.PiecesState.Size()) + for i := range states { + states[i] = pc.torrent.PiecesState.getState(i) == stateFinished + } + return bitfield.Have(states) +} + +// 获取要写入到的文件 +/*func (ps *peerConn) getWriteFile(request *message.Request) string { + info := ps.torrent.MetaInfo.Info + // 单文件 + if len(info.Files) == 0 { + return info.Name + } else { + request.Index * info.PieceLength + be + for i := 0; i < len(info.Files); i++ { + + } + } +} +*/ diff --git a/down/bt/torrent/peer_conn_test.go b/down/bt/torrent/peer_conn_test.go new file mode 100644 index 00000000..09656843 --- /dev/null +++ b/down/bt/torrent/peer_conn_test.go @@ -0,0 +1,147 @@ +package torrent + +import ( + "fmt" + "github.com/monkeyWie/gopeed/down/bt/peer" + "github.com/monkeyWie/gopeed/down/bt/tracker" + "sync" + "testing" + "time" +) + +func Test_peerState_ready(t *testing.T) { + + ps := getUsablePeer() + fmt.Println("get a peer") + + // 下载前准备 + err := ps.ready() + fmt.Println("ready end") + if err != nil { + fmt.Println(ps.peer.Address()) + t.Fatal(err) + } + +} + +func Test_peerState_download(t *testing.T) { + + psCh := getUsablePeerMore() + + lock := sync.Mutex{} + state := map[int]bool{} + + for { + select { + case ps := <-psCh: + go func() { + // 下载前准备 + err := ps.ready() + if err != nil { + fmt.Println("ready error") + return + } + have := ps.getHavePieces(ps.bitfield) + fmt.Printf("download is ready,have %d\n", len(have)) + if len(have) > 0 { + // 获取分片的长度 + for _, index := range have { + lock.Lock() + if state[index] { + lock.Unlock() + continue + } else { + state[index] = true + lock.Unlock() + } + fmt.Printf("download index %d\n", index) + ps.downloadPiece(index) + } + return + } + }() + } + } +} + +func Test_peerState_downloadPiece(t *testing.T) { + + ps := getUsablePeer() + fmt.Println("get a peer") + + // 下载前准备 + err := ps.ready() + if err != nil { + t.Fatal(err) + } + have := ps.getHavePieces(ps.bitfield) + fmt.Printf("download is ready,have %d\n", len(have)) + if len(have) > 0 { + // 获取分片的长度 + for _, index := range have { + fmt.Printf("download index %d\n", index) + ps.downloadPiece(index) + } + return + } +} + +// 获取一个能连接上的peer +func getUsablePeer() *peerConn { + ch := getUsablePeerMore() + ps := <-ch + return ps +} + +// 获取多个能连接上的peer +func getUsablePeerMore() chan *peerConn { + torrent := buildTorrent() + tracker := &tracker.Tracker{ + PeerID: torrent.PeerID, + MetaInfo: torrent.MetaInfo, + } + peers, err := tracker.Tracker() + if err != nil { + panic(err) + } + fmt.Printf("Tracker end,peer count:%d\n", len(peers)) + ch := make(chan *peerConn) + for i := range peers { + go peerTest(torrent, &peers[i], ch) + } + return ch +} + +func peerTest(torrent *Torrent, peer *peer.Peer, ch chan *peerConn) { + ps := &peerConn{ + torrent: torrent, + peer: peer, + } + // 连接至peer + err := ps.ready() + if err != nil { + return + } + ps.conn.Close() + ch <- ps +} + +func TestSome(t *testing.T) { + ch := make(chan interface{}) + + go func() { + time.Sleep(time.Second) + select { + case ch <- nil: + fmt.Println("write") + case <-time.After(time.Second * 3): + fmt.Println("timeout") + } + }() + + time.Sleep(time.Second * 5) + r := <-ch + fmt.Println(r) + fmt.Println("111") + +} diff --git a/down/bt/torrent/peer_pool.go b/down/bt/torrent/peer_pool.go new file mode 100644 index 00000000..77dd6231 --- /dev/null +++ b/down/bt/torrent/peer_pool.go @@ -0,0 +1,82 @@ +package torrent + +import ( + "github.com/monkeyWie/gopeed/down/bt/peer" + "sync" +) + +type peerPool struct { + peers []peerState + lock *sync.Mutex +} + +type peerState struct { + peer *peer.Peer + using bool + ready bool +} + +func newPeerPool() *peerPool { + return &peerPool{ + lock: &sync.Mutex{}, + } +} + +func (pp *peerPool) put(peers []peer.Peer) { + pp.lock.Lock() + defer pp.lock.Unlock() + for i := range peers { + pp.peers = append(pp.peers, peerState{ + peer: &peers[i], + using: false, + }) + } +} + +func (pp *peerPool) get() *peer.Peer { + pp.lock.Lock() + defer pp.lock.Unlock() + for i, p := range pp.peers { + if !p.using { + pp.peers[i].using = true + return pp.peers[i].peer + } + } + return nil +} + +func (pp *peerPool) ready(peer *peer.Peer) { + pp.lock.Lock() + defer pp.lock.Unlock() + i := findPeer(pp, peer) + if i != -1 { + pp.peers[i].ready = true + } +} + +func (pp *peerPool) release(peer *peer.Peer) { + pp.lock.Lock() + defer pp.lock.Unlock() + i := findPeer(pp, peer) + if i != -1 { + pp.peers[i].using = false + } +} + +func (pp *peerPool) remove(peer *peer.Peer) { + pp.lock.Lock() + defer pp.lock.Unlock() + i := findPeer(pp, peer) + if i != -1 { + pp.peers = append(pp.peers[0:i], pp.peers[i+1:]...) + } +} + +func findPeer(pp *peerPool, peer *peer.Peer) int { + for i := range pp.peers { + if pp.peers[i].peer == peer { + return i + } + } + return -1 +} diff --git a/down/bt/torrent/peer_pool_test.go b/down/bt/torrent/peer_pool_test.go new file mode 100644 index 00000000..bd35c213 --- /dev/null +++ b/down/bt/torrent/peer_pool_test.go @@ -0,0 +1,40 @@ +package torrent + +import ( + "fmt" + "github.com/monkeyWie/gopeed/down/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.remove(get) + for i := 0; i < 5; i++ { + fmt.Println(pool.get()) + } + fmt.Println(pool.get()) +} diff --git a/down/bt/torrent/peer_state.go b/down/bt/torrent/peer_state.go deleted file mode 100644 index 9c4c7710..00000000 --- a/down/bt/torrent/peer_state.go +++ /dev/null @@ -1,292 +0,0 @@ -package torrent - -import ( - "bufio" - "encoding/binary" - "fmt" - "github.com/cenkalti/mse" - "github.com/monkeyWie/gopeed/down/bt/metainfo" - "github.com/monkeyWie/gopeed/down/bt/peer" - "github.com/monkeyWie/gopeed/down/bt/peer/message" - "io" - "math" - "net" - "os" - "path" - "time" -) - -type peerState struct { - torrent *Torrent - *peer.Peer - conn net.Conn - // this client is choking the peer - amChoking bool - // this client is interested in the peer - amInterested bool - // peer is choking this client - peerChoking bool - // peer is interested in this client - peerInterested bool - - writeMsgCh chan []byte - bitfield *message.Bitfield -} - -// 使用MSE加密来避免运营商对bt流量的封锁，基本上现在市面上BT客户端都默认开启了，不用MSE的话很多Peer拒绝连接 -// see http://wiki.vuze.com/w/Message_Stream_Encryption -func (ps *peerState) dialMse() error { - conn, err := net.DialTimeout("tcp", ps.Address(), time.Second*time.Duration(30)) - if err != nil { - return err - } - mseConn := mse.WrapConn(conn) - infoHash := ps.torrent.MetaInfo.GetInfoHash() - _, err = mseConn.HandshakeOutgoing(infoHash[:], mse.PlainText, nil) - if err != nil { - mseConn.Close() - return err - } - ps.conn = mseConn - return nil -} - -// Handshake of Peer wire protocol -// see https://wiki.theory.org/index.php/BitTorrentSpecification#Handshake -func (ps *peerState) handshake() (*peer.Handshake, error) { - handshakeRes, err := func() (*peer.Handshake, error) { - handshakeReq := peer.NewHandshake([8]byte{}, ps.torrent.MetaInfo.GetInfoHash(), ps.torrent.PeerID) - _, err := ps.conn.Write(handshakeReq.Encode()) - if err != nil { - return nil, err - } - var read [68]byte - _, err = io.ReadFull(ps.conn, read[:]) - if err != nil { - return nil, err - } - handshakeRes := &peer.Handshake{} - err = handshakeRes.Decode(read[:]) - if err != nil { - return nil, err - } - // InfoHash不匹配 - if handshakeRes.InfoHash != handshakeReq.InfoHash { - return nil, fmt.Errorf("info_hash not currently serving") - } - return handshakeRes, nil - }() - if err != nil { - ps.conn.Close() - return nil, err - } - // init state - ps.amChoking = true - ps.amInterested = false - ps.peerChoking = true - ps.peerInterested = false - return handshakeRes, nil -} - -func (ps *peerState) download() error { - // 异步写，防止读取阻塞 - ps.writeMsgCh = make(chan []byte, 64) - go func() { - for buf := range ps.writeMsgCh { - _, err := ps.conn.Write(buf) - if err != nil { - fmt.Println(err) - ps.conn.Close() - } - } - }() - scanner := bufio.NewScanner(ps.conn) - scanner.Split(message.SplitMessage) - for scanner.Scan() { - buf := scanner.Bytes() - length := binary.BigEndian.Uint32(buf[:4]) - if length == 0 { - // keepalive - } else { - messageID := message.ID(buf[4]) - switch messageID { - case message.IdChoke: - break - case message.IdUnchoke: - ps.handleUnchoke() - break - case message.IdInterested: - break - case message.IdNotinterested: - break - case message.IdHave: - break - case message.IdBitfield: - ps.handleBitfield(buf) - break - case message.IdRequest: - break - case message.IdPiece: - ps.handlePiece(buf) - break - case message.IdCancel: - break - } - } - } - return nil -} - -func downloadPiece() { - -} - -func (ps *peerState) handleUnchoke() { - ps.peerChoking = false - // 如果客户端对peer感兴趣并且peer没有choked客户端，就可以开始下载了 - if ps.amInterested { - have := ps.getHavePieces(ps.bitfield) - if len(have) > 0 { - // 获取分片的长度 - for _, index := range have { - pieceLength := ps.torrent.MetaInfo.GetPieceSize(index) - lastOffset := uint64(0) - // 按块下载分片 - for lastOffset < pieceLength { - blockLength := math.Min(2<<13, float64(pieceLength-lastOffset)) - ps.write(message.NewRequest(uint32(index), uint32(lastOffset), uint32(blockLength)).Encode()) - lastOffset += uint64(blockLength) - } - } - return - } - } - ps.conn.Close() -} - -func (ps *peerState) handleBitfield(buf []byte) { - ps.bitfield = &message.Bitfield{} - ps.bitfield.Decode(buf) - have := ps.getHavePieces(ps.bitfield) - if len(have) > 0 { - // 表示对该peer感兴趣，并且不choked该peer - ps.write(message.NewInterested().Encode()) - ps.amInterested = true - - ps.write(message.NewUnchoke().Encode()) - ps.amChoking = false - } else { - ps.conn.Close() - } -} - -// 处理分片下载响应 -func (ps *peerState) handlePiece(buf []byte) { - piece := &message.Piece{} - piece.Decode(buf) - info := ps.torrent.MetaInfo.Info - blockLength := int64(len(piece.Block)) - pieceBegin := int64(piece.Index)*int64(info.PieceLength) + int64(piece.Begin) - // 获取对应要的文件 - var fileBlocks []fileBlock - if len(info.Files) == 0 { - // 单文件 - fileBlocks = append(fileBlocks, fileBlock{ - filepath: info.Name, - fileSeek: pieceBegin, - blockBegin: 0, - blockEnd: blockLength, - }) - } else { - // 获取要写入的第一个文件和偏移 - writeIndex := getWriteFile(pieceBegin, ps.torrent.MetaInfo.GetFileDetails()) - // block对应种子的偏移 - blockBegin := pieceBegin - // block对应文件的偏移 - var blockSeek int64 = 0 - for _, f := range ps.torrent.MetaInfo.GetFileDetails()[writeIndex:] { - // 计算文件可剩余写入字节数 - fileWritable := f.End - blockBegin - // 计算block剩余写入字节数 - blockWritable := blockLength - blockSeek - fb := fileBlock{ - filepath: f.Path[0], - fileSeek: blockBegin - f.Begin, - blockBegin: blockSeek, - } - if fileWritable >= blockWritable { - // 若够block写入直接跳出循环 - fb.blockEnd = blockSeek + blockWritable - fileBlocks = append(fileBlocks, fb) - break - } else { - // 否则计算剩余可写字节数，写入到下一个文件中 - fb.blockEnd = blockSeek + fileWritable - fileBlocks = append(fileBlocks, fb) - blockBegin = f.End - blockSeek += fileWritable - } - } - } - fmt.Println(fileBlocks) - for _, f := range fileBlocks { - func() { - name := path.Join("e:/testbt/download", f.filepath) - file, err := os.OpenFile(name, os.O_RDWR|os.O_CREATE, 0644) - if err != nil { - panic(err) - } - defer file.Close() - _, err = file.WriteAt(piece.Block[f.blockBegin:f.blockEnd], f.fileSeek) - if err != nil { - fmt.Println(err) - } - }() - } -} - -// 获取piece对应要写入的文件 -func getWriteFile(pieceBegin int64, fds []metainfo.FileDetail) int { - for i, f := range fds { - if f.Begin <= pieceBegin && f.End > pieceBegin { - return i - } - } - return -1 -} - -type fileBlock struct { - filepath string - fileSeek int64 - blockBegin int64 - blockEnd int64 -} - -// 获取peer能提供需要下载的文件分片 -func (ps *peerState) getHavePieces(bitfield *message.Bitfield) []int { - states := make([]bool, len(ps.torrent.PieceState)) - for i := range states { - states[i] = ps.torrent.PieceState[i].complete - } - return bitfield.Have(states) -} - -// 异步写入数据 -func (ps *peerState) write(buf []byte) { - ps.writeMsgCh <- buf -} - -// 获取要写入到的文件 -/*func (ps *peerState) getWriteFile(request *message.Request) string { - info := ps.torrent.MetaInfo.Info - // 单文件 - if len(info.Files) == 0 { - return info.Name - } else { - request.Index * info.PieceLength + be - for i := 0; i < len(info.Files); i++ { - - } - } -} -*/ diff --git a/down/bt/torrent/peer_state_test.go b/down/bt/torrent/peer_state_test.go deleted file mode 100644 index 62c4a2f0..00000000 --- a/down/bt/torrent/peer_state_test.go +++ /dev/null @@ -1,85 +0,0 @@ -package torrent - -import ( - "fmt" - "github.com/monkeyWie/gopeed/down/bt/metainfo" - "github.com/monkeyWie/gopeed/down/bt/peer" - "github.com/monkeyWie/gopeed/down/bt/tracker" - "testing" -) - -func Test_peerState_handshake(t *testing.T) { - - ps := getUsablePeer() - - // 发出握手请求 - handshake, err := ps.handshake() - fmt.Println("Handshake end") - if err != nil { - fmt.Println(ps.Address()) - panic(err) - } - fmt.Println(handshake) -} - -func Test_peerState_download(t *testing.T) { - - ps := getUsablePeer() - - // 发出握手请求 - _, err := ps.handshake() - if err != nil { - panic(err) - } - fmt.Println("handshake end") - - // 开始下载 - err = ps.download() - -} - -func buildTorrent() *Torrent { - metaInfo, err := metainfo.ParseFromFile("../testdata/Game.of.Thrones.S08E05.720p.WEB.H264-MEMENTO.torrent") - if err != nil { - panic(err) - } - metaInfo.Announce = "udp://exodus.desync.com:6969/announce" - metaInfo.AnnounceList = [][]string{} - return NewTorrent(peer.GenPeerID(), metaInfo) -} - -// 获取一个能连接上的peer -func getUsablePeer() *peerState { - torrent := buildTorrent() - tracker := &tracker.Tracker{ - PeerID: torrent.PeerID, - MetaInfo: torrent.MetaInfo, - } - peers, err := tracker.Tracker() - if err != nil { - panic(err) - } - fmt.Println("Tracker end") - ch := make(chan *peerState, 1) - for i := range peers { - go peerTest(torrent, &peers[i], ch) - } - return <-ch -} - -func peerTest(torrent *Torrent, peer *peer.Peer, ch chan *peerState) { - ps := &peerState{ - torrent: torrent, - Peer: peer, - } - // 连接至peer - err := ps.dialMse() - if err != nil { - return - } - ch <- ps -} - -func TestSome(t *testing.T) { - fmt.Println(11) -} diff --git a/down/bt/torrent/piece_state.go b/down/bt/torrent/piece_state.go index 8967138e..34d245c8 100644 --- a/down/bt/torrent/piece_state.go +++ b/down/bt/torrent/piece_state.go @@ -1,5 +1,56 @@ package torrent -type pieceState struct { - complete bool +import "sync" + +type state int + +const ( + stateReady = iota + stateDownloading + stateFinished +) + +// 所有piece的下载状态 +type piecesState struct { + rw *sync.RWMutex + states []state +} + +func NewPiecesState(size int) *piecesState { + ps := &piecesState{ + rw: &sync.RWMutex{}, + states: make([]state, size), + } + for i := 0; i < size; i++ { + ps.states[i] = stateReady + } + return ps +} + +func (ps *piecesState) getState(index int) state { + ps.rw.RLock() + defer ps.rw.RUnlock() + return ps.states[index] +} + +func (ps *piecesState) setState(index int, state state) { + ps.rw.Lock() + defer ps.rw.Unlock() + ps.states[index] = state +} + +func (ps *piecesState) getReadyAndDownload() int { + ps.rw.Lock() + defer ps.rw.Unlock() + for i, s := range ps.states { + if s == stateReady { + ps.states[i] = stateDownloading + return i + } + } + return -1 +} + +func (ps *piecesState) Size() int { + return len(ps.states) } diff --git a/down/bt/torrent/torrent.go b/down/bt/torrent/torrent.go index 3ee15edf..ae31d32c 100644 --- a/down/bt/torrent/torrent.go +++ b/down/bt/torrent/torrent.go @@ -1,22 +1,100 @@ package torrent import ( + "fmt" "github.com/monkeyWie/gopeed/down/bt/metainfo" + "github.com/monkeyWie/gopeed/down/bt/tracker" + "sync" + "time" ) type Torrent struct { - PeerID [20]byte - MetaInfo *metainfo.MetaInfo - PieceState []pieceState + PeerID [20]byte + MetaInfo *metainfo.MetaInfo - Peers []peerState + PiecesState *piecesState + peerPool *peerPool + + Path string } func NewTorrent(peerID [20]byte, metaInfo *metainfo.MetaInfo) *Torrent { torrent := &Torrent{ - PeerID: peerID, - MetaInfo: metaInfo, - PieceState: make([]pieceState, len(metaInfo.Info.Pieces)), + PeerID: peerID, + MetaInfo: metaInfo, + PiecesState: NewPiecesState(len(metaInfo.Info.Pieces)), } return torrent } + +func (t *Torrent) Download(path string) { + t.Path = path + t.peerPool = newPeerPool() + t.fetchPeers() + + taskCh := make(chan interface{}, 16) + defer close(taskCh) + for { + index := t.PiecesState.getReadyAndDownload() + if index == -1 { + return + } + taskCh <- nil + go func(index int) { + peer := t.peerPool.get() + if peer == nil { + fmt.Println("no peer") + time.Sleep(time.Second * 10) + t.PiecesState.setState(index, stateReady) + <-taskCh + return + } + pc := NewPeerConn(t, peer) + err := pc.ready() + if err != nil { + t.PiecesState.setState(index, stateReady) + t.peerPool.remove(peer) + } else { + fmt.Printf("peer ready:%s,download %d\n", peer.Address(), index) + err = pc.downloadPiece(index) + if err != nil { + t.PiecesState.setState(index, stateDownloading) + } + t.peerPool.release(peer) + } + <-taskCh + }(index) + } +} + +// 获取可用的peer +func (t *Torrent) fetchPeers() { + tracker := &tracker.Tracker{ + PeerID: t.PeerID, + MetaInfo: t.MetaInfo, + } + urls := []string{ + "udp://exodus.desync.com:6969/announce", + "udp://tracker.openbittorrent.com:80/announce", + "udp://tracker.leechers-paradise.org:6969", + } + var wg sync.WaitGroup + ch := make(chan interface{}) + wg.Add(len(urls)) + for _, url := range urls { + go func(url string) { + defer wg.Done() + peers, err := tracker.DoTracker(url) + if err == nil { + fmt.Println("tracker end:" + url) + ch <- nil + t.peerPool.put(peers) + } + }(url) + } + go func() { + wg.Wait() + close(ch) + }() + <-ch +} diff --git a/down/bt/torrent/torrent_test.go b/down/bt/torrent/torrent_test.go new file mode 100644 index 00000000..409249f1 --- /dev/null +++ b/down/bt/torrent/torrent_test.go @@ -0,0 +1,32 @@ +package torrent + +import ( + "github.com/monkeyWie/gopeed/down/bt/metainfo" + "github.com/monkeyWie/gopeed/down/bt/peer" + "testing" +) + +func TestTorrent_Download(t *testing.T) { + torrent := buildTorrent() + torrent.Download("e:/testbt/download") +} + +func buildTorrent() *Torrent { + metaInfo, err := metainfo.ParseFromFile("../testdata/Game.of.Thrones.S08E05.720p.WEB.H264-MEMENTO.torrent") + if err != nil { + panic(err) + } + // metaInfo.Announce = "udp://exodus.desync.com:6969/announce" + // metaInfo.Announce = "udp://tracker.openbittorrent.com:80/announce" + // metaInfo.Announce = "udp://tracker.cyberia.is:6969/announce" + // metaInfo.Announce = "udp://9.rarbg.me:2780/announce" + // metaInfo.Announce = "udp://9.rarbg.to:2760/announce" + // metaInfo.Announce = "udp://tracker.leechers-paradise.org:6969" + metaInfo.AnnounceList = [][]string{} + metaInfo.AnnounceList = append(metaInfo.AnnounceList, []string{ + "udp://exodus.desync.com:6969/announce", + "udp://tracker.openbittorrent.com:80/announce", + "udp://tracker.leechers-paradise.org:6969", + }) + return NewTorrent(peer.GenPeerID(), metaInfo) +} diff --git a/down/bt/tracker/tracker.go b/down/bt/tracker/tracker.go index 4d61016d..f652ba35 100644 --- a/down/bt/tracker/tracker.go +++ b/down/bt/tracker/tracker.go @@ -246,7 +246,7 @@ func (tracker *Tracker) Tracker() (peers []peer.Peer, err error) { for _, announceArr := range metaInfo.AnnounceList { if len(announceArr) > 0 { for _, announce := range announceArr { - peers, err = tracker.doTracker(announce) + peers, err = tracker.DoTracker(announce) if err == nil { return } @@ -254,18 +254,16 @@ func (tracker *Tracker) Tracker() (peers []peer.Peer, err error) { } } } else { - peers, err = tracker.doTracker(metaInfo.Announce) + peers, err = tracker.DoTracker(metaInfo.Announce) } return } -func (tracker *Tracker) doTracker(announce string) (peers []peer.Peer, err error) { +func (tracker *Tracker) DoTracker(announce string) (peers []peer.Peer, err error) { if announce != "" { url, _ := url.Parse(announce) switch url.Scheme { case "http", "https": - // TODO UDP test - break return tracker.httpTracker(url) case "udp", "udp4", "udp6": return tracker.udpTracker(url)