From: Matt Joiner <anacrolix@gmail.com>
Date: Thu, 3 Apr 2014 12:16:59 +0000 (+1100)
Subject: Break up client.go into several files and a few fixes
X-Git-Tag: v1.0.0~1761
X-Git-Url: http://www.git.stargrave.org/?a=commitdiff_plain;h=9ba3af19ba811a8513c235712a3dc8d7b990099a;p=btrtrc.git
Break up client.go into several files and a few fixes
Suppress expected errors when initiating connections.
Add an explicit timeout to initiate connections.
Put a lock on closing a connection to prevent data race warnings.
---
diff --git a/client.go b/client.go
index a0e375c6..fac59b75 100644
--- a/client.go
+++ b/client.go
@@ -3,7 +3,6 @@ package torrent
import (
"bufio"
"container/list"
- "crypto"
"crypto/rand"
"encoding"
"errors"
@@ -13,9 +12,8 @@ import (
mathRand "math/rand"
"net"
"os"
- "path/filepath"
- "sort"
"sync"
+ "syscall"
"time"
metainfo "github.com/nsf/libtorgo/torrent"
@@ -23,290 +21,8 @@ import (
"bitbucket.org/anacrolix/go.torrent/peer_protocol"
"bitbucket.org/anacrolix/go.torrent/tracker"
_ "bitbucket.org/anacrolix/go.torrent/tracker/udp"
- "launchpad.net/gommap"
)
-const (
- PieceHash = crypto.SHA1
- maxRequests = 250
- chunkSize = 0x4000 // 16KiB
- BEP20 = "-GT0000-"
-)
-
-type InfoHash [20]byte
-
-type pieceSum [20]byte
-
-func copyHashSum(dst, src []byte) {
- if len(dst) != len(src) || copy(dst, src) != len(dst) {
- panic("hash sum sizes differ")
- }
-}
-
-func BytesInfoHash(b []byte) (ih InfoHash) {
- if len(b) != len(ih) || copy(ih[:], b) != len(ih) {
- panic("bad infohash bytes")
- }
- return
-}
-
-type piece struct {
- Hash pieceSum
- PendingChunkSpecs map[ChunkSpec]struct{}
- Hashing bool
- QueuedForHash bool
- EverHashed bool
-}
-
-func (p *piece) Complete() bool {
- return len(p.PendingChunkSpecs) == 0 && p.EverHashed
-}
-
-func lastChunkSpec(pieceLength peer_protocol.Integer) (cs ChunkSpec) {
- cs.Begin = (pieceLength - 1) / chunkSize * chunkSize
- cs.Length = pieceLength - cs.Begin
- return
-}
-
-func (t *Torrent) PieceNumPendingBytes(index peer_protocol.Integer) (count peer_protocol.Integer) {
- pendingChunks := t.Pieces[index].PendingChunkSpecs
- count = peer_protocol.Integer(len(pendingChunks)) * chunkSize
- _lastChunkSpec := lastChunkSpec(t.PieceLength(index))
- if _lastChunkSpec.Length != chunkSize {
- if _, ok := pendingChunks[_lastChunkSpec]; ok {
- count += _lastChunkSpec.Length - chunkSize
- }
- }
- return
-}
-
-type ChunkSpec struct {
- Begin, Length peer_protocol.Integer
-}
-
-type Request struct {
- Index peer_protocol.Integer
- ChunkSpec
-}
-
-type Connection struct {
- Socket net.Conn
- Closed bool
- post chan encoding.BinaryMarshaler
- write chan []byte
-
- // Stuff controlled by the local peer.
- Interested bool
- Choked bool
- Requests map[Request]struct{}
-
- // Stuff controlled by the remote peer.
- PeerId [20]byte
- PeerInterested bool
- PeerChoked bool
- PeerRequests map[Request]struct{}
- PeerExtensions [8]byte
- PeerPieces []bool
-}
-
-func (c *Connection) Close() {
- if c.Closed {
- return
- }
- c.Socket.Close()
- close(c.post)
- c.Closed = true
-}
-
-func (c *Connection) PeerHasPiece(index peer_protocol.Integer) bool {
- if c.PeerPieces == nil {
- return false
- }
- return c.PeerPieces[index]
-}
-
-func (c *Connection) Post(msg encoding.BinaryMarshaler) {
- c.post <- msg
-}
-
-// Returns true if more requests can be sent.
-func (c *Connection) Request(chunk Request) bool {
- if len(c.Requests) >= maxRequests {
- return false
- }
- if !c.PeerPieces[chunk.Index] {
- return true
- }
- c.SetInterested(true)
- if c.PeerChoked {
- return false
- }
- if _, ok := c.Requests[chunk]; !ok {
- c.Post(peer_protocol.Message{
- Type: peer_protocol.Request,
- Index: chunk.Index,
- Begin: chunk.Begin,
- Length: chunk.Length,
- })
- }
- if c.Requests == nil {
- c.Requests = make(map[Request]struct{}, maxRequests)
- }
- c.Requests[chunk] = struct{}{}
- return true
-}
-
-func (c *Connection) Unchoke() {
- if !c.Choked {
- return
- }
- c.Post(peer_protocol.Message{
- Type: peer_protocol.Unchoke,
- })
- c.Choked = false
-}
-
-func (c *Connection) SetInterested(interested bool) {
- if c.Interested == interested {
- return
- }
- c.Post(peer_protocol.Message{
- Type: func() peer_protocol.MessageType {
- if interested {
- return peer_protocol.Interested
- } else {
- return peer_protocol.NotInterested
- }
- }(),
- })
- c.Interested = interested
-}
-
-var (
- keepAliveBytes [4]byte
-)
-
-func (conn *Connection) writer() {
- timer := time.NewTimer(0)
- defer timer.Stop()
- for {
- if !timer.Reset(time.Minute) {
- <-timer.C
- }
- var b []byte
- select {
- case <-timer.C:
- b = keepAliveBytes[:]
- case b = <-conn.write:
- if b == nil {
- return
- }
- }
- n, err := conn.Socket.Write(b)
- if err != nil {
- log.Print(err)
- break
- }
- if n != len(b) {
- panic("didn't write all bytes")
- }
- }
-}
-
-func (conn *Connection) writeOptimizer() {
- pending := list.New()
- var nextWrite []byte
- defer close(conn.write)
- for {
- write := conn.write
- if pending.Len() == 0 {
- write = nil
- } else {
- var err error
- nextWrite, err = pending.Front().Value.(encoding.BinaryMarshaler).MarshalBinary()
- if err != nil {
- panic(err)
- }
- }
- select {
- case msg, ok := <-conn.post:
- if !ok {
- return
- }
- pending.PushBack(msg)
- case write <- nextWrite:
- pending.Remove(pending.Front())
- }
- }
-}
-
-type Torrent struct {
- InfoHash InfoHash
- Pieces []*piece
- Data MMapSpan
- MetaInfo *metainfo.MetaInfo
- Conns []*Connection
- Peers []Peer
- Priorities *list.List
- // BEP 12 Multitracker Metadata Extension. The tracker.Client instances
- // mirror their respective URLs from the announce-list key.
- Trackers [][]tracker.Client
-}
-
-func (t *Torrent) NumPieces() int {
- return len(t.MetaInfo.Pieces) / PieceHash.Size()
-}
-
-func (t *Torrent) NumPiecesCompleted() (num int) {
- for _, p := range t.Pieces {
- if p.Complete() {
- num++
- }
- }
- return
-}
-
-func (t *Torrent) Length() int64 {
- return int64(t.PieceLength(peer_protocol.Integer(len(t.Pieces)-1))) + int64(len(t.Pieces)-1)*int64(t.PieceLength(0))
-}
-
-func (t *Torrent) Close() (err error) {
- t.Data.Close()
- for _, conn := range t.Conns {
- conn.Close()
- }
- return
-}
-
-type pieceByBytesPendingSlice struct {
- Pending, Indices []peer_protocol.Integer
-}
-
-func (pcs pieceByBytesPendingSlice) Len() int {
- return len(pcs.Indices)
-}
-
-func (me pieceByBytesPendingSlice) Less(i, j int) bool {
- return me.Pending[me.Indices[i]] < me.Pending[me.Indices[j]]
-}
-
-func (me pieceByBytesPendingSlice) Swap(i, j int) {
- me.Indices[i], me.Indices[j] = me.Indices[j], me.Indices[i]
-}
-
-func (t *Torrent) piecesByPendingBytesDesc() (indices []peer_protocol.Integer) {
- slice := pieceByBytesPendingSlice{
- Pending: make([]peer_protocol.Integer, 0, len(t.Pieces)),
- Indices: make([]peer_protocol.Integer, 0, len(t.Pieces)),
- }
- for i := range t.Pieces {
- slice.Pending = append(slice.Pending, t.PieceNumPendingBytes(peer_protocol.Integer(i)))
- slice.Indices = append(slice.Indices, peer_protocol.Integer(i))
- }
- sort.Sort(sort.Reverse(slice))
- return slice.Indices
-}
-
// Currently doesn't really queue, but should in the future.
func (cl *Client) queuePieceCheck(t *Torrent, pieceIndex peer_protocol.Integer) {
piece := t.Pieces[pieceIndex]
@@ -317,25 +33,6 @@ func (cl *Client) queuePieceCheck(t *Torrent, pieceIndex peer_protocol.Integer)
go cl.verifyPiece(t, pieceIndex)
}
-func (t *Torrent) offsetRequest(off int64) (req Request, ok bool) {
- req.Index = peer_protocol.Integer(off / t.MetaInfo.PieceLength)
- if req.Index < 0 || int(req.Index) >= len(t.Pieces) {
- return
- }
- off %= t.MetaInfo.PieceLength
- pieceLeft := t.PieceLength(req.Index) - peer_protocol.Integer(off)
- if pieceLeft <= 0 {
- return
- }
- req.Begin = chunkSize * (peer_protocol.Integer(off) / chunkSize)
- req.Length = chunkSize
- if req.Length > pieceLeft {
- req.Length = pieceLeft
- }
- ok = true
- return
-}
-
func (cl *Client) PrioritizeDataRegion(ih InfoHash, off, len_ int64) {
cl.mu.Lock()
defer cl.mu.Unlock()
@@ -369,79 +66,6 @@ func (cl *Client) PrioritizeDataRegion(ih InfoHash, off, len_ int64) {
}
}
-func (t *Torrent) WriteChunk(piece int, begin int64, data []byte) (err error) {
- _, err = t.Data.WriteAt(data, int64(piece)*t.MetaInfo.PieceLength+begin)
- return
-}
-
-func (t *Torrent) bitfield() (bf []bool) {
- for _, p := range t.Pieces {
- bf = append(bf, p.EverHashed && len(p.PendingChunkSpecs) == 0)
- }
- return
-}
-
-func (t *Torrent) pendAllChunkSpecs(index peer_protocol.Integer) {
- piece := t.Pieces[index]
- if piece.PendingChunkSpecs == nil {
- piece.PendingChunkSpecs = make(
- map[ChunkSpec]struct{},
- (t.MetaInfo.PieceLength+chunkSize-1)/chunkSize)
- }
- c := ChunkSpec{
- Begin: 0,
- }
- cs := piece.PendingChunkSpecs
- for left := peer_protocol.Integer(t.PieceLength(index)); left != 0; left -= c.Length {
- c.Length = left
- if c.Length > chunkSize {
- c.Length = chunkSize
- }
- cs[c] = struct{}{}
- c.Begin += c.Length
- }
- return
-}
-
-func (t *Torrent) requestHeat() (ret map[Request]int) {
- ret = make(map[Request]int)
- for _, conn := range t.Conns {
- for req, _ := range conn.Requests {
- ret[req]++
- }
- }
- return
-}
-
-type Peer struct {
- Id [20]byte
- IP net.IP
- Port int
-}
-
-func (t *Torrent) PieceLength(piece peer_protocol.Integer) (len_ peer_protocol.Integer) {
- if int(piece) == t.NumPieces()-1 {
- len_ = peer_protocol.Integer(t.Data.Size() % t.MetaInfo.PieceLength)
- }
- if len_ == 0 {
- len_ = peer_protocol.Integer(t.MetaInfo.PieceLength)
- }
- return
-}
-
-func (t *Torrent) HashPiece(piece peer_protocol.Integer) (ps pieceSum) {
- hash := PieceHash.New()
- n, err := t.Data.WriteSectionTo(hash, int64(piece)*t.MetaInfo.PieceLength, t.MetaInfo.PieceLength)
- if err != nil {
- panic(err)
- }
- if peer_protocol.Integer(n) != t.PieceLength(piece) {
- panic(fmt.Sprintf("hashed wrong number of bytes: expected %d; did %d; piece %d", t.PieceLength(piece), n, piece))
- }
- copyHashSum(ps[:], hash.Sum(nil))
- return
-}
-
type DataSpec struct {
InfoHash
Request
@@ -464,10 +88,6 @@ type Client struct {
dataWaiter chan struct{}
}
-var (
- ErrDataNotReady = errors.New("data not ready")
-)
-
func (cl *Client) TorrentReadAt(ih InfoHash, off int64, p []byte) (n int, err error) {
cl.mu.Lock()
defer cl.mu.Unlock()
@@ -576,54 +196,6 @@ func (cl *Client) acceptConnections() {
}
}
-func mmapTorrentData(metaInfo *metainfo.MetaInfo, location string) (mms MMapSpan, err error) {
- defer func() {
- if err != nil {
- mms.Close()
- mms = nil
- }
- }()
- for _, miFile := range metaInfo.Files {
- fileName := filepath.Join(append([]string{location, metaInfo.Name}, miFile.Path...)...)
- err = os.MkdirAll(filepath.Dir(fileName), 0777)
- if err != nil {
- return
- }
- var file *os.File
- file, err = os.OpenFile(fileName, os.O_CREATE|os.O_RDWR, 0666)
- if err != nil {
- return
- }
- func() {
- defer file.Close()
- var fi os.FileInfo
- fi, err = file.Stat()
- if err != nil {
- return
- }
- if fi.Size() < miFile.Length {
- err = file.Truncate(miFile.Length)
- if err != nil {
- return
- }
- }
- var mMap gommap.MMap
- mMap, err = gommap.MapRegion(file.Fd(), 0, miFile.Length, gommap.PROT_READ|gommap.PROT_WRITE, gommap.MAP_SHARED)
- if err != nil {
- return
- }
- if int64(len(mMap)) != miFile.Length {
- panic("mmap has wrong length")
- }
- mms = append(mms, MMap{mMap})
- }()
- if err != nil {
- return
- }
- }
- return
-}
-
func (me *Client) torrent(ih InfoHash) *Torrent {
for _, t := range me.torrents {
if t.InfoHash == ih {
@@ -639,18 +211,33 @@ func (me *Client) initiateConn(peer Peer, torrent *Torrent) {
}
me.halfOpen++
go func() {
- conn, err := net.DialTCP("tcp", nil, &net.TCPAddr{
+ addr := &net.TCPAddr{
IP: peer.IP,
Port: peer.Port,
- })
+ }
+ conn, err := net.DialTimeout(addr.Network(), addr.String(), dialTimeout)
- me.mu.Lock()
- me.halfOpen--
- me.openNewConns()
- me.mu.Unlock()
+ go func() {
+ me.mu.Lock()
+ defer me.mu.Unlock()
+ if me.halfOpen == 0 {
+ panic("assert")
+ }
+ me.halfOpen--
+ me.openNewConns()
+ }()
+ if netOpErr, ok := err.(*net.OpError); ok {
+ if netOpErr.Timeout() {
+ return
+ }
+ switch netOpErr.Err {
+ case syscall.ECONNREFUSED:
+ return
+ }
+ }
if err != nil {
- log.Printf("error connecting to peer: %s", err)
+ log.Printf("error connecting to peer: %s %#v", err, err)
return
}
log.Printf("connected to %s", conn.RemoteAddr())
@@ -661,24 +248,6 @@ func (me *Client) initiateConn(peer Peer, torrent *Torrent) {
}()
}
-func (t *Torrent) haveAllPieces() bool {
- for _, piece := range t.Pieces {
- if !piece.Complete() {
- return false
- }
- }
- return true
-}
-
-func (me *Torrent) haveAnyPieces() bool {
- for _, piece := range me.Pieces {
- if piece.Complete() {
- return true
- }
- }
- return false
-}
-
func (me *Client) runConnection(sock net.Conn, torrent *Torrent) (err error) {
conn := &Connection{
Socket: sock,
@@ -765,11 +334,6 @@ func (me *Client) peerGotPiece(torrent *Torrent, conn *Connection, piece int) {
}
}
-func (t *Torrent) wantPiece(index int) bool {
- p := t.Pieces[index]
- return p.EverHashed && len(p.PendingChunkSpecs) != 0
-}
-
func (me *Client) peerUnchoked(torrent *Torrent, conn *Connection) {
me.replenishConnRequests(torrent, conn)
}
@@ -851,7 +415,7 @@ func (me *Client) connectionLoop(torrent *Torrent, conn *Connection) error {
case peer_protocol.Piece:
request_ := Request{msg.Index, ChunkSpec{msg.Begin, peer_protocol.Integer(len(msg.Piece))}}
if _, ok := conn.Requests[request_]; !ok {
- err = errors.New("unexpected piece")
+ err = fmt.Errorf("unexpected piece: %s", request_)
break
}
delete(conn.Requests, request_)
@@ -1086,7 +650,7 @@ func (me *Client) replenishConnRequests(torrent *Torrent, conn *Connection) {
}
}
}
- // Then finish of incomplete pieces in order of bytes remaining.
+ // Then finish off incomplete pieces in order of bytes remaining.
for _, index := range torrent.piecesByPendingBytesDesc() {
if torrent.PieceNumPendingBytes(index) == torrent.PieceLength(index) {
continue
diff --git a/connection.go b/connection.go
new file mode 100644
index 00000000..cbe0e91c
--- /dev/null
+++ b/connection.go
@@ -0,0 +1,174 @@
+package torrent
+
+import (
+ "container/list"
+ "encoding"
+ "log"
+ "net"
+ "sync"
+ "time"
+
+ "bitbucket.org/anacrolix/go.torrent/peer_protocol"
+)
+
+// Maintains the state of a connection with a peer.
+type Connection struct {
+ Socket net.Conn
+ closed bool
+ mu sync.Mutex // Only for closing.
+ post chan encoding.BinaryMarshaler
+ write chan []byte
+
+ // Stuff controlled by the local peer.
+ Interested bool
+ Choked bool
+ Requests map[Request]struct{}
+
+ // Stuff controlled by the remote peer.
+ PeerId [20]byte
+ PeerInterested bool
+ PeerChoked bool
+ PeerRequests map[Request]struct{}
+ PeerExtensions [8]byte
+ PeerPieces []bool
+}
+
+func (c *Connection) Close() {
+ c.mu.Lock()
+ if c.closed {
+ return
+ }
+ c.Socket.Close()
+ close(c.post)
+ c.closed = true
+ c.mu.Unlock()
+}
+
+func (c *Connection) getClosed() bool {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return c.closed
+}
+
+func (c *Connection) PeerHasPiece(index peer_protocol.Integer) bool {
+ if c.PeerPieces == nil {
+ return false
+ }
+ return c.PeerPieces[index]
+}
+
+func (c *Connection) Post(msg encoding.BinaryMarshaler) {
+ c.post <- msg
+}
+
+// Returns true if more requests can be sent.
+func (c *Connection) Request(chunk Request) bool {
+ if len(c.Requests) >= maxRequests {
+ return false
+ }
+ if !c.PeerPieces[chunk.Index] {
+ return true
+ }
+ c.SetInterested(true)
+ if c.PeerChoked {
+ return false
+ }
+ if _, ok := c.Requests[chunk]; !ok {
+ c.Post(peer_protocol.Message{
+ Type: peer_protocol.Request,
+ Index: chunk.Index,
+ Begin: chunk.Begin,
+ Length: chunk.Length,
+ })
+ }
+ if c.Requests == nil {
+ c.Requests = make(map[Request]struct{}, maxRequests)
+ }
+ c.Requests[chunk] = struct{}{}
+ return true
+}
+
+func (c *Connection) Unchoke() {
+ if !c.Choked {
+ return
+ }
+ c.Post(peer_protocol.Message{
+ Type: peer_protocol.Unchoke,
+ })
+ c.Choked = false
+}
+
+func (c *Connection) SetInterested(interested bool) {
+ if c.Interested == interested {
+ return
+ }
+ c.Post(peer_protocol.Message{
+ Type: func() peer_protocol.MessageType {
+ if interested {
+ return peer_protocol.Interested
+ } else {
+ return peer_protocol.NotInterested
+ }
+ }(),
+ })
+ c.Interested = interested
+}
+
+var (
+ // Four consecutive zero bytes that comprise a keep alive on the wire.
+ keepAliveBytes [4]byte
+)
+
+func (conn *Connection) writer() {
+ timer := time.NewTimer(0)
+ defer timer.Stop()
+ for {
+ if !timer.Reset(time.Minute) {
+ <-timer.C
+ }
+ var b []byte
+ select {
+ case <-timer.C:
+ b = keepAliveBytes[:]
+ case b = <-conn.write:
+ if b == nil {
+ return
+ }
+ }
+ _, err := conn.Socket.Write(b)
+ if conn.getClosed() {
+ break
+ }
+ if err != nil {
+ log.Print(err)
+ break
+ }
+ }
+}
+
+func (conn *Connection) writeOptimizer() {
+ pending := list.New()
+ var nextWrite []byte
+ defer close(conn.write)
+ for {
+ write := conn.write
+ if pending.Len() == 0 {
+ write = nil
+ } else {
+ var err error
+ nextWrite, err = pending.Front().Value.(encoding.BinaryMarshaler).MarshalBinary()
+ if err != nil {
+ panic(err)
+ }
+ }
+ select {
+ case msg, ok := <-conn.post:
+ if !ok {
+ return
+ }
+ pending.PushBack(msg)
+ case write <- nextWrite:
+ pending.Remove(pending.Front())
+ }
+ }
+}
diff --git a/misc.go b/misc.go
new file mode 100644
index 00000000..e5644005
--- /dev/null
+++ b/misc.go
@@ -0,0 +1,133 @@
+package torrent
+
+import (
+ "crypto"
+ "errors"
+ "os"
+ "path/filepath"
+ "time"
+
+ "bitbucket.org/anacrolix/go.torrent/peer_protocol"
+ metainfo "github.com/nsf/libtorgo/torrent"
+ "launchpad.net/gommap"
+)
+
+const (
+ PieceHash = crypto.SHA1
+ maxRequests = 250
+ chunkSize = 0x4000 // 16KiB
+ BEP20 = "-GT0000-"
+ dialTimeout = time.Second * 15
+)
+
+type InfoHash [20]byte
+
+type pieceSum [20]byte
+
+func copyHashSum(dst, src []byte) {
+ if len(dst) != len(src) || copy(dst, src) != len(dst) {
+ panic("hash sum sizes differ")
+ }
+}
+
+func BytesInfoHash(b []byte) (ih InfoHash) {
+ if len(b) != len(ih) || copy(ih[:], b) != len(ih) {
+ panic("bad infohash bytes")
+ }
+ return
+}
+
+type piece struct {
+ Hash pieceSum
+ PendingChunkSpecs map[ChunkSpec]struct{}
+ Hashing bool
+ QueuedForHash bool
+ EverHashed bool
+}
+
+func (p *piece) Complete() bool {
+ return len(p.PendingChunkSpecs) == 0 && p.EverHashed
+}
+
+func lastChunkSpec(pieceLength peer_protocol.Integer) (cs ChunkSpec) {
+ cs.Begin = (pieceLength - 1) / chunkSize * chunkSize
+ cs.Length = pieceLength - cs.Begin
+ return
+}
+
+type ChunkSpec struct {
+ Begin, Length peer_protocol.Integer
+}
+
+type Request struct {
+ Index peer_protocol.Integer
+ ChunkSpec
+}
+
+type pieceByBytesPendingSlice struct {
+ Pending, Indices []peer_protocol.Integer
+}
+
+func (pcs pieceByBytesPendingSlice) Len() int {
+ return len(pcs.Indices)
+}
+
+func (me pieceByBytesPendingSlice) Less(i, j int) bool {
+ return me.Pending[me.Indices[i]] < me.Pending[me.Indices[j]]
+}
+
+func (me pieceByBytesPendingSlice) Swap(i, j int) {
+ me.Indices[i], me.Indices[j] = me.Indices[j], me.Indices[i]
+}
+
+var (
+ ErrDataNotReady = errors.New("data not ready")
+)
+
+func mmapTorrentData(metaInfo *metainfo.MetaInfo, location string) (mms MMapSpan, err error) {
+ defer func() {
+ if err != nil {
+ mms.Close()
+ mms = nil
+ }
+ }()
+ for _, miFile := range metaInfo.Files {
+ fileName := filepath.Join(append([]string{location, metaInfo.Name}, miFile.Path...)...)
+ err = os.MkdirAll(filepath.Dir(fileName), 0777)
+ if err != nil {
+ return
+ }
+ var file *os.File
+ file, err = os.OpenFile(fileName, os.O_CREATE|os.O_RDWR, 0666)
+ if err != nil {
+ return
+ }
+ func() {
+ defer file.Close()
+ var fi os.FileInfo
+ fi, err = file.Stat()
+ if err != nil {
+ return
+ }
+ if fi.Size() < miFile.Length {
+ err = file.Truncate(miFile.Length)
+ if err != nil {
+ return
+ }
+ }
+ var mMap gommap.MMap
+ mMap, err = gommap.MapRegion(file.Fd(), 0, miFile.Length, gommap.PROT_READ|gommap.PROT_WRITE, gommap.MAP_SHARED)
+ if err != nil {
+ return
+ }
+ if int64(len(mMap)) != miFile.Length {
+ panic("mmap has wrong length")
+ }
+ mms = append(mms, MMap{mMap})
+ }()
+ if err != nil {
+ return
+ }
+ }
+ return
+}
diff --git a/torrent.go b/torrent.go
new file mode 100644
index 00000000..19c64d2f
--- /dev/null
+++ b/torrent.go
@@ -0,0 +1,189 @@
+package torrent
+
+import (
+ "container/list"
+ "fmt"
+ "net"
+ "sort"
+
+ "bitbucket.org/anacrolix/go.torrent/peer_protocol"
+ "bitbucket.org/anacrolix/go.torrent/tracker"
+ metainfo "github.com/nsf/libtorgo/torrent"
+)
+
+func (t *Torrent) PieceNumPendingBytes(index peer_protocol.Integer) (count peer_protocol.Integer) {
+ pendingChunks := t.Pieces[index].PendingChunkSpecs
+ count = peer_protocol.Integer(len(pendingChunks)) * chunkSize
+ _lastChunkSpec := lastChunkSpec(t.PieceLength(index))
+ if _lastChunkSpec.Length != chunkSize {
+ if _, ok := pendingChunks[_lastChunkSpec]; ok {
+ count += _lastChunkSpec.Length - chunkSize
+ }
+ }
+ return
+}
+
+type Torrent struct {
+ InfoHash InfoHash
+ Pieces []*piece
+ Data MMapSpan
+ MetaInfo *metainfo.MetaInfo
+ Conns []*Connection
+ Peers []Peer
+ Priorities *list.List
+ // BEP 12 Multitracker Metadata Extension. The tracker.Client instances
+ // mirror their respective URLs from the announce-list key.
+ Trackers [][]tracker.Client
+}
+
+func (t *Torrent) NumPieces() int {
+ return len(t.MetaInfo.Pieces) / PieceHash.Size()
+}
+
+func (t *Torrent) NumPiecesCompleted() (num int) {
+ for _, p := range t.Pieces {
+ if p.Complete() {
+ num++
+ }
+ }
+ return
+}
+
+func (t *Torrent) Length() int64 {
+ return int64(t.PieceLength(peer_protocol.Integer(len(t.Pieces)-1))) + int64(len(t.Pieces)-1)*int64(t.PieceLength(0))
+}
+
+func (t *Torrent) Close() (err error) {
+ t.Data.Close()
+ for _, conn := range t.Conns {
+ conn.Close()
+ }
+ return
+}
+
+func (t *Torrent) piecesByPendingBytesDesc() (indices []peer_protocol.Integer) {
+ slice := pieceByBytesPendingSlice{
+ Pending: make([]peer_protocol.Integer, 0, len(t.Pieces)),
+ Indices: make([]peer_protocol.Integer, 0, len(t.Pieces)),
+ }
+ for i := range t.Pieces {
+ slice.Pending = append(slice.Pending, t.PieceNumPendingBytes(peer_protocol.Integer(i)))
+ slice.Indices = append(slice.Indices, peer_protocol.Integer(i))
+ }
+ sort.Sort(sort.Reverse(slice))
+ return slice.Indices
+}
+
+func (t *Torrent) offsetRequest(off int64) (req Request, ok bool) {
+ req.Index = peer_protocol.Integer(off / t.MetaInfo.PieceLength)
+ if req.Index < 0 || int(req.Index) >= len(t.Pieces) {
+ return
+ }
+ off %= t.MetaInfo.PieceLength
+ pieceLeft := t.PieceLength(req.Index) - peer_protocol.Integer(off)
+ if pieceLeft <= 0 {
+ return
+ }
+ req.Begin = chunkSize * (peer_protocol.Integer(off) / chunkSize)
+ req.Length = chunkSize
+ if req.Length > pieceLeft {
+ req.Length = pieceLeft
+ }
+ ok = true
+ return
+}
+
+func (t *Torrent) WriteChunk(piece int, begin int64, data []byte) (err error) {
+ _, err = t.Data.WriteAt(data, int64(piece)*t.MetaInfo.PieceLength+begin)
+ return
+}
+
+func (t *Torrent) bitfield() (bf []bool) {
+ for _, p := range t.Pieces {
+ bf = append(bf, p.EverHashed && len(p.PendingChunkSpecs) == 0)
+ }
+ return
+}
+
+func (t *Torrent) pendAllChunkSpecs(index peer_protocol.Integer) {
+ piece := t.Pieces[index]
+ if piece.PendingChunkSpecs == nil {
+ piece.PendingChunkSpecs = make(
+ map[ChunkSpec]struct{},
+ (t.MetaInfo.PieceLength+chunkSize-1)/chunkSize)
+ }
+ c := ChunkSpec{
+ Begin: 0,
+ }
+ cs := piece.PendingChunkSpecs
+ for left := peer_protocol.Integer(t.PieceLength(index)); left != 0; left -= c.Length {
+ c.Length = left
+ if c.Length > chunkSize {
+ c.Length = chunkSize
+ }
+ cs[c] = struct{}{}
+ c.Begin += c.Length
+ }
+ return
+}
+
+func (t *Torrent) requestHeat() (ret map[Request]int) {
+ ret = make(map[Request]int)
+ for _, conn := range t.Conns {
+ for req, _ := range conn.Requests {
+ ret[req]++
+ }
+ }
+ return
+}
+
+type Peer struct {
+ Id [20]byte
+ IP net.IP
+ Port int
+}
+
+func (t *Torrent) PieceLength(piece peer_protocol.Integer) (len_ peer_protocol.Integer) {
+ if int(piece) == t.NumPieces()-1 {
+ len_ = peer_protocol.Integer(t.Data.Size() % t.MetaInfo.PieceLength)
+ }
+ if len_ == 0 {
+ len_ = peer_protocol.Integer(t.MetaInfo.PieceLength)
+ }
+ return
+}
+
+func (t *Torrent) HashPiece(piece peer_protocol.Integer) (ps pieceSum) {
+ hash := PieceHash.New()
+ n, err := t.Data.WriteSectionTo(hash, int64(piece)*t.MetaInfo.PieceLength, t.MetaInfo.PieceLength)
+ if err != nil {
+ panic(err)
+ }
+ if peer_protocol.Integer(n) != t.PieceLength(piece) {
+ panic(fmt.Sprintf("hashed wrong number of bytes: expected %d; did %d; piece %d", t.PieceLength(piece), n, piece))
+ }
+ copyHashSum(ps[:], hash.Sum(nil))
+ return
+}
+func (t *Torrent) haveAllPieces() bool {
+ for _, piece := range t.Pieces {
+ if !piece.Complete() {
+ return false
+ }
+ }
+ return true
+}
+
+func (me *Torrent) haveAnyPieces() bool {
+ for _, piece := range me.Pieces {
+ if piece.Complete() {
+ return true
+ }
+ }
+ return false
+}
+
+func (t *Torrent) wantPiece(index int) bool {
+ p := t.Pieces[index]
+ return p.EverHashed && len(p.PendingChunkSpecs) != 0
+}